pass current message to handler pre_process method
MODIFY
lib/beetle/handler.rb
lib/beetle/handler.rb
@@ -49,7 +49,7 @@ module Beetle
end
# called before message processing starts
- def pre_process
+ def pre_process(message)
end
# called for message processing if no processor was specfied when the handler instance

MODIFY
lib/beetle/message.rb
lib/beetle/message.rb
@@ -257,11 +257,18 @@ module Beetle
# process this message and do not allow any exception to escape to the caller
def process(handler)
- logger.debug "Beetle: processing message #{msg_id}(#{timestamp})"
result = nil
+ begin
+ # pre_process might set up log routing and it might raise
+ handler.pre_process(self)
+ rescue Exception => @pre_exception
+ Beetle::reraise_expectation_errors!
+ logger.error "Beetle: preprocessing error #{@pre_exception.class}(#{@pre_exception}) for #{msg_id}"
+ end
+ logger.debug "Beetle: processing message #{msg_id}(#{timestamp})"
begin
result = process_internal(handler)
- handler.process_exception(@exception) if @exception
+ handler.process_exception(@exception || @pre_exception) if (@exception || @pre_exception)
handler.process_failure(result) if result.failure?
rescue Exception => e
Beetle::reraise_expectation_errors!
@@ -278,6 +285,9 @@ module Beetle
if @exception
ack!
RC::DecodingError
+ elsif @pre_exception
+ ack!
+ RC::PreprocessingError
elsif expired?
logger.warn "Beetle: ignored expired message (#{msg_id})!"
ack!

MODIFY
lib/beetle/r_c.rb
lib/beetle/r_c.rb
@@ -39,6 +39,7 @@ module Beetle
rc :MutexLocked, :reject
rc :InternalError, :reject
rc :DecodingError, :failure
+ rc :PreprocessingError, :failure
end
end

MODIFY
lib/beetle/subscriber.rb
lib/beetle/subscriber.rb
@@ -175,10 +175,9 @@ module Beetle
return
end
begin
- processor = Handler.create(handler, opts)
- processor.pre_process
message_options = opts.merge(:server => server, :store => @client.deduplication_store)
m = Message.new(amqp_queue_name, header, data, message_options)
+ processor = Handler.create(handler, opts)
result = m.process(processor)
if result.reject?
if @client.config.dead_lettering_enabled?
@@ -203,7 +202,8 @@ module Beetle
logger.debug "Beetle: completed #{msg_id}"
begin
processor.post_process
- rescue Exception
+ rescue Exception => e
+ logger.error "Beetle: post_process error #{e.class}(#{e}) for #{msg_id}"
Beetle::reraise_expectation_errors!
end
end

MODIFY
test/beetle/message_test.rb
test/beetle/message_test.rb
@@ -117,6 +117,7 @@ module Beetle
def setup
@store = DeduplicationStore.new
@store.flushdb
+ @null_handler = Handler.create(lambda {|*args|})
end
test "should be able to extract msg_id from any key" do
@@ -135,7 +136,7 @@ module Beetle
assert !message.expired?
assert !message.redundant?
- message.process(lambda {|*args|})
+ message.process(@null_handler)
@store.keys(message.msg_id).each do |key|
assert !@store.redis.exists(key)
@@ -150,8 +151,8 @@ module Beetle
assert !message.expired?
assert message.redundant?
- message.process(lambda {|*args|})
- message.process(lambda {|*args|})
+ message.process(@null_handler)
+ message.process(@null_handler)
@store.keys(message.msg_id).each do |key|
assert !@store.redis.exists(key)
@@ -166,7 +167,7 @@ module Beetle
assert !message.expired?
assert message.redundant?
- message.process(lambda {|*args|})
+ message.process(@null_handler)
assert @store.exists(message.msg_id, :status)
assert @store.exists(message.msg_id, :expires)
@@ -183,6 +184,7 @@ module Beetle
def setup
@store = DeduplicationStore.new
@store.flushdb
+ @null_handler = Handler.create(lambda {|*args|})
end
test "an expired message should be acked without calling the handler" do
@@ -192,7 +194,7 @@ module Beetle
assert message.expired?
processed = :no
- message.process(lambda {|*args| processed = true})
+ message.process(Handler.create(lambda {|*args| processed = true}))
assert_equal :no, processed
end
@@ -205,7 +207,7 @@ module Beetle
assert message.delayed?
processed = :no
- message.process(lambda {|*args| processed = true})
+ message.process(Handler.create(lambda {|*args| processed = true}))
assert_equal :no, processed
end
@@ -214,7 +216,7 @@ module Beetle
header.expects(:ack)
message = Message.new("somequeue", header, 'foo', :store => @store)
- message.process(lambda {|*args|})
+ message.process(@null_handler)
assert !message.redundant?
assert !@store.exists(message.msg_id, :ack_count)
end
@@ -225,7 +227,7 @@ module Beetle
message.expects(:ack!)
assert message.redundant?
- message.process(lambda {|*args|})
+ message.process(@null_handler)
end
test "acking a redundant message should increment the ack_count key" do
@@ -234,7 +236,7 @@ module Beetle
message = Message.new("somequeue", header, 'foo', :store => @store)
assert_nil @store.get(message.msg_id, :ack_count)
- message.process(lambda {|*args|})
+ message.process(@null_handler)
assert message.redundant?
assert_equal "1", @store.get(message.msg_id, :ack_count)
end
@@ -244,8 +246,8 @@ module Beetle
header.expects(:ack).twice
message = Message.new("somequeue", header, 'foo', :store => @store)
- message.process(lambda {|*args|})
- message.process(lambda {|*args|})
+ message.process(@null_handler)
+ message.process(@null_handler)
assert message.redundant?
assert !@store.exists(message.msg_id, :ack_count)
end
@@ -263,11 +265,12 @@ module Beetle
message = Message.new("somequeue", header, 'foo', :attempts => 2, :store => @store)
assert !message.attempts_limit_reached?
- proc = mock("proc")
+ handler = mock("handler")
s = sequence("s")
- proc.expects(:call).in_sequence(s)
+ handler.expects(:pre_process).with(message).in_sequence(s)
+ handler.expects(:call).in_sequence(s)
header.expects(:ack).in_sequence(s)
- assert_equal RC::OK, message.process(proc)
+ assert_equal RC::OK, message.process(handler)
end
test "after processing a redundant fresh message successfully the ack count should be 1 and the status should be completed" do
@@ -298,25 +301,27 @@ module Beetle
header = header_with_params({})
message = Message.new("somequeue", header, 'foo', :attempts => 1, :store => @store)
- proc = mock("proc")
+ handler = mock("handler")
s = sequence("s")
+ handler.expects(:pre_process).with(message).in_sequence(s)
header.expects(:ack).in_sequence(s)
- proc.expects(:call).in_sequence(s)
- assert_equal RC::OK, message.process(proc)
+ handler.expects(:call).in_sequence(s)
+ assert_equal RC::OK, message.process(handler)
end
test "when processing a simple message, RC::AttemptsLimitReached should be returned if the handler crashes" do
header = header_with_params({})
message = Message.new("somequeue", header, 'foo', :attempts => 1, :store => @store)
- proc = mock("proc")
+ handler = mock("handler")
s = sequence("s")
+ handler.expects(:pre_process).with(message).in_sequence(s)
header.expects(:ack).in_sequence(s)
e = Exception.new("ohoh")
- proc.expects(:call).in_sequence(s).raises(e)
- proc.expects(:process_exception).with(e).in_sequence(s)
- proc.expects(:process_failure).with(RC::AttemptsLimitReached).in_sequence(s)
- assert_equal RC::AttemptsLimitReached, message.process(proc)
+ handler.expects(:call).in_sequence(s).raises(e)
+ handler.expects(:process_exception).with(e).in_sequence(s)
+ handler.expects(:process_failure).with(RC::AttemptsLimitReached).in_sequence(s)
+ assert_equal RC::AttemptsLimitReached, message.process(handler)
end
end
@@ -422,7 +427,6 @@ module Beetle
assert !message.timed_out?
proc = lambda {|*args| raise "crash"}
- s = sequence("s")
message.expects(:completed!).once
header.expects(:ack)
assert_equal RC::AttemptsLimitReached, message.__send__(:process_internal, proc)
@@ -436,7 +440,7 @@ module Beetle
@store.flushdb
end
- test "a message with an exception set should not be processed at all, but it should be acked" do
+ test "a message with a decoding error should not be processed at all, but it should be acked" do
header = {}
message = Message.new("somequeue", header, 'foo')
assert message.exception
@@ -447,6 +451,17 @@ module Beetle
assert_equal RC::DecodingError, message.__send__(:process_internal, proc)
end
+ test "a message with a preprocessing error set should not be processed at all, but it should be acked" do
+ header = header_with_params({})
+ message = Message.new("somequeue", header, 'foo')
+ message.instance_eval { @pre_exception = StandardError.new("shoo") }
+
+ proc = mock("proc")
+ proc.expects(:call).never
+ message.expects(:ack!)
+ assert_equal RC::PreprocessingError, message.__send__(:process_internal, proc)
+ end
+
test "a completed existing message should be just acked and not run the handler" do
header = header_with_params({})
message = Message.new("somequeue", header, 'foo', :attempts => 2, :store => @store)

pass current message to handler pre_process method