Add timeout to RMQ

lib/freddy.rb

@@ -74,10 +74,10 @@ def initialize(connection, logger, max_concurrency)
   #     if id = register(attributes)
   #       handler.success(id: id)
   #     else
-  #       handler.error(message: 'Can not do')
+  #       handler.error(message: 'Cannot do')
   #     end
   #   end
-  def respond_to(destination, &callback)
+  def respond_to(destination, options = {}, &callback)
     @logger.info "Listening for requests on #{destination}"
 
     channel = @connection.create_channel(prefetch: @prefetch_buffer_size)
@@ -89,7 +89,9 @@ def respond_to(destination, &callback)
         thread_pool: Concurrent::FixedThreadPool.new(@prefetch_buffer_size),
         destination: destination,
         channel: channel,
-        handler_adapter_factory: handler_adapter_factory
+        handler_adapter_factory: handler_adapter_factory,
+        timeout_in_seconds: options[:timeout],
+        logger: @logger
       },
       &callback
     )

lib/freddy/consumers/respond_to_consumer.rb

@@ -7,11 +7,13 @@ def self.consume(**attrs, &block)
         new(**attrs).consume(&block)
       end
 
-      def initialize(thread_pool:, destination:, channel:, handler_adapter_factory:)
+      def initialize(thread_pool:, destination:, channel:, handler_adapter_factory:, timeout_in_seconds:, logger:)
         @consume_thread_pool = thread_pool
         @destination = destination
         @channel = channel
         @handler_adapter_factory = handler_adapter_factory
+        @timeout_in_seconds = timeout_in_seconds
+        @logger = logger
       end
 
       def consume
@@ -35,8 +37,12 @@ def consume_from_destination(&block)
 
       def process_message(delivery)
         @consume_thread_pool.post do
-          delivery.in_span do
-            yield(delivery)
+          Timeout.timeout(@timeout_in_seconds) do
+            delivery.in_span do
+              yield(delivery)
+            end
+          rescue Timeout::Error
+            @logger.warn "Timed out while responding to a message from destination #{@destination}"
           end
         ensure
           @channel.acknowledge(delivery.tag, false)

spec/freddy/consumers/respond_to_consumer_spec.rb

@@ -6,7 +6,9 @@
       thread_pool: thread_pool,
       destination: destination,
       channel: channel,
-      handler_adapter_factory: msg_handler_adapter_factory
+      handler_adapter_factory: msg_handler_adapter_factory,
+      timeout_in_seconds: timeout_in_seconds,
+      logger: logger
     )
   end
 
@@ -17,6 +19,7 @@
   let(:msg_handler_adapter) { Freddy::MessageHandlerAdapters::NoOpHandler.new }
   let(:prefetch_buffer_size) { 2 }
   let(:thread_pool) { Concurrent::FixedThreadPool.new(prefetch_buffer_size) }
+  let(:timeout_in_seconds) { nil }
 
   after do
     connection.close
@@ -35,6 +38,27 @@
     end
   end
 
+  context 'when timeout is given' do
+    let(:timeout_in_seconds) { 1 }
+    let(:channel) { connection.create_channel(prefetch: 2) }
+
+    it 'skips the message after timeout' do
+      consumed = false
+      consumer.consume do
+        sleep 2
+        consumed = true
+      end
+
+      deliver_message
+
+      expect(consumed).to eq(false)
+    end
+
+    def deliver_message
+      channel.default_exchange.publish('{}', routing_key: destination)
+    end
+  end
+
   context 'when thread pool is full' do
     let(:prefetch_buffer_size) { 1 }
     let(:msg_count) { prefetch_buffer_size + 1 }