Allow to specify that the implementer of ConsumerInterface will handle ACKs

RabbitMq/Consumer.php

@@ -190,7 +190,7 @@ protected function handleProcessMessage(AMQPMessage $msg, $processFlag)
         } else if ($processFlag === ConsumerInterface::MSG_REJECT) {
             // Reject and drop
             $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], false);
-        } else {
+        } else if ($processFlag !== ConsumerInterface::MSG_ACK_SENT) {
             // Remove message from queue only if callback return not false
             $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
         }

RabbitMq/ConsumerInterface.php

@@ -26,6 +26,10 @@ interface ConsumerInterface
      */
     const MSG_REJECT = -1;
 
+    /**
+     * Flag for consumers that wants to handle ACKs on their own
+     */
+    const MSG_ACK_SENT = -2;
 
     /**
      * @param AMQPMessage $msg The message

Tests/RabbitMq/ConsumerTest.php

@@ -38,7 +38,7 @@ protected function prepareAMQPChannel()
      *
      * @dataProvider processMessageProvider
      */
-    public function testProcessMessage($processFlag, $expectedMethod, $expectedRequeue = null)
+    public function testProcessMessage($processFlag, $expectedMethod = null, $expectedRequeue = null)
     {
         $amqpConnection = $this->prepareAMQPConnection();
         $amqpChannel = $this->prepareAMQPChannel();
@@ -52,18 +52,24 @@ public function testProcessMessage($processFlag, $expectedMethod, $expectedReque
         $amqpMessage->delivery_info['channel'] = $amqpChannel;
         $amqpMessage->delivery_info['delivery_tag'] = 0;
 
-        $amqpChannel->expects($this->any())
-            ->method('basic_reject')
-            ->will($this->returnCallback(function($delivery_tag, $requeue) use ($expectedMethod, $expectedRequeue) {
-                \PHPUnit_Framework_Assert::assertSame($expectedMethod, 'basic_reject'); // Check if this function should be called.
-                \PHPUnit_Framework_Assert::assertSame($requeue, $expectedRequeue); // Check if the message should be requeued.
-            }));
-
-        $amqpChannel->expects($this->any())
-            ->method('basic_ack')
-            ->will($this->returnCallback(function($delivery_tag) use ($expectedMethod) {
-                \PHPUnit_Framework_Assert::assertSame($expectedMethod, 'basic_ack'); // Check if this function should be called.
-            }));
+        if ($expectedMethod) {
+            $amqpChannel->expects($this->any())
+                ->method('basic_reject')
+                ->will($this->returnCallback(function ($delivery_tag, $requeue) use ($expectedMethod, $expectedRequeue) {
+                    \PHPUnit_Framework_Assert::assertSame($expectedMethod, 'basic_reject'); // Check if this function should be called.
+                    \PHPUnit_Framework_Assert::assertSame($requeue, $expectedRequeue); // Check if the message should be requeued.
+                }));
+
+            $amqpChannel->expects($this->any())
+                ->method('basic_ack')
+                ->will($this->returnCallback(function ($delivery_tag) use ($expectedMethod) {
+                    \PHPUnit_Framework_Assert::assertSame($expectedMethod, 'basic_ack'); // Check if this function should be called.
+                }));
+        } else {
+            $amqpChannel->expects($this->never())->method('basic_reject');
+            $amqpChannel->expects($this->never())->method('basic_ack');
+            $amqpChannel->expects($this->never())->method('basic_nack');
+        }
         $eventDispatcher = $this->getMockBuilder('Symfony\Component\EventDispatcher\EventDispatcherInterface')
             ->getMock();
         $consumer->setEventDispatcher($eventDispatcher);
@@ -87,6 +93,7 @@ public function processMessageProvider()
             array(ConsumerInterface::MSG_ACK, 'basic_ack'), // Remove message from queue only if callback return not false
             array(ConsumerInterface::MSG_REJECT_REQUEUE, 'basic_reject', true), // Reject and requeue message to RabbitMQ
             array(ConsumerInterface::MSG_REJECT, 'basic_reject', false), // Reject and drop
+            array(ConsumerInterface::MSG_ACK_SENT), // ack not sent by the consumer but should be sent by the implementer of ConsumerInterface
         );
     }