There are a bunch of different ways to get connected to RabbitMQ. Thankfully the community has somewhat rallied behind the queue messaging system and there are a growing number of clients/libraries to aid in accessing the queue’s functionality.

In this post, we’ll explore the different options for connecting a ruby program to the AMQP-driven queue server. So, with no further ado:

AMQP: the grandfather

Installing

gem sources -a http://gems.github.com
sudo gem install tmm1-amqp
# OR, for the developers in the crowd
git clone git://github.com/tmm1/amqp.git
rake gem
sudo gem install amqp-*.gem

Now that we’ve got it installed, let’s write a quick example to get connected:

require "rubygems"
require 'mq'

EM.run {
  AMQP.start(:host => 'localhost') do
    amq = MQ.new
    EM.add_periodic_timer(1) { amq.queue("noises").publish("moo") }
  end
}

Exactly as it looks, we’ll be using EventMachine to provide us with a periodic timer. When the timer has been exhausted, the block will be run, in our case we’ll push the message “moo” to the queue named “noises” at the RabbitMQ host (in this example, it’s at localhost).

Now, let’s consume a message now. This is nearly straight from the examples/ directory of tmm1-amqp library:

require "rubygems"
require 'mq'

AMQP.start(:host => 'localhost') do
  MQ.queue("noises").publish("moo")
  MQ.queue("noises").publish("quack")
  MQ.queue("noises").publish("bark")
  
  i = 0
  
  MQ.queue('noises').subscribe(:ack => true) do |h,m|
    if (i+=1) == 3
      puts 'Shutting down...'
      AMQP.stop{ EM.stop }
    end

    if AMQP.closing?
      puts "#{m} (ignored, redelivered later)"
    else
      puts m
      h.ack
    end
  end
end  

# moo
# quack
# Shutting down...
# bark (ignored, redelivered later)

In this example, we’ve pushed a few sounds to the noises queue and then later subscribe to the queue (and demanding an acknowledgement receipt).

There are alternative methods to pull a message off the queue. For instance, you can use pop (if you don’t need an ack of the message):

queue.pop{ |msg|
  unless msg
    # queue was empty
    p [Time.now, :queue_empty!]

    # try again in 1 second
    EM.add_timer(1){ queue.pop }
  else
    # process this message
    p [Time.now, msg]

    # get the next message in the queue
    queue.pop
  end
}

Up until now, we’ve only used a queue type called: type. Now, there are N+1 ways to skin a cat, so let’s look at sending a queue message everyone who is subscribed to the fanout. First off, to understand how this works, let’s define a fanout. A fanout in RabbitMQ terminology is like a direct exchange, except that it represents a 1:N message delivery pattern. Now, one more, we need to define a binding as well, you’ll see why in just a minute. First, a binding, from the RabbitMQ documentation:

‘When you publish a message, you send a "routing key" along with it, that's used by the exchange when it decides which queues to forward a copy of the message on to. The links between exchanges and queues are created through binding, with a "binding pattern" that is used by the exchange when comparing against routing keys.

Now that we’ve got that under control, let’s actually write one. Again, this is nearly directly from the examples/mq/clock.rb if you’d like to look at a more complete example

require "rubygems"
require "mq"

AMQP.start(:host => 'localhost') do
  def log *args
    p args
  end
  
  clock = MQ.new.fanout('clock')
  EM.add_periodic_timer(1){
    log :publishing, time = Time.now
    clock.publish(Marshal.dump(time))
  }

  amq = MQ.new
  amq.queue('every second').bind(amq.fanout('clock')).subscribe{ |time|
    log 'every second', :received, Marshal.load(time)
  }
  
  amq = MQ.new
  amq.queue('every 5 seconds').bind(amq.fanout('clock')).subscribe{ |time|
    time = Marshal.load(time)
    log 'every 5 seconds', :received, time if time.strftime('%S').to_i%5 == 0
  }
end

Okay, So first off, we created a new fanout and using MQ and EventMachine again, we’ve created a periodic timer that will publish to the clock fanout (again, a 1:N communication). Then we bind (recall a binding is just a link to the queue’s “router”) to the fanout named “clock” and subscribe to it whn there is a new message in the clock queue. Notice we take two differently named queues (one is called “every second” and the other “every 5 seconds”) and bind them to the same fanout. This works because the fanout provides the 1:N communication again.

Now, notice that we are only logging if the time mod 5 is zero, or essentially every 5 seconds. However, if we remove that if block, then you’ll notice that the queue still receives messages. This is because the fanout is publishing the message to EVERYONE who is subscribed to the fanout. That’s how the fanout works.

Now, the AMQP library provides something really kind of neat. You can send “RPC” calls across the wire to the queue. How? Let’s look at another example from the examples/ directory:

require "rubygems"
require 'mq'

AMQP.start(:host => 'localhost') do

  def log *args
    p args
  end

  class HashTable < Hash
    def get(key)
      self[key]
    end
    
    def set(key, value)
      self[key] = value
    end
  end

  server = MQ.new.rpc('hash table node', HashTable.new)

  client = MQ.new.rpc('hash table node')
  client.set(:now, time = Time.now)
  client.get(:now) do |res|
    log 'client', :now => res, :eql? => res == time
  end

  client.set(:one, 1)
  client.get :one do |res| 
    p res
  end
  client.keys do |res|
    log 'client', :keys => res
    AMQP.stop{ EM.stop }
  end

end

Sweetness. We are taking a HashTable object and storing it’s in a Marshalled format (thanks to the amqp gem) into the queue. When we pop it out, we get a full object back.

A few other notes.

We can create a topic exchange, which is kind of like a fanout, except that instead of 1:N, it’s N:N. To do this, we’ll write a mini chat server (well, we’ll pull it from the very smart blog: http://www.randomhacks.net/articles/2009/05/08/chat-client-ruby-amqp-eventmachine-shoes). Some more quick overhead before we get to the code. When we created the binding before, we set a :key that was equivalent to the name of the queue. We can do better than that and specifically set it to an exact queue channel via the bind method:

queue.bind('chat', :key => $channel)

This sets a binding on the chat queue to match any key with the name of the channel in the chat program. You can set wildcards, and arrays in the key parameter. Now, when we publish a message, we’ll use the routing key to ensure it gets sent to the right bindings:

MQ.queue('chat').publish(" -> #{data}", :routing_key => $channel)

Now, on to the entire program:

require 'rubygems'
gem 'amqp'
require 'mq'

unless ARGV.length == 2
  STDERR.puts "Usage: #{$0} <channel> <nick>"
  exit 1
end
$channel, $nick = ARGV

AMQP.start(:host => 'localhost') do
  $chat = MQ.topic('chat')

  # Print any messages on our channel.
  queue = MQ.queue($nick)
  queue.bind('chat', :key => $channel)
  queue.subscribe do |msg|
    if msg.index("#{$nick}:") != 0
      puts msg
    end
  end

  # Forward console input to our channel.
  module KeyboardInput
    include EM::Protocols::LineText2
    def receive_line data
      $chat.publish("#{$nick}: #{data}",
                    :routing_key => $channel)
    end
  end
  EM.open_keyboard(KeyboardInput)
end

Last example of the day: We can distribute work in our programs as well.

require "rubygems"
require 'mq'

MAX = 1000

# logging
def log *args
  p args
end

# spawn workers
workers = ARGV[0] ? (Integer(ARGV[0]) rescue 1) : 1
AMQP.fork(workers) do

  log MQ.id, :started

  class Fixnum
    def prime?
      ('1' * self) !~ /^1?$|^(11+?)\1+$/
    end
  end

  class PrimeChecker
    def is_prime? number
      log "prime checker #{MQ.id}", :prime?, number
      number.prime?
    end
  end

  MQ.rpc('prime checker', PrimeChecker.new)

end

# use workers to check which numbers are prime
AMQP.start(:host => 'localhost') do
  
  prime_checker = MQ.rpc('prime checker')

  (10_000...(10_000+MAX)).each do |num|
    log :checking, num

    prime_checker.is_prime?(num) { |is_prime|
      log :prime?, num, is_prime
      (@primes||=[]) << num if is_prime
      
      if (@responses = (@responses || 0) + 1) == MAX
        log :primes=, @primes
        EM.stop_event_loop
      end
    }

  end
  
end

By using the fork method in AMQP, we are forking several consumers of the messages in the ‘prime checker’ rpc queue. This shows just how easy it is to multithread a consumer for an amqp queue.