README.textile examples/queue_polling.rb examples/queue_statistics.rb examples/standalone_poller.rb examples/standalone_producer.rb script/daemonize test/acceptance/queue_management_test.rb test/acceptance/queue_messaging_test.rb vendor/beanstalk-client-1.0.2/beanstalk-client.rb vendor/beanstalk-client-1.0.2/beanstalk-client/connection.rb vendor/beanstalk-client-1.0.2/beanstalk-client/errors.rb vendor/beanstalk-client-1.0.2/beanstalk-client/job.rb vendor/beanstalk-client-1.0.2/beanstalk-client/version.rb @@ -3,13 +3,25 @@ require 'rake/testtask' require 'rake/rdoctask' desc 'Default: run unit tests.' -task :default => :test +task :default => 'test:unit' -desc 'Test the beanstalk_messaging plugin.' -Rake::TestTask.new(:test) do |t| - t.libs << 'lib' - t.pattern = 'test/**/*_test.rb' - t.verbose = true +namespace :test do + desc 'Test the beanstalk_messaging plugin.' + Rake::TestTask.new(:unit) do |t| + t.libs << 'lib' + t.pattern = 'test/*_test.rb' + t.verbose = true + end + + desc 'End to end acceptance tests, requires beanstalkd daemon.' + Rake::TestTask.new(:acceptance) do |t| + t.libs << 'lib' + t.pattern = 'test/acceptance/*_test.rb' + t.verbose = true + end + + desc 'Run all tests' + task :all => [:unit, :acceptance] end desc 'Generate documentation for the beanstalk_messaging plugin.' Rakefile @@ -1,3 +1,3 @@ -$LOAD_PATH << File.join(File.dirname(__FILE__), *%w[vendor/beanstalk-client-0.6.0]) +$LOAD_PATH << File.join(File.dirname(__FILE__), *%w[vendor/beanstalk-client-1.0.2]) require 'beanstalk_messaging' \ No newline at end of file init.rb @@ -4,6 +4,17 @@ module Beanstalk @pool = pool @stale = false end + + def self.connect(host, port, timeout = 10000) + Timeout.timeout(timeout) do + pool = Beanstalk::Pool.new(["#{host}:#{port}"]) + if pool.open_connections.length >= 1 + new(pool) + else + raise ConnectionError.new("Beanstalk::Pool connection failed") + end + end + end def stale? @stale lib/beanstalk/queue.rb @@ -1,10 +1,15 @@ module Beanstalk class QueueManager - def initialize(config_path) - @config = YAML.load(File.open(config_path)) + def initialize(config) + @config = config @queues = {} end + def self.load(config_path) + config = YAML.load(File.open(config_path)) + new(config) + end + def available_queues @config.keys end @@ -31,21 +36,11 @@ module Beanstalk raise Messaging::UnknownQueue.new("Unknown queue: #{queue_name}. Check your configuration.") unless @config[queue_name] host, port = @config[queue_name][:host], @config[queue_name][:port] begin - Queue.new(create_pool(host, port)) - rescue Errno::ECONNREFUSED, Errno::EADDRNOTAVAIL + Queue.connect(host, port, Beanstalk.connection_timeout) + rescue ConnectionError NullQueue.new - rescue ConnectionTimeout - NullQueue.new - end - end - - def create_pool(host, port) - begin - Timeout::timeout(Beanstalk.connection_timeout) do - Beanstalk::Pool.new(["#{host}:#{port}"]) - end rescue Timeout::Error - raise ConnectionTimeout + NullQueue.new end end end lib/beanstalk/queue_manager.rb @@ -22,7 +22,7 @@ module Beanstalk end def api_connection - RawConnection.new("#{@host}:#{@port}") + Connection.new("#{@host}:#{@port}") rescue Errno::ECONNREFUSED nil end lib/beanstalk_manager.rb @@ -15,45 +15,17 @@ module Beanstalk end end - class RawConnection - def kick(max_jobs) - @socket.write("kick #{max_jobs}\r\n") - check_resp('KICKED')[0].to_i - end - end - - class Connection - def kick(max_jobs) - raw_connection = @free.take() - raw_connection.kick(max_jobs) - end - end - - # File lib/beanstalk-client/connection.rb, line 199 - # monkey-patch for version 0.6 of beanstalk-client gem class Pool - def connect() - @connections ||= {} - @addrs.each do |addr| - begin - if !@connections.include?(addr) - @connections[addr] = CleanupWrapper.new(addr, self) - end - # WE WANT THE EXCEPTIONS TO BE RAISED - # rescue Exception => ex - # puts "#{ex.class}: #{ex}" - # #puts begin ex.fixed_backtrace rescue ex.backtrace end - end - end - @connections.size + def stdout_messages + @stdout_messages ||= [] end - def kick(max_jobs = 1) - pick_connection.kick(max_jobs) + def puts(message) + stdout_messages << message end end - class ConnectionTimeout < Exception; end + class ConnectionError < StandardError; end end require 'beanstalk/queue' lib/beanstalk_messaging.rb @@ -3,7 +3,7 @@ require 'yaml' require 'pp' require 'stringio' -$LOAD_PATH << File.join(File.dirname(__FILE__), *%w[.. vendor plugins beanstalk_messaging vendor beanstalk-client-0.6.0]) +$LOAD_PATH << File.join(File.dirname(__FILE__), *%w[.. vendor plugins beanstalk_messaging vendor beanstalk-client-1.0.2]) $LOAD_PATH << File.join(File.dirname(__FILE__), *%w[.. vendor plugins beanstalk_messaging lib]) require 'beanstalk_manager' @@ -60,7 +60,7 @@ begin $stdout = StringIO.new title = "Beanstalk Queue Statistics" - queue_manager = Beanstalk::QueueManager.new(CONFIG_PATH) + queue_manager = Beanstalk::QueueManager.load(CONFIG_PATH) stats_printer = Beanstalk::StatisticsTablePrinter.new(queue_manager, title) rendered_table = stats_printer.render(*MANAGER.configured_queues) $stdout = STDOUT script/beanstalk @@ -2,9 +2,8 @@ require File.dirname(__FILE__) + '/test_helper' require 'beanstalk-client/version' class BeanstalkTest < Test::Unit::TestCase - def test_should_be_using_version_0_point_6_of_the_gem - assert_equal '0.6.0', Beanstalk::VERSION::STRING, - "We currently monkey-patch the Pool class - you'll need to check this before upgrading the version" + def test_should_be_using_version_1_point_0_point_2_of_the_beanstalk_client_gem + assert_equal '1.0.2', Beanstalk::VERSION::STRING end def test_should_have_a_default_connection_timeout_of_1_second test/beanstalk_messaging_test.rb @@ -2,25 +2,24 @@ require File.dirname(__FILE__) + '/test_helper' class BeanstalkQueueManagerTest < Test::Unit::TestCase - def setup - @config_path = '/tmp/beanstalk.yml' - - File.stubs(:open).with(@config_path).returns(config_file = stub('io')) - YAML.stubs(:load).with(config_file).returns({ + def setup + @queue_manager = Beanstalk::QueueManager.new({ :queue_one => { :host => 'localhost', :port => 12000 }, :queue_two => { :host => 'localhost', :port => 12001 } }) - - @queue_manager = Beanstalk::QueueManager.new(@config_path) end def test_should_return_a_beanstalk_queue_using_the_config_for_the_requested_queue_if_it_exists_in_the_config - Beanstalk::Pool.stubs(:new).with(['localhost:12000']).returns(pool = stub('pool')) - Beanstalk::Queue.stubs(:new).with(pool).returns(queue = stub('queue', :stale? => false)) - + Beanstalk::Queue.expects(:connect).with('localhost', 12000, anything).returns(queue = stub('queue', :stale? => false)) assert_equal queue, @queue_manager.queue(:queue_one) end + def test_should_use_connection_timeout_when_connecting_to_a_queue + Beanstalk.connection_timeout = 2 + Beanstalk::Queue.expects(:connect).with(anything, anything, 2).returns(stub_everything('queue')) + @queue_manager.queue(:queue_one) + end + def test_should_raise_an_exception_when_trying_to_access_a_queue_which_is_not_defined_in_the_config assert_raises(Messaging::UnknownQueue) do @queue_manager.queue(:queue_three) @@ -28,41 +27,28 @@ class BeanstalkQueueManagerTest < Test::Unit::TestCase end def test_should_return_the_same_instance_of_a_queue_if_asking_for_the_same_queue_more_than_once - Beanstalk::Pool.stubs(:new).with(['localhost:12000']).returns(pool = stub('pool')) - Beanstalk::Queue.stubs(:new).with(pool).returns(queue = stub('queue', :stale? => false)) - + Beanstalk::Queue.expects(:connect).once.returns(queue = stub('queue', :stale? => false)) assert_equal queue, @queue_manager.queue(:queue_one) assert_equal queue, @queue_manager.queue(:queue_one) end def test_should_return_a_new_instance_of_a_queue_if_existing_queue_has_been_marked_as_stale - Beanstalk::Pool.stubs(:new).with(['localhost:12000']).returns(pool = stub('pool')) - Beanstalk::Queue.stubs(:new).with(pool).returns(queue_instance_one = stub('queue 1', :stale? => false)) - + Beanstalk::Queue.stubs(:connect).returns(queue_instance_one = stub('queue 1', :stale? => false)) assert_equal queue_instance_one, @queue_manager.queue(:queue_one) queue_instance_one.stubs(:stale?).returns(true) - Beanstalk::Pool.stubs(:new).with(['localhost:12000']).returns(pool_two = stub('pool')) - Beanstalk::Queue.stubs(:new).with(pool_two).returns(queue_instance_two = stub('queue 2', :stale? => false)) - + + Beanstalk::Queue.expects(:connect).returns(queue_instance_two = stub('queue 2', :stale? => false)) assert_equal queue_instance_two, @queue_manager.queue(:queue_one) end - def test_should_return_a_null_queue_if_queue_connection_cannot_be_established - Beanstalk::Pool.stubs(:new).raises(Errno::ECONNREFUSED) - - assert_instance_of Beanstalk::NullQueue, @queue_manager.queue(:queue_two) - end - - def test_should_return_a_null_queue_if_queue_connection_address_is_unavailable - Beanstalk::Pool.stubs(:new).raises(Errno::EADDRNOTAVAIL) - + def test_should_return_a_null_queue_if_a_connection_error_occurs + Beanstalk::Queue.stubs(:connect).raises(Beanstalk::ConnectionError) assert_instance_of Beanstalk::NullQueue, @queue_manager.queue(:queue_two) end def test_should_return_null_queue_if_pool_times_out_while_connecting - @queue_manager.stubs(:create_pool).raises(Beanstalk::ConnectionTimeout) - + Beanstalk::Queue.stubs(:connect).raises(Timeout::Error) assert_instance_of Beanstalk::NullQueue, @queue_manager.queue(:queue_two) end test/beanstalk_queue_manager_test.rb @@ -97,7 +97,6 @@ class LazyConnectionPool end class BeanstalkQueueWithTimingOutConnectionPool < Test::Unit::TestCase - def setup @connection_pool = LazyConnectionPool.new @queue = Beanstalk::Queue.new(@connection_pool) @@ -109,5 +108,34 @@ class BeanstalkQueueWithTimingOutConnectionPool < Test::Unit::TestCase @queue.push('foobar') assert @queue.stale? end +end + +class BeanstalkQueueConnectionTest < Test::Unit::TestCase + + def test_should_establish_a_pool_connection_and_return_a_new_queue_instance + Beanstalk::Pool.expects(:new).with(["localhost:4000"]).returns(pool = stub(:open_connections => [stub])) + Beanstalk::Queue.stubs(:new).with(pool).returns(queue = stub) + assert_equal queue, Beanstalk::Queue.connect('localhost', 4000) + end + + def test_should_raise_a_connection_error_if_beanstalk_pool_has_no_open_connections_after_creating + Beanstalk::Pool.stubs(:new).returns(pool = stub(:open_connections => [])) + assert_raises(Beanstalk::ConnectionError) do + Beanstalk::Queue.connect('localhost', 4000) + end + end + + def test_should_raise_timeout_error_if_connection_cannot_be_established_in_the_specified_time + Timeout.expects(:timeout).with(5).raises(Timeout::Error) + assert_raises(Timeout::Error) do + Beanstalk::Queue.connect('localhost', 4000, timeout = 5) + end + end + + def test_should_use_a_really_long_timeout_duration_to_simulate_no_timeout_if_not_specified + Timeout.expects(:timeout).with(long_time = 10000).yields + Beanstalk::Pool.stubs(:new).returns(stub(:open_connections => [stub])) + Beanstalk::Queue.connect('localhost', 4000) + end end \ No newline at end of file test/beanstalk_queue_test.rb README vendor/beanstalk-client-0.6.0/beanstalk-client.rb vendor/beanstalk-client-0.6.0/beanstalk-client/bag.rb vendor/beanstalk-client-0.6.0/beanstalk-client/connection.rb vendor/beanstalk-client-0.6.0/beanstalk-client/job.rb vendor/beanstalk-client-0.6.0/beanstalk-client/version.rb f5c370876b4c72f880171a2db7d3714f24dce479 lukeredpath lukeredpath@53438b35-87ee-0310-8825-c8c94c0097b6 http://github.com/lukeredpath/beanstalk-messaging/commit/6d1d19a84585c9ab682d1a8dc79c03817c93625e 6d1d19a84585c9ab682d1a8dc79c03817c93625e 2008-07-25T03:08:56-07:00 2008-07-25T03:08:56-07:00 Merge branch 'master' into revieworld git-svn-id: http://svn/repos/plugins/beanstalk_messaging@11665 53438b35-87ee-0310-8825-c8c94c0097b6 2b2f40181584669da854029d904edaf0e3fa96ad lukeredpath lukeredpath@53438b35-87ee-0310-8825-c8c94c0097b6