added pre_process and post_process hooks on Beetle::Handler
MODIFY
lib/beetle/handler.rb
lib/beetle/handler.rb
@@ -48,12 +48,20 @@ module Beetle
@__handler_called__ = true
end
+ # called before message processing starts
+ def pre_process
+ end
+
# called for message processing if no processor was specfied when the handler instance
# was created
def process
logger.info "Beetle: received message #{message.inspect}"
end
+ # called after message processing finishes
+ def post_process
+ end
+
# should not be overriden in subclasses
def process_exception(exception) #:nodoc:
if @error_callback

MODIFY
lib/beetle/message.rb
lib/beetle/message.rb
@@ -264,7 +264,7 @@ 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}"
+ logger.debug "Beetle: processing message #{msg_id}(#{timestamp})"
result = nil
begin
result = process_internal(handler)

MODIFY
lib/beetle/subscriber.rb
lib/beetle/subscriber.rb
@@ -168,14 +168,15 @@ module Beetle
def create_subscription_callback(queue_name, amqp_queue_name, handler, opts)
server = @server
lambda do |header, data|
- msg_id, timestamp = header.attributes.values_at(:message_id, :timestamp)
+ msg_id = header.attributes[:message_id]
if channel(server).closing?
- logger.info "Beetle: ignored message since channel to server #{server} was already closed: #{msg_id}(#{timestamp})"
+ timestamp = header.attributes[:timestamp]
+ logger.info "Beetle: ignored message #{msg_id}(#{timestamp}) since channel to server #{server} was already closed"
return
end
begin
- logger.debug "Beetle: received #{msg_id}(#{timestamp})"
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)
result = m.process(processor)
@@ -195,12 +196,16 @@ module Beetle
end
rescue Exception
Beetle::reraise_expectation_errors!
- # swallow all exceptions
logger.debug "Beetle: internal error on #{msg_id}: #{$!}: #{$!.backtrace.join("\n")}"
ensure
# processing_completed swallows all exceptions, so we don't need to protect this call
processor.processing_completed
logger.debug "Beetle: completed #{msg_id}"
+ begin
+ processor.post_process
+ rescue Execption
+ Beetle::reraise_expectation_errors!
+ end
end
end
end

added pre_process and post_process hooks on Beetle::Handler