AMQP Adapter

RosettaQueue comes with two AMQP adapters: a synchronous adapter and an evented one. The former uses the synchronous AMQP client Bunny by celldee and the latter, the evented AMQP client by Aman Gupta.

Set Up

Install Erlang, unless already installed, and the RabbitMQ messaging broker. Mac OSX users can install from ports. Linux users should use their package manager of choice. Once installed, you can launch rabbitmq by running:

  rabbitmq-server

or

  rabbitmqctl start_app

Configuration

If you are using an AMQP adapter, there are some important rules when defining the adapter and mapping your destinations. Two of the basic building blocks of AMQP are queues and exchanges. Queues bind to exchanges in several different ways. A queue can bind to an exchange and request messages that match a specific routing key, which is called ‘direct exchange’ and is analogous to the ‘point-to-point’ messaging pattern. Alternately, queues can bind to an exchange to receive all messages sent to that exchange; this is called ‘fanout exchange’ and is analogous to the ‘publish-subscribe’ pattern. So, when you want to define a simple ‘direct-exchange’ queue, your queue name must begin with the term “queue”.

   RosettaQueue::Destinations.define do |queue|
     queue.map :test_queue, 'queue.my_test_queue'
   end

And, when you want to define a queue that will bind to a ‘fanout-exchange’, your queue name must begin with the term “fanout”.

   RosettaQueue::Destinations.define do |queue|
     queue.map :test_queue, 'fanout.my_test_queue'
   end

Finally, when defining your adapter, the synchronous AMQP adapter should be defined as :amqp_synch, and the evented adapter as :amqp_evented.

Evented AMQP adapter

When publishing and subscribing using the evented AMQP adapter, you need to wrap that code in an event machine run block. For example, publishing a message might look like this:

  EM.run do
    Em.add_periodic_timer(1) do
      RosettaQueue::Producer.publish(:foo, "Hello World!")
    end
  end

And a sample consumer might look like this:

  EM.run do
    RosettaQueue::Consumer.new(MyMessageHandler.new).receive
  end

Although, if you are using the evented AMQP adapter to consume messages you could simply wrap your consumers in RosettaQueue’s EventedManager:

  RosettaQueue::EventedManager.create do |m|
    m.add(MyFirstMessageHandler.new)
    m.add(MySecondMessageHandler.new)
    m.add(MyThirdMessageHandler.new)

    m.start
  end