Edited to handle more option in exchanges / queues. Major work done by d0rc
Easy way to get a queue/exchange worker:
import Exrabbit.DSL amqp_worker TestQ, queue: "testQ" do on json = %{} do IO.puts "JSON: #{inspect json}" end on <<"hello">> do IO.puts "Hello-hello from MQ" end on text do IO.puts "Some random binary: #{inspect text}" end endN.B. Instead of passing configuration options when defining module with amqp_worker one can add following to config.exs:
[ exrabbit: [ my_queue: [queue: "TestQ"] ] ]and then define module as:
amqp_worker TestQ, config_name: :my_queue, decode_json: [keys: :atoms] do on %{cmd: "resize_image", image: image} do IO.puts "Resizing image: #{inspect image}" end endChecking if message was published:
publish(channel, exchange, routing_key, message, :wait_confirmation)Workflow to send message:
amqp = Exrabbit.Utils.connect channel = Exrabbit.Utils.channel amqp Exrabbit.Utils.publish channel, "testExchange", "", "hello, world"To get messages, almost the same, but functions are
Exrabbit.Utils.get_messages channel, "testQueue" case Exrabbit.Utils.get_messages_ack channel, "testQueue" do nil -> IO.puts "No messages waiting" [tag: tag, content: message] -> IO.puts "Got message #{message}" Exrabbit.Utils.ack tag # acking message endPlease consult: http://www.rabbitmq.com/erlang-client-user-guide.html#returns to find out how to write gen_server consuming messages.
defmodule Consumer do use GenServer.Behaviour import Exrabbit.Utils require Lager def start_link, do: :gen_server.start_link(Consumer, [], []) def init(_opts) do amqp = connect channel = channel amqp subscribe channel, "testQ" {:ok, [connection: amqp, channel: channel]} end def handle_info(request, state) do case parse_message(request) do nil -> Lager.info "Got nil message" {tag, payload} -> Lager.info "Got message with tag #{tag} and payload #{payload}" ack state[:channel], tag end { :noreply, state} end endOr same, using behaviours:
defmodule Test do use Exrabbit.Subscriber def handle_message(msg, _state) do case parse_message(msg) do nil -> IO.puts "Nil" {tag,json} -> IO.puts "Msg: #{json}" ack _state[:channel], tag {tag,json,reply_to} -> IO.puts "For RPC messaging: #{json}" publish(_state[:channel], "", reply_to, "#{json}") # Return ECHO ack _state[:channel], tag end end end :gen_server.start Test, [queue: "testQ"], []