on_exceptions option
MODIFY
lib/beetle/message.rb
lib/beetle/message.rb
require "timeout"
module Beetle
- # Instances of class Message are created when a scubscription callback fires. Class
+ # Instances of class Message are created when a subscription callback fires. Class
# Message contains the code responsible for message deduplication and determining if it
# should retry executing the message handler after a handler has crashed (or forcefully
# aborted).
@@ -52,6 +52,8 @@ module Beetle
attr_reader :attempts_limit
# how many exceptions we should tolerate before giving up
attr_reader :exceptions_limit
+ # array of exceptions accepted to be rescued and retried
+ attr_reader :on_exceptions
# exception raised by handler execution
attr_reader :exception
# value returned by handler execution
@@ -72,6 +74,7 @@ module Beetle
@attempts_limit = opts[:attempts] || DEFAULT_HANDLER_EXECUTION_ATTEMPTS
@exceptions_limit = opts[:exceptions] || DEFAULT_EXCEPTION_LIMIT
@attempts_limit = @exceptions_limit + 1 if @attempts_limit <= @exceptions_limit
+ @on_exceptions = opts[:on_exceptions] || nil
@store = opts[:store]
max_delay = opts[:max_delay] || @delay
@max_delay = max_delay if max_delay >= 2*@delay
@@ -138,7 +141,7 @@ module Beetle
Time.now.to_i
end
- # a message has expired if the header expiration timestamp is msaller than the current time
+ # a message has expired if the header expiration timestamp is smaller than the current time
def expired?
@expires_at < now
end
@@ -218,6 +221,10 @@ module Beetle
@store.get(msg_id, :exceptions).to_i > exceptions_limit
end
+ def exception_accepted?
+ on_exceptions.nil? || on_exceptions.any?{ |klass| @exception.is_a? klass}
+ end
+
# have we already seen this message? if not, set the status to "incomplete" and store
# the message exipration timestamp in the deduplication store.
def key_exists?
@@ -347,6 +354,11 @@ module Beetle
ack!
logger.debug "Beetle: reached the handler exceptions limit: #{exceptions_limit} on #{msg_id}"
RC::ExceptionsLimitReached
+ elsif !exception_accepted?
+ completed!
+ ack!
+ logger.debug "Beetle: `#{@exception.class.name}` not accepted: `on_exceptions`=[#{on_exceptions.join(',')}] on #{msg_id}"
+ RC::ExceptionNotAccepted
else
delete_mutex!
timed_out!

MODIFY
lib/beetle/r_c.rb
lib/beetle/r_c.rb
@@ -30,6 +30,7 @@ module Beetle
rc :Ancient
rc :AttemptsLimitReached, :failure
rc :ExceptionsLimitReached, :failure
+ rc :ExceptionNotAccepted, :failure
rc :Delayed, :reject
rc :HandlerCrash, :reject
rc :HandlerNotYetTimedOut, :reject

MODIFY
test/beetle/message_test.rb
test/beetle/message_test.rb
@@ -377,6 +377,25 @@ module Beetle
assert_equal RC::ExceptionsLimitReached, message.__send__(:process_internal, proc)
end
+ test "a message should be acked if the handler crashes and the exception has not been registered" do
+ header = header_with_params({})
+ RegisteredException = Class.new(StandardError)
+ OtherException = Class.new(StandardError)
+ message = Message.new("somequeue", header, 'foo', :timeout => 10.seconds, :exceptions => 2,
+ :on_exceptions => [RegisteredException], :store => @store)
+ assert !message.attempts_limit_reached?
+ assert !message.exceptions_limit_reached?
+ assert !message.timed_out?
+ assert !message.simple?
+ assert message.exception_accepted? # @exception yet nil, hence 'accepted'
+
+ proc = lambda {|*args| raise OtherException, "crash"}
+ s = sequence("s")
+ message.expects(:completed!).once
+ header.expects(:ack)
+ assert_equal RC::ExceptionNotAccepted, message.__send__(:process_internal, proc)
+ end
+
test "a message should be acked if the handler crashes and the attempts limit has been reached" do
header = header_with_params({})
message = Message.new("somequeue", header, 'foo', :timeout => 10.seconds, :attempts => 2, :store => @store)

on_exceptions option