Rabbit is a set of tools for building applications with RabbitMQ.
The package can be installed by adding rabbit to your list of dependencies in mix.exs:
defdepsdo[{:rabbit,"~> 0.20"}]endPlease see HexDocs for additional documentation.
Connections form the basis of any application that is working with RabbitMQ. A connection module is needed by all the other modules included with Rabbit.
Upon start, a connection will automatically create a pool of RabbitMQ connections to utilize.
defmoduleMyConnectiondouseRabbit.Connectiondefstart_link(opts\\[])doRabbit.Connection.start_link(__MODULE__,opts,name: __MODULE__)end# Callbacks@implRabbit.Connectiondefinit(:connection_pool,opts)do# Perform runtime pool config{:ok,opts}enddefinit(:connection,opts)do# Perform runtime connection configuri=System.get_env("RABBITMQ_URI")||"amqp://guest:guest@127.0.0.1:5672"opts=Keyword.put(opts,:uri,uri){:ok,opts}endendMyConnection.start_link()Consumers are the "workers" of your application. They must be provided a connection
module and queue to consume. Every message received is then passed along to your
handle_message/1 callback within its own process.
You can optionally implement the handle_setup/2 callback to perform any work
needed to declare queues/exchanges/bindings.
defmoduleMyConsumerdouseRabbit.Consumerdefstart_link(opts\\[])doRabbit.Consumer.start_link(__MODULE__,opts,name: __MODULE__)end# Callbacks@implRabbit.Consumerdefinit(:consumer,opts)do# Perform runtime config{:ok,opts}end@implRabbit.Consumerdefhandle_setup(state)do# Optional callback to perform any exchange or queue setupAMQP.Queue.declare(state.channel,state.queue):okend@implRabbit.Consumerdefhandle_message(message)do# Handle message consumptionIO.inspect(message.payload){:ack,message}end@implRabbit.Consumerdefhandle_error(message)do# Handle message errors{:nack,message}endendMyConsumer.start_link(connection: MyConnection,queue: "my_queue",prefetch_count: 10)Consumer supervisors provide an easy way to start and supervise multiple consumer processes. Rather than creating a module for each consumer and implementing repetitive logic - the same callbacks are used across all consumers.
defmoduleMyConsumerSupervisordouseRabbit.ConsumerSupervisordefstart_link(consumers\\[])doRabbit.ConsumerSupervisor.start_link(__MODULE__,consumers,name: __MODULE__)end# Callbacks@implRabbit.ConsumerSupervisordefinit(:consumer_supervisor,_consumers)do# Perform runtime config for the consumer supervisorconsumers=[[connection: MyConnection,queue: "my_queue",prefetch_count: 5],[connection: MyConnection,queue: "my_queue_2",prefetch_count: 10],]{:ok,consumers}enddefinit(:consumer,opts)do# Perform runtime config per consumer{:ok,opts}end@implRabbit.ConsumerSupervisordefhandle_setup(state)do# Optional callback to perform any exchange or queue setup per consumerAMQP.Queue.declare(state.channel,state.queue):okend@implRabbit.ConsumerSupervisordefhandle_message(message)do# Handle message consumption per consumerIO.inspect(message.payload){:ack,message}end@implRabbit.ConsumerSupervisordefhandle_error(message)do# Handle message errors per consumer{:nack,message}endendMyConsumerSupervisor.start_link()In order to publish messages to RabbitMQ, we must create a producer module. They must be provided a connection module.
Upon start, a producer will automatically create a pool of RabbitMQ channels to publish from.
You can optionally implement the handle_setup/1 callback to perform any work
needed to declare queues/exchanges/bindings.
defmoduleMyProducerdouseRabbit.Producerdefstart_link(opts\\[])doRabbit.Producer.start_link(__MODULE__,opts,name: __MODULE__)end# Callbacks@implRabbit.Producerdefinit(:producer_pool,opts)do# Perform runtime config for the producer pool{:ok,opts}enddefinit(:producer,opts)do# Perform runtime config per producer{:ok,opts}endendMyProducer.start_link(connection: MyConnection)Rabbit.Producer.publish(MyProducer,"","my_queue","hello")Topology provides a way to centralize any RabbitMQ setup required by your application. In that sense, it should be started BEFORE any of your producers or consumers.
Using a topology, you can automatically setup queues, exchanges and bindings with simple keyword lists.
defmoduleMyTopologydouseRabbit.Topologydefstart_link(opts\\[])doRabbit.Topology.start_link(__MODULE__,opts,name: __MODULE__)end# Callbacks@implRabbit.Topologydefinit(:topology,opts)do# Perform runtime config{:ok,opts}endendMyTopology.start_link(connection: MyConnection,queues: [[name: "my_queue",durable: true],[name: "my_queue_2",durable: true],],exchanges: [[name: "my_exchange"],[name: "my_exchange_2",type: :fanout,durable: true],],bindings: [[type: :queue,source: "my_exchange",destination: "my_queue",routing_key: "my_key"],[type: :exchange,source: "my_exchange_2",destination: "my_exchange_1"]])Brokers encapsulate all of the above components into a single easy-to-use module. It provides a single place to handle your RabbitMQ connections, topology, producers and consumers.
defmoduleMyBrokerdouseRabbit.Brokerdefstart_link(opts\\[])doRabbit.Broker.start_link(__MODULE__,opts,name: __MODULE__)end# Callbacks@implRabbit.Broker# Perform runtime configuration per componentdefinit(:connection_pool,opts),do: {:ok,opts}definit(:connection,opts),do: {:ok,opts}definit(:topology,opts),do: {:ok,opts}definit(:producer_pool,opts),do: {:ok,opts}definit(:producer,opts),do: {:ok,opts}definit(:consumer_supervisor,opts),do: {:ok,opts}definit(:consumer,opts),do: {:ok,opts}@implRabbit.Brokerdefhandle_message(message)do# Handle message consumption per consumerIO.inspect(message.payload){:ack,message}end@implRabbit.Brokerdefhandle_error(message)do# Handle message errors per consumer{:nack,message}endendMyBroker.start_link(connection: [uri: "amqp://guest:guest@127.0.0.1:5672"],topology: [queues: [[name: "my_queue",durable: true],[name: "my_queue_2",durable: true]]],producer: [pool_size: 10],consumers: [[queue: "my_queue"],[queue: "my_queue_2",prefetch_count: 10]])Rabbit.Broker.publish(MyBroker,"","my_queue","hello")