do not use deprecated AMQPMessager->delivery_info

Author: ramunasd

README.md

@@ -922,7 +922,7 @@ class DevckBasicConsumer implements BatchConsumerInterface
         $result = [];
         /** @var AMQPMessage $message */
         foreach ($messages as $message) {
-            $result[(int)$message->delivery_info['delivery_tag']] = $this->executeSomeLogicPerMessage($message);
+            $result[$message->getDeliveryTag()] = $this->executeSomeLogicPerMessage($message);
         }
 
         // you ack only some messages that have return true

RabbitMq/BatchConsumer.php

@@ -300,7 +300,7 @@ private function resetBatch()
     private function addMessage(AMQPMessage $message)
     {
         $this->batchCounter++;
-        $this->messages[(int)$message->delivery_info['delivery_tag']] = $message;
+        $this->messages[$message->getDeliveryTag()] = $message;
     }
 
     /**
@@ -330,7 +330,7 @@ private function getMessageChannel($deliveryTag)
             throw new AMQPRuntimeException(sprintf('Unknown delivery_tag %d!', $deliveryTag));
         }
 
-        return $message->delivery_info['channel'];
+        return $message->getChannel();
     }
 
     /**

RabbitMq/Consumer.php

@@ -197,16 +197,16 @@ protected function handleProcessMessage(AMQPMessage $msg, $processFlag)
     {
         if ($processFlag === ConsumerInterface::MSG_REJECT_REQUEUE || false === $processFlag) {
             // Reject and requeue message to RabbitMQ
-            $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], true);
+            $msg->reject();
         } else if ($processFlag === ConsumerInterface::MSG_SINGLE_NACK_REQUEUE) {
             // NACK and requeue message to RabbitMQ
-            $msg->delivery_info['channel']->basic_nack($msg->delivery_info['delivery_tag'], false, true);
+            $msg->nack(true);
         } else if ($processFlag === ConsumerInterface::MSG_REJECT) {
             // Reject and drop
-            $msg->delivery_info['channel']->basic_reject($msg->delivery_info['delivery_tag'], false);
+            $msg->reject(false);
         } 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']);
+            $msg->ack();
         }
 
         $this->consumed++;

RabbitMq/RpcServer.php

@@ -17,7 +17,7 @@ public function initServer($name)
     public function processMessage(AMQPMessage $msg)
     {
         try {
-            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
+            $msg->ack();
             $result = call_user_func($this->callback, $msg);
             $result = call_user_func($this->serializer, $result);
             $this->sendReply($result, $msg->get('reply_to'), $msg->get('correlation_id'));

Tests/RabbitMq/ConsumerTest.php

@@ -56,8 +56,8 @@ public function testProcessMessage($processFlag, $expectedMethod = null, $expect
 
         // Create a default message
         $amqpMessage = new AMQPMessage('foo body');
-        $amqpMessage->delivery_info['channel'] = $amqpChannel;
-        $amqpMessage->delivery_info['delivery_tag'] = 0;
+        $amqpMessage->setChannel($amqpChannel);
+        $amqpMessage->setDeliveryTag(0);
 
         if ($expectedMethod) {
             $amqpChannel->expects($this->any())

Tests/RabbitMq/MultipleConsumerTest.php

@@ -65,13 +65,8 @@ public function testProcessMessage($processFlag, $expectedMethod, $expectedReque
 
         $this->prepareAMQPChannelExpectations($expectedMethod, $expectedRequeue);
 
-        // Create a default message
-        $amqpMessage = new AMQPMessage('foo body');
-        $amqpMessage->delivery_info['channel'] = $this->amqpChannel;
-        $amqpMessage->delivery_info['delivery_tag'] = 0;
-
-        $this->multipleConsumer->processQueueMessage('test-1', $amqpMessage);
-        $this->multipleConsumer->processQueueMessage('test-2', $amqpMessage);
+        $this->multipleConsumer->processQueueMessage('test-1', $this->createMessage());
+        $this->multipleConsumer->processQueueMessage('test-2', $this->createMessage());
     }
 
     /**
@@ -106,13 +101,8 @@ public function testQueuesProvider($processFlag, $expectedMethod, $expectedReque
 
         $this->prepareAMQPChannelExpectations($expectedMethod, $expectedRequeue);
 
-        // Create a default message
-        $amqpMessage = new AMQPMessage('foo body');
-        $amqpMessage->delivery_info['channel'] = $this->amqpChannel;
-        $amqpMessage->delivery_info['delivery_tag'] = 0;
-
-        $this->multipleConsumer->processQueueMessage('test-1', $amqpMessage);
-        $this->multipleConsumer->processQueueMessage('test-2', $amqpMessage);
+        $this->multipleConsumer->processQueueMessage('test-1', $this->createMessage());
+        $this->multipleConsumer->processQueueMessage('test-2', $this->createMessage());
     }
 
     /**
@@ -154,15 +144,10 @@ public function testQueuesProviderAndStaticQueuesTogether($processFlag, $expecte
 
         $this->prepareAMQPChannelExpectations($expectedMethod, $expectedRequeue);
 
-        // Create a default message
-        $amqpMessage = new AMQPMessage('foo body');
-        $amqpMessage->delivery_info['channel'] = $this->amqpChannel;
-        $amqpMessage->delivery_info['delivery_tag'] = 0;
-
-        $this->multipleConsumer->processQueueMessage('test-1', $amqpMessage);
-        $this->multipleConsumer->processQueueMessage('test-2', $amqpMessage);
-        $this->multipleConsumer->processQueueMessage('test-3', $amqpMessage);
-        $this->multipleConsumer->processQueueMessage('test-4', $amqpMessage);
+        $this->multipleConsumer->processQueueMessage('test-1', $this->createMessage());
+        $this->multipleConsumer->processQueueMessage('test-2', $this->createMessage());
+        $this->multipleConsumer->processQueueMessage('test-3', $this->createMessage());
+        $this->multipleConsumer->processQueueMessage('test-4', $this->createMessage());
     }
 
     public function processMessageProvider()
@@ -300,4 +285,13 @@ private function prepareCallback($processFlag)
             return $processFlag;
         };
     }
+
+    private function createMessage()
+    {
+        $amqpMessage = new AMQPMessage('foo body');
+        $amqpMessage->setChannel($this->amqpChannel);
+        $amqpMessage->setDeliveryTag(0);
+
+        return $amqpMessage;
+    }
 }

Tests/RabbitMq/RpcServerTest.php

@@ -18,14 +18,14 @@ public function testProcessMessageWithCustomSerializer()
         $message = $this->getMockBuilder('\PhpAmqpLib\Message\AMQPMessage')
             ->setMethods( array('get'))
             ->getMock();
-        $message->delivery_info = array(
-            'channel' => $this->getMockBuilder('\PhpAmqpLib\Channel\AMQPChannel')
+        $message->setChannel(
+            $this->getMockBuilder('\PhpAmqpLib\Channel\AMQPChannel')
                 ->setMethods(array())->setConstructorArgs(array())
                 ->setMockClassName('')
                 ->disableOriginalConstructor()
-                ->getMock(),
-            'delivery_tag' => null
+                ->getMock()
         );
+        $message->setDeliveryTag(0);
         $server->setCallback(function() {
             return 'message';
         });