Do not ack/nack if consumer rejects/requeues message

lib/hutch/consumer.rb

@@ -14,12 +14,18 @@ def self.included(base)
 
     def reject!
       broker.reject(delivery_info.delivery_tag)
+      @delivery_rejected = true
     end
 
     def requeue!
       broker.requeue(delivery_info.delivery_tag)
+      @delivery_requeued = true
     end
-
+
+    def already_responded?
+      @delivery_rejected || @delivery_requeued
+    end
+
     def logger
       Hutch::Logging.logger
     end

lib/hutch/worker.rb

@@ -112,9 +112,9 @@ def handle_message(consumer, delivery_info, properties, payload)
         message = Message.new(delivery_info, properties, payload)
         consumer_instance = consumer.new.tap { |c| c.broker, c.delivery_info = @broker, delivery_info }
         with_tracing(consumer_instance).handle(message)
-        broker.ack(delivery_info.delivery_tag)
+        broker.ack(delivery_info.delivery_tag) unless consumer_instance.already_responded?
       rescue StandardError => ex
-        broker.nack(delivery_info.delivery_tag)
+        broker.nack(delivery_info.delivery_tag) unless consumer_instance && consumer_instance.already_responded?
         handle_error(properties.message_id, payload, consumer, ex)
       end
     end

spec/hutch/worker_spec.rb

@@ -49,6 +49,7 @@
     before { allow(broker).to receive(:nack) }
     before { allow(consumer_instance).to receive(:broker=) }
     before { allow(consumer_instance).to receive(:delivery_info=) }
+    before { allow(consumer_instance).to receive(:already_responded?).and_return(false) }
 
     it 'passes the message to the consumer' do
       expect(consumer_instance).to receive(:process).
@@ -62,6 +63,43 @@
       worker.handle_message(consumer, delivery_info, properties, payload)
     end
 
+    context 'when the consumer requeues a message' do
+      class Rejecter
+        include Hutch::Consumer
+
+        def process(message)
+          requeue!
+        end
+      end
+
+      it 'requeues the message' do
+        expect(broker).to_not receive(:ack)
+        expect(broker).to_not receive(:nack)
+        expect(broker).to receive(:requeue)
+
+        worker.handle_message(Rejecter, delivery_info, properties, payload)
+      end
+    end
+
+    context 'when the consumer requeues a message and then crashes' do
+      class RejecterCrash
+        include Hutch::Consumer
+
+        def process(message)
+          requeue!
+          raise "a consumer error"
+        end
+      end
+
+      it 'requeues the message' do
+        expect(broker).to_not receive(:ack)
+        expect(broker).to_not receive(:nack)
+        expect(broker).to receive(:requeue)
+
+        worker.handle_message(RejecterCrash, delivery_info, properties, payload)
+      end
+    end
+
     context 'when the consumer raises an exception' do
       before { allow(consumer_instance).to receive(:process).and_raise('a consumer error') }