Merge pull request #291 from php-enqueue/dbal-requeue-fix

[dbal] Fix message re-queuing. Reuse producer for it.

Author: makasim

composer.json

@@ -29,7 +29,7 @@
     "enqueue/async-event-dispatcher": "*@dev",
     "queue-interop/queue-interop": "^0.6@dev",
     "queue-interop/amqp-interop": "^0.7@dev",
-    "queue-interop/queue-spec": "^0.5.1@dev",
+    "queue-interop/queue-spec": "^0.5.4@dev",
 
     "phpunit/phpunit": "^5",
     "doctrine/doctrine-bundle": "~1.2",

pkg/dbal/DbalConsumer.php

@@ -130,33 +130,10 @@ public function reject(PsrMessage $message, $requeue = false)
     {
         InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);
 
-        if (false == $requeue) {
-            return;
-        }
+        if ($requeue) {
+            $this->context->createProducer()->send($this->queue, $message);
 
-        $dbalMessage = [
-            'body' => $message->getBody(),
-            'headers' => JSON::encode($message->getHeaders()),
-            'properties' => JSON::encode($message->getProperties()),
-            'priority' => $message->getPriority(),
-            'queue' => $this->queue->getQueueName(),
-            'redelivered' => true,
-        ];
-
-        $affectedRows = $this->dbal->insert($this->context->getTableName(), $dbalMessage, [
-            'body' => Type::TEXT,
-            'headers' => Type::TEXT,
-            'properties' => Type::TEXT,
-            'priority' => Type::SMALLINT,
-            'queue' => Type::STRING,
-            'redelivered' => Type::BOOLEAN,
-        ]);
-
-        if (1 !== $affectedRows) {
-            throw new \LogicException(sprintf(
-                'Expected record was inserted but it is not. message: "%s"',
-                JSON::encode($dbalMessage)
-            ));
+            return;
         }
     }
 

pkg/dbal/Tests/DbalConsumerTest.php

@@ -10,6 +10,7 @@
 use Enqueue\Dbal\DbalContext;
 use Enqueue\Dbal\DbalDestination;
 use Enqueue\Dbal\DbalMessage;
+use Enqueue\Dbal\DbalProducer;
 use Enqueue\Test\ClassExtensionTrait;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrConsumer;
@@ -67,70 +68,58 @@ public function testRejectShouldThrowIfInstanceOfMessageIsInvalid()
         $consumer->reject(new InvalidMessage());
     }
 
-    public function testRejectShouldInsertNewMessageOnRequeue()
+    public function testShouldDoNothingOnReject()
     {
-        $expectedMessage = [
-            'body' => 'theBody',
-            'headers' => '[]',
-            'properties' => '[]',
-            'priority' => 0,
-            'queue' => 'queue',
-            'redelivered' => true,
-        ];
+        $queue = new DbalDestination('queue');
 
-        $dbal = $this->createConnectionMock();
-        $dbal
-            ->expects($this->once())
-            ->method('insert')
-            ->with('tableName', $this->equalTo($expectedMessage))
-            ->will($this->returnValue(1))
-        ;
+        $message = new DbalMessage();
+        $message->setBody('theBody');
 
         $context = $this->createContextMock();
         $context
-            ->expects($this->once())
-            ->method('getDbalConnection')
-            ->will($this->returnValue($dbal))
-        ;
-        $context
-            ->expects($this->once())
-            ->method('getTableName')
-            ->will($this->returnValue('tableName'))
+            ->expects($this->never())
+            ->method('createProducer')
         ;
 
-        $message = new DbalMessage();
-        $message->setBody('theBody');
+        $consumer = new DbalConsumer($context, $queue);
 
-        $consumer = new DbalConsumer($context, new DbalDestination('queue'));
-        $consumer->reject($message, true);
+        $consumer->reject($message);
     }
 
-    public function testRejectShouldThrowIfMessageWasNotInserted()
+    public function testRejectShouldReSendMessageToSameQueueOnRequeue()
     {
-        $dbal = $this->createConnectionMock();
-        $dbal
+        $queue = new DbalDestination('queue');
+
+        $message = new DbalMessage();
+        $message->setBody('theBody');
+
+        $producerMock = $this->createProducerMock();
+        $producerMock
             ->expects($this->once())
-            ->method('insert')
-            ->willReturn(0)
+            ->method('send')
+            ->with($this->identicalTo($queue), $this->identicalTo($message))
         ;
 
         $context = $this->createContextMock();
         $context
             ->expects($this->once())
-            ->method('getDbalConnection')
-            ->will($this->returnValue($dbal))
+            ->method('createProducer')
+            ->will($this->returnValue($producerMock))
         ;
 
-        $message = new DbalMessage();
-        $message->setBody('theBody');
-
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Expected record was inserted but it is not. message:');
+        $consumer = new DbalConsumer($context, $queue);
 
-        $consumer = new DbalConsumer($context, new DbalDestination('queue'));
         $consumer->reject($message, true);
     }
 
+    /**
+     * @return DbalProducer|\PHPUnit_Framework_MockObject_MockObject
+     */
+    private function createProducerMock()
+    {
+        return $this->createMock(DbalProducer::class);
+    }
+
     /**
      * @return \PHPUnit_Framework_MockObject_MockObject|Connection
      */

pkg/dbal/Tests/Spec/DbalRequeueMessageTest.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Enqueue\Dbal\Tests\Spec;
+
+use Interop\Queue\Spec\RequeueMessageSpec;
+
+/**
+ * @group functional
+ */
+class DbalRequeueMessageTest extends RequeueMessageSpec
+{
+    use CreateDbalContextTrait;
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function createContext()
+    {
+        return $this->createDbalContext();
+    }
+}

pkg/dbal/composer.json

@@ -15,7 +15,7 @@
         "enqueue/test": "^0.8@dev",
         "enqueue/enqueue": "^0.8@dev",
         "enqueue/null": "^0.8@dev",
-        "queue-interop/queue-spec": "^0.5.3@dev",
+        "queue-interop/queue-spec": "^0.5.4@dev",
         "symfony/dependency-injection": "^2.8|^3|^4",
         "symfony/config": "^2.8|^3|^4"
     },