play/type blog

I did some refactoring this week, and the result is a much improved workling. Adding new work brokers is a snap now. I’ll show you how I added RabbitMQ, so that you can go about adding your own brokers. So without further ado…

RabbitMQ, a quick introduction

A lot of Ruby people have been talking about using RabbitMQ as their Queue of choice. Soundcloud.com are using it, as is new bamboo founder Johnathan Conway, who is using it at his video startup http://www.vzaar.com/. He says:

RabbitMQ – Now this is the matrons knockers when it comes to kick ass, ultra fast and scalable messaging. It simply rocks, with performance off the hook. It’s written in Erlang and supports the AMPQ protocol.

Follow the instructions here to get this beauty installed.

Adding Ampq to Workling

There are two new base classes you can extend to add new brokers. I’ll describe how this is done, but the code i show is already a part of workling. Skip to ‘activating amqp in your application’ to see how this is activated.

Clients

Clients help workling to connect to job brokers. To add an AmqpClient, we need to extend from Workling::Client::Base and implement a couple of methods.

require 'workling/clients/base'
require 'mq'

#
#  An Ampq client
#
module Workling
  module Clients
    class AmqpClient < Workling::Clients::Base

      # starts the client. 
      def connect
        @amq = MQ.new
      end

      # stops the client.
      def close
        @amq.close
      end

      # request work
      def request(queue, value)
        @amq.queue(queue).publish(value)
      end

      # retrieve work
      def retrieve(queue)
        @amq.queue(queue)
      end

      # subscribe to a queue
      def subscribe(queue)
        @amq.queue(queue).subscribe do |value|
          yield value
        end
      end

    end
  end
end

Were’s using the eventmachine amqp client for this, you can find it up on github. connect and close do exactly what it says on the tin: connecting to rabbitmq and closing the connection.

request and retrieve are responsible for placing work on rabbitmq. The methods are passed the correct queue, and a value that contains the worker method arguments. If you need control over the queue names, look at the RDoc for Workling::Routing::Base. In our case, there’s no special requirement here.

Finally, we implement a subscribe method. Use this if your broker supports callbacks, as is the case with amqp. This method expects to a block, which we pass into the amqp subscribe method here. The block will be called when a message is available on the queue, and the result is yielded into the block.

Having subscription callbacks is very nice, because this way, we don’t need to keep calling get on the queue to see if something new is waiting.

So now we’re done! That’s all you need to add RabbitMQ to workling. Configure it in your application as descibed below.

Invokers

There’s still potential to improve things though. Workling 0.4.0 introduces the idea of invokers. Invokers grab work off a job broker, using a client (see above). They subclass Workling::Remote::Invokers::Base. Read the RDoc for a description of the methods.

Workling comes with a couple of standard invokers, like the BasicPoller. This invoker simply keeps hitting the broker every n seconds, checking for new work and executing it immediately. The ThreadedInvoker does the same, but spawns a Thread for every Worker class the project defines.

So Amqp: it would be nice if we had an invoker that makes use of the subscription callbacks. Easily done, lets have a look:

require 'eventmachine'
require 'workling/remote/invokers/base'

#
#  Subscribes the workers to the correct queues. 
# 
module Workling
  module Remote
    module Invokers
      class EventmachineSubscriber < Workling::Remote::Invokers::Base

        def initialize(routing, client_class)
          super
        end

        #
        #  Starts EM loop and sets up subscription callbacks for workers. 
        #
        def listen
          EM.run do
            connect do
              routes.each do |queue|
                @client.subscribe(queue) do |args|
                  run(queue, args)
                end
              end
            end
          end
        end

        def stop
          EM.stop if EM.reactor_running?
        end
      end
    end
  end
end

Invokers have to implement two methods, listen and stop. Listen starts the main listener loop, which is responsible for starting work when it becomes available.

In our case, we need to start an EM loop around listen. This is because the Ruby AMQP library needs to run inside of an eventmachine reactor loop.

Next, inside of listen, we need to iterate through all defined routes. There is a route for each worker method you defined in your application. The routes double as queue names. For this, you can use the helper method routes. Now we attach a callback to each queue. We can use the helper method run, which executes the worker method associated with the queue, passing along any supplied arguments.

That’s it! We now have a more effective Invoker, which we can activate in our application like this…

Configuring your application to use AMQP (with RabbitMQ)

Again, follow the instructions here to get RabbitMQ running on your OSX development machine.

Workling::Remote.invoker = Workling::Remote::Invokers::EventmachineSubscriber
Workling::Remote.dispatcher = Workling::Remote::Runners::ClientRunner.new
Workling::Remote.dispatcher.client = Workling::Clients::AmqpClient.new

enjoy!