Add timeout to RMQ

ENG-227

lib/freddy.rb

@@ -77,7 +77,7 @@ def initialize(connection, logger, max_concurrency)
   #       handler.error(message: 'Can not do')
   #     end
   #   end
-  def respond_to(destination, &callback)
+  def respond_to(destination, &callback, timeout=30)
     @logger.info "Listening for requests on #{destination}"
 
     channel = @connection.create_channel(prefetch: @prefetch_buffer_size)
@@ -91,6 +91,7 @@ def respond_to(destination, &callback)
         channel: channel,
         handler_adapter_factory: handler_adapter_factory
       },
+      timeout,
       &callback
     )
   end

lib/freddy/consumers/respond_to_consumer.rb

@@ -3,8 +3,8 @@
 class Freddy
   module Consumers
     class RespondToConsumer
-      def self.consume(**attrs, &block)
-        new(**attrs).consume(&block)
+      def self.consume(**attrs, timeout, &block)
+        new(**attrs).consume(timeout, &block)
       end
 
       def initialize(thread_pool:, destination:, channel:, handler_adapter_factory:)
@@ -14,8 +14,8 @@ def initialize(thread_pool:, destination:, channel:, handler_adapter_factory:)
         @handler_adapter_factory = handler_adapter_factory
       end
 
-      def consume
-        consumer = consume_from_destination do |delivery|
+      def consume(timeout)
+        consumer = consume_from_destination(timeout) do |delivery|
           adapter = @handler_adapter_factory.for(delivery)
 
           msg_handler = MessageHandler.new(adapter, delivery)
@@ -27,16 +27,20 @@ def consume
 
       private
 
-      def consume_from_destination(&block)
+      def consume_from_destination(timeout, &block)
         @channel.queue(@destination).subscribe(manual_ack: true) do |delivery|
-          process_message(delivery, &block)
+          process_message(delivery, timeout, &block)
         end
       end
 
-      def process_message(delivery)
+      def process_message(delivery, timeout)
         @consume_thread_pool.post do
-          delivery.in_span do
-            yield(delivery)
+          Timeout.timeout(timeout) do
+            delivery.in_span do
+              yield(delivery)
+            end
+          rescue Timeout::Error
+            # log a warning here maybe?
           end
         ensure
           @channel.acknowledge(delivery.tag, false)