lib/rosetta_queue/exception_handler.rb spec/rosetta_queue/exception_handler_spec.rb spec/rosetta_queue/message_handler_spec.rb @@ -4,3 +4,4 @@ tmp/* *.swp log/* pkg +features/support/tmp .gitignore @@ -1,6 +1,9 @@ -== 0.2.x (git) +== 0.4.x (git) +== 0.4.0 === New features + * Allows for custom exception handling logic to be registered for publishing and consuming actions. See ExceptionHandler for more info. (Ben Mabey) + * RQ was doing a blanket rescue for publishing and consuming and then logging the errors. If you still want this behaviour you will need to register a block/class to do that. Again, see ExceptionHandler for more details. * Added core time extension when ActiveSupport not present. (Chris Wyckoff) === Bufixes * Closing the connection after publishing message now instead of unsubscribing. (Chris Wyckoff) @@ -29,7 +32,7 @@ * Allows user to purge a queue. * Works for AMQP adapters only. * Beanstalk Adaper (David Brady) - * Still needs some work to have it take advantage of beanstalk's subscribe funtionality. + * Still needs some work to have it take advantage of beanstalk's subscribe functionality. == 0.1.0 / 2008-01-28 - Initial Release RosettaQueue was realased in the wild! RQ's initial development was primarily sponsored by Alliance Health Networks (thanks!!). The original authors were Chris Wyckoff and Ben Mabey. The initial release included adapters for stomp and amqp, in addition to a null and fake adapters for testing. History.txt @@ -1,10 +1,10 @@ require 'rubygems' require 'spec/rake/spectask' require 'cucumber/rake/task' - + Cucumber::Rake::Task.new do |t| t.cucumber_opts = "--format pretty" -end +end desc "Run the specs under spec" Spec::Rake::SpecTask.new do |t| Rakefile @@ -11,6 +11,7 @@ require 'rosetta_queue/exceptions' require 'rosetta_queue/filters' require 'rosetta_queue/logger' require 'rosetta_queue/message_handler' +require 'rosetta_queue/exception_handler' require 'rosetta_queue/producer' require 'rosetta_queue/consumer_managers/base' require 'rosetta_queue/consumer_managers/threaded' lib/rosetta_queue.rb @@ -68,7 +68,7 @@ module RosettaQueue ack = @options[:ack] queue.subscribe(@options) do |header, msg| RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") - message_handler.on_message(Filters.process_receiving(msg)) + message_handler.handle_message(msg) header.ack if ack end end @@ -107,7 +107,7 @@ module RosettaQueue queue.bind(exchange).subscribe(@options) do |header, msg| RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") - message_handler.on_message(Filters.process_receiving(msg)) + message_handler.handle_message(msg) header.ack if ack end end lib/rosetta_queue/adapters/amqp_evented.rb @@ -68,7 +68,7 @@ module RosettaQueue @queue = conn.queue(destination, @options) @queue.subscribe(@options) do |msg| RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") - message_handler.on_message(Filters.process_receiving(msg)) + message_handler.handle_message(msg) @queue.ack if ack end end @@ -100,7 +100,7 @@ module RosettaQueue @queue.bind(exchange) @queue.subscribe(@options) do |msg| RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") - message_handler.on_message(Filters.process_receiving(msg)) + message_handler.handle_message(msg) @queue.ack if ack end end lib/rosetta_queue/adapters/amqp_synch.rb @@ -10,7 +10,7 @@ module RosettaQueue end def destination_for(message_handler) - raise DestinationNotFound.new("Missing destination!") unless message_handler.destination + raise DestinationNotFound.new("Missing destination on message handler #{message_handler.inspect}.") unless message_handler.destination @dest ||= Destinations.lookup(message_handler.destination.to_sym) end lib/rosetta_queue/adapters/base.rb @@ -36,7 +36,7 @@ module RosettaQueue running do msg = receive.body RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") - message_handler.on_message(filter_receiving(msg)) + message_handler.handle_message(msg) end end lib/rosetta_queue/adapters/beanstalk.rb @@ -47,7 +47,7 @@ module RosettaQueue msg = receive(options).body Thread.current[:processing] = true RosettaQueue.logger.info("Receiving from #{destination} :: #{msg}") - message_handler.on_message(filter_receiving(msg)) + message_handler.handle_message(msg) Thread.current[:processing] = false end end lib/rosetta_queue/adapters/stomp.rb @@ -37,6 +37,22 @@ module RosettaQueue @receiving.call(message) end + def safe_process_sending(message) + safe(:process_sending, message) + end + + def safe_process_receiving(message) + safe(:process_receiving, message) + end + + private + + def safe(filter_call, message) + send(filter_call) + rescue StandardError + message + end + end end end lib/rosetta_queue/filters.rb @@ -2,7 +2,6 @@ module RosettaQueue module MessageHandler module ClassMethods - attr_reader :destination, :options_hash def options(options = {}) @@ -20,20 +19,34 @@ module RosettaQueue def self.included(receiver) receiver.extend(ClassMethods) - attr_accessor :adapter_proxy + end - def destination - self.class.destination - end + attr_accessor :adapter_proxy - def options_hash - self.class.options_hash - end + def destination + self.class.destination + end - def ack - adapter_proxy.ack unless adapter_proxy.nil? + def options_hash + self.class.options_hash + end + + def handle_message(unfiltered_message) + ExceptionHandler::handle(:publishing, + lambda { + { :message => Filters.safe_process_receiving(unfiltered_message), + :destination => destination, + :action => :consuming, + :options => options_hash + } + } ) do + on_message(Filters.process_receiving(unfiltered_message)) end + end + def ack + adapter_proxy.ack unless adapter_proxy.nil? end + end end lib/rosetta_queue/message_handler.rb @@ -4,11 +4,18 @@ module RosettaQueue include MessageHandler def self.publish(destination, message, options = {}) - RosettaQueue::Adapter.instance.send_message(Destinations.lookup(destination), Filters.process_sending(message), options) - - rescue Exception => e - RosettaQueue.logger.error("Caught exception in Consumer.publish: #{$!}\n" + e.backtrace.join("\n\t")) + ExceptionHandler::handle(:publishing, + lambda { + {:message => Filters.safe_process_sending(message), + :action => :publishing, + :destination => destination, + :options => options} + }) do + RosettaQueue::Adapter.instance.send_message( + Destinations.lookup(destination), + Filters.process_sending(message), + options) + end end - end end lib/rosetta_queue/producer.rb @@ -28,7 +28,7 @@ module RosettaQueue # Example: # consume_once_with ClientStatusConsumer def consume_once_with(consumer) - consumer.new.on_message(RosettaQueue::Consumer.receive(consumer.destination)) + consumer.new.handle_message(RosettaQueue::Consumer.receive(consumer.destination)) end # Consumes the first message on queue and returns it. lib/rosetta_queue/spec_helpers/helpers.rb @@ -20,10 +20,9 @@ module RosettaQueue::Gateway describe "#do_exchange" do - it "should filter the message and forward it to the handler" do + it "should delegate message handling to the message handler" do when_receiving_exchange { - ::RosettaQueue::Filters.should_receive(:process_receiving).with(@msg).and_return("Filtered Message") - @handler.should_receive(:on_message).with("Filtered Message") + @handler.should_receive(:handle_message).with("Hello World!") } end end @@ -35,7 +34,7 @@ module RosettaQueue::Gateway RosettaQueue.logger.stub!(:info) @msg = "Hello World!" @adapter = AmqpSynchAdapter.new({:user => "foo", :password => "bar", :host => "localhost"}) - @handler = mock("handler", :on_message => true, :destination => :foo, :options_hash => {:durable => true}) + @handler = mock("handler", :handle_message => true, :destination => :foo, :options_hash => {:durable => true}) end describe AmqpSynchAdapter do @@ -68,7 +67,7 @@ module RosettaQueue::Gateway end before(:each) do - @handler = mock("handler", :on_message => true, :destination => :foo, :options_hash => {:durable => true }) + @handler = mock("handler", :handle_message => true, :destination => :foo, :options_hash => {:durable => true }) end it "should pass message handler to exchange strategy" do @@ -97,7 +96,7 @@ module RosettaQueue::Gateway @queue = mock("Bunny::Queue", :pop => @msg, :publish => true, :unsubscribe => true) Bunny.stub!(:new).and_return(@conn = mock("Bunny::Client", :queue => @queue, :exchange => @exchange, :status => :connected, :stop => nil)) @queue.stub!(:subscribe).and_yield(@msg) - @handler = mock("handler", :on_message => true, :destination => :foo) + @handler = mock("handler", :handle_message => true, :destination => :foo) @exchange = SynchExchange::DirectExchange.new({:user => 'user', :password => 'pass', :host => 'host', :opts => {:vhost => "foo"}}) end @@ -170,7 +169,7 @@ module RosettaQueue::Gateway @queue = mock("Bunny::Queue", :pop => @msg, :bind => @bound_queue = mock("Bunny::Queue", :pop => @msg), :publish => true, :unbind => true) Bunny.stub!(:new).and_return(@conn = mock("Bunny::Client", :queue => @queue, :exchange => @exchange, :status => :connected)) @queue.stub!(:subscribe).and_yield(@msg) - @handler = mock("handler", :on_message => true, :destination => :foo, :options => {:durable => false}) + @handler = mock("handler", :handle_message => true, :destination => :foo, :options => {:durable => false}) end def do_receiving_exchange @@ -214,7 +213,7 @@ module RosettaQueue::Gateway it "should forward the message body onto the handler" do when_receiving_exchange { - @handler.should_receive(:on_message).with("Hello World!") + @handler.should_receive(:handle_message).with("Hello World!") } end spec/rosetta_queue/adapters/amqp_synchronous_spec.rb @@ -9,7 +9,7 @@ module RosettaQueue before(:each) do @msg = "Hello World!" - @handler = mock('handler', :on_message => "", :destination => :foo) + @handler = mock('handler', :handle_message => "", :destination => :foo) @msg_obj = mock("message", :body => @msg, :delete => true) @conn = mock("Beanstalk::Pool", :put => true, :reserve => @msg_obj) ::Beanstalk::Pool.stub!(:new).and_return(@conn) spec/rosetta_queue/adapters/beanstalk_spec.rb @@ -9,7 +9,7 @@ module RosettaQueue before(:each) do @msg = "Hello World!" - @handler = mock('handler', :destination => :foo, :options_hash => {:persistent => false, :ack => "client"}, :on_message => "") + @handler = mock('handler', :destination => :foo, :options_hash => {:persistent => false, :ack => "client"}, :handle_message => "") @msg_obj = mock("message", :body => @msg, :headers => {"message-id" => 2}) @conn = mock("Stomp::Connection", :ack => true, :send => true, :subscribe => true, :receive => @msg_obj, :unsubscribe => true, :disconnect => true) ::Stomp::Connection.stub!(:open).and_return(@conn) @@ -69,7 +69,7 @@ module RosettaQueue describe "no ack" do before(:each) do - @handler = mock('handler', :destination => :foo, :options_hash => {:persistent => false}, :on_message => "") + @handler = mock('handler', :destination => :foo, :options_hash => {:persistent => false}, :handle_message => "") end it "should not acknowledge client" do spec/rosetta_queue/adapters/stomp_spec.rb @@ -22,6 +22,19 @@ module RosettaQueue end end + ['sending', 'receiving'].each do |action| + describe "#safe_process_#{action}" do + it "returns the orginal message if an exception occurs while filtering" do + Filters.define do |f| + f.send(action) { |message| raise "foo" } + end + + Filters.send("safe_process_#{action}", "Bar").should == "Bar" + end + + end + end + describe "#process_sending" do it "should process the passed in message with the defined sending filter" do spec/rosetta_queue/filters_spec.rb @@ -30,12 +30,6 @@ module RosettaQueue RosettaQueue::Adapter.stub!(:instance).and_return(@adapter) end - # it "should look up the destination defined on the class" do - # Destinations.should_receive(:lookup).with(:test_queue).and_return("/queue/test_queue") - # # when - # @gateway.publish('some message') - # end - it "should publish messages to queue with the options defined in the class" do # TO DO: REFACTOR #publish METHOD SO THAT YOU PASS IN MESSAGE HANDLER AS WITH CONSUMER pending @@ -48,18 +42,35 @@ module RosettaQueue end describe ".publish" do - # it "should look up the destination defined on the class" do - # Destinations.should_receive(:lookup).with(:test_queue).and_return("/queue/test_queue") - # # when - # Producer.publish(:test_queue, "blah") - # end - it "should send the message to the adpater along with the options" do # expect @adapter.should_receive(:send_message).with("/queue/test_queue", "Hello World!", {:persistent => true}) # when Producer.publish(:test_queue, "Hello World!", {:persistent => true}) end + + it "delgates exception handling to the ExceptionHandler for :publishing" do + ExceptionHandler.should_receive(:handle).with(:publishing, anything) + Producer.publish(:test_queue, "Hello World!", {:persistent => true}) + end + + it "wraps the publishing in an ExceptionHandler::handler block" do + @adapter.should_not_receive(:send_message) + ExceptionHandler.stub!(:handle).and_return("I was wrapped") + Producer.publish(:test_queue, "m").should == "I was wrapped" + end + + it "provides additional message information to the ExceptionHandler" do + ExceptionHandler.should_receive(:handle).with do |_, hash_proc| + hash_proc.call.should == { + :message => "message", + :action => :publishing, + :destination => :test_queue, + :options => {:persistent => true}} + end + Producer.publish(:test_queue, "message", {:persistent => true}) + end + end end spec/rosetta_queue/producer_spec.rb features/support/tmp/barconsumer.log features/support/tmp/fooconsumer.log features/support/tmp/point-to-point.log features/support/tmp/pub-sub.log 896758e1eaac8e7731c544e566d8c02ad510cea9 Ben Mabey ben@benmabey.com http://github.com/bmabey/rosetta_queue/commit/944c9778bcecbbeeb5040fd5b476b535d91fdcca 944c9778bcecbbeeb5040fd5b476b535d91fdcca 2009-09-28T15:18:05-07:00 2009-09-28T15:12:33-07:00 Allowing registration of custom exception handling. See lib/rosetta_queue/exception_handler.rb for more info. This removes some default exception handling that was present in RQ prior to this for publishing and consuming. Since every app tends to want to handle exceptions differently for messaging but are generally pretty consistent across consumers/publishers this feature gives the needed flexibility and helps reduce duplication within apps. a051043428b040e765f9728c9d15734bc689c0f2 Ben Mabey ben@benmabey.com