@@ -3,7 +3,7 @@ SweatShop provides an api to background resource intensive tasks. Much of the api design was copied from Workling, with a few tweaks. Currently, it runs rabbitmq and kestrel, but it can support any number of queues. -## Installing +## Installing gem install sweat_shop freeze in your gems directory (add config.gem 'sweat_shop' to your environment) @@ -32,7 +32,7 @@ call: That will do the work immediately, without placing the task on the queue. You can also define a `queue_group` at the top of the file which will allow you to split workers out into logical groups. This is important if you have various machines serving different -queues. +queues. ## Running the queue @@ -44,7 +44,43 @@ http://github.com/robey/kestrel/tree/master Rabbit: http://github.com/ezmobius/nanite/tree/master -config/sweatshop.yml specifies the machine address of the queue (default localhost:5672). You can also specify the queue type with the queue param. +config/sweatshop.yml specifies the machine address of the queue +(default localhost:5672). You can also specify the queue type with the +queue param. + +## Rabbit cluster support + +The following example configuration shows support for Rabbit clusters +within a queue group: + + default: + queue: rabbit + cluster: + - host: hostA + port: 5672 + - host: hostB + port: 5672 + user: 'guest' + pass: 'guest' + vhost: '/' + enable: true + +Sweatshop will attempt to connect to each server listed under +"cluster" in order, until it either manages to establish a connection +or until it runs out of servers. + +If you only have a single Rabbit server, you can omit the "cluster" +option and just add "host" and "port" options, as shown below: + + default: + queue: rabbit + host: localhost + port: 5672 + user: 'guest' + pass: 'guest' + vhost: '/' + enable: true + ## Running the workers README.markdown @@ -19,7 +19,7 @@ module MessageQueue end def enqueue(queue, data) - send_command do + send_command do client.queue(queue, :durable => true).publish(Marshal.dump(data), :persistent => true) end end @@ -55,14 +55,36 @@ module MessageQueue end def client - @client ||= Carrot.new( - :host => @opts['host'], - :port => @opts['port'].to_i, - :user => @opts['user'], - :pass => @opts['pass'], - :vhost => @opts['vhost'], - :insist => @opts['insist'] - ) + if @opts['cluster'] + @opts['cluster'].each_with_index do |server, i| + begin + @client ||= Carrot.new( + :host => server['host'], + :port => server['port'].to_i, + :user => @opts['user'], + :pass => @opts['pass'], + :vhost => @opts['vhost'], + :insist => @opts['insist'] + ) + rescue Carrot::AMQP::Server::ServerDown => e + if i == (@opts['cluster'].size-1) + raise e + else + next + end + end + end + else + @client ||= Carrot.new( + :host => @opts['host'], + :port => @opts['port'].to_i, + :user => @opts['user'], + :pass => @opts['pass'], + :vhost => @opts['vhost'], + :insist => @opts['insist'] + ) + end + @client end def stop lib/message_queue/rabbit.rb @@ -10,7 +10,7 @@ module SweatShop if method.to_s =~ /^async_(.*)/ method = $1 expected_args = instance.method(method).arity - if expected_args != args.size + if expected_args != args.size raise ArgumentError.new("#{method} expects #{expected_args} arguments") end return instance.send(method, *args) unless config['enable'] @@ -66,7 +66,7 @@ module SweatShop do_task(task) end end - + def self.do_tasks while task = dequeue do_task(task) @@ -150,7 +150,7 @@ module SweatShop instance.stop end - # called before we exit -- subclass can implement this method + # called before we exit -- subclass can implement this method def stop; end; lib/sweat_shop/worker.rb @@ -9,29 +9,54 @@ class WorkerTest < Test::Unit::TestCase end def teardown + SweatShop.instance_variable_set("@config", nil) + SweatShop.instance_variable_set("@queues", nil) File.delete(HelloWorker::TEST_FILE) if File.exist?(HelloWorker::TEST_FILE) end + def send_message + HelloWorker.async_hello('Amos') + + worker = File.expand_path(File.dirname(__FILE__) + '/hello_worker') + sweatd = "#{File.dirname(__FILE__)}/../lib/sweat_shop/sweatd.rb" + + `ruby #{sweatd} --worker-file #{worker} start` + `ruby #{sweatd} stop` + + File.delete('sweatd.log') if File.exist?('sweatd.log') + assert_equal 'Hi, Amos', File.read(HelloWorker::TEST_FILE) + end + should "daemonize" do begin SweatShop.config['enable'] = true SweatShop.logger = :silent - HelloWorker.async_hello('Amos') + send_message - worker = File.expand_path(File.dirname(__FILE__) + '/hello_worker') - sweatd = "#{File.dirname(__FILE__)}/../lib/sweat_shop/sweatd.rb" + rescue Exception => e + puts e.message + puts e.backtrace.join("\n") + fail "\n\n*** Functional test failed, is the rabbit server running on localhost? ***\n" + end + end - `ruby #{sweatd} --worker-file #{worker} start` - `ruby #{sweatd} stop` + should "connect to fallback servers if the default one is down" do + begin + SweatShop.config['enable'] = true + + SweatShop.config['default']['cluster'] = + [ + { 'host' => 'localhost', 'port' => 5671}, #invalid + { 'host' => 'localhost', 'port' => 5672} #valid + ] - File.delete('sweatd.log') if File.exist?('sweatd.log') - assert_equal 'Hi, Amos', File.read(HelloWorker::TEST_FILE) + send_message rescue Exception => e puts e.message puts e.backtrace.join("\n") fail "\n\n*** Functional test failed, is the rabbit server running on localhost? ***\n" end end - + end test/test_functional_worker.rb 1bb68f7151ae48b412e708c493bcb6f252d7b2a2 Ubiratan Pires Alberton u.alberton@gmail.com http://github.com/Bira/sweat_shop/commit/7b232cbfca8123abba93e12d6472bdeca249fc9c 7b232cbfca8123abba93e12d6472bdeca249fc9c 2009-09-30T10:50:10-07:00 2009-09-30T10:50:10-07:00 Add Rabbit cluster and failover support. 07e2302bbe497fa3031e152554766476c1fc69a7 Ubiratan Pires Alberton u.alberton@gmail.com