[dbal] Sort priority messages by published at date too.

pkg/dbal/DbalConsumer.php

@@ -186,6 +186,7 @@ protected function convertMessage(array $dbalMessage)
         $message->setBody($dbalMessage['body']);
         $message->setPriority((int) $dbalMessage['priority']);
         $message->setRedelivered((bool) $dbalMessage['redelivered']);
+        $message->setPublishedAt((int) $dbalMessage['published_at']);
 
         if ($dbalMessage['headers']) {
             $message->setHeaders(JSON::decode($dbalMessage['headers']));
@@ -213,6 +214,7 @@ private function fetchPrioritizedMessage($now)
             ->andWhere('priority IS NOT NULL')
             ->andWhere('(delayed_until IS NULL OR delayed_until <= :delayedUntil)')
             ->addOrderBy('priority', 'desc')
+            ->addOrderBy('published_at', 'asc')
             ->setMaxResults(1)
         ;
 

pkg/dbal/DbalMessage.php

@@ -41,6 +41,15 @@ class DbalMessage implements PsrMessage
      */
     private $timeToLive;
 
+    /**
+     * Milliseconds, for example 15186054527288.
+     *
+     * Could be generated by the code: (int) (microtime(true) * 10000)
+     *
+     * @var int
+     */
+    private $publishedAt;
+
     /**
      * @param string $body
      * @param array  $properties
@@ -259,7 +268,7 @@ public function getTimestamp()
     {
         $value = $this->getHeader('timestamp');
 
-        return $value === null ? null : (int) $value;
+        return null === $value ? null : (int) $value;
     }
 
     /**
@@ -269,4 +278,20 @@ public function setTimestamp($timestamp)
     {
         $this->setHeader('timestamp', $timestamp);
     }
+
+    /**
+     * @return int
+     */
+    public function getPublishedAt()
+    {
+        return $this->publishedAt;
+    }
+
+    /**
+     * @param int $publishedAt
+     */
+    public function setPublishedAt($publishedAt)
+    {
+        $this->publishedAt = $publishedAt;
+    }
 }

pkg/dbal/DbalProducer.php

@@ -54,7 +54,7 @@ public function send(PsrDestination $destination, PsrMessage $message)
         InvalidDestinationException::assertDestinationInstanceOf($destination, DbalDestination::class);
         InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);
 
-        if (null !== $this->priority && null === $message->getPriority()) {
+        if (null !== $this->priority && 0 === $message->getPriority()) {
             $message->setPriority($this->priority);
         }
         if (null !== $this->deliveryDelay && null === $message->getDeliveryDelay()) {
@@ -81,9 +81,14 @@ public function send(PsrDestination $destination, PsrMessage $message)
             throw new \LogicException('The generated uuid is empty');
         }
 
+        $publishedAt = null !== $message->getPublishedAt() ?
+            $message->getPublishedAt() :
+            (int) (microtime(true) * 10000)
+        ;
+
         $dbalMessage = [
             'id' => $uuid,
-            'published_at' => (int) (microtime(true) * 10000),
+            'published_at' => $publishedAt,
             'body' => $body,
             'headers' => JSON::encode($message->getHeaders()),
             'properties' => JSON::encode($message->getProperties()),

pkg/dbal/Tests/DbalConsumerTest.php

@@ -2,10 +2,6 @@
 
 namespace Enqueue\Dbal\Tests;
 
-use Doctrine\DBAL\Connection;
-use Doctrine\DBAL\Platforms\AbstractPlatform;
-use Doctrine\DBAL\Query\QueryBuilder;
-use Doctrine\DBAL\Statement;
 use Enqueue\Dbal\DbalConsumer;
 use Enqueue\Dbal\DbalContext;
 use Enqueue\Dbal\DbalDestination;
@@ -120,45 +116,13 @@ private function createProducerMock()
         return $this->createMock(DbalProducer::class);
     }
 
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|Connection
-     */
-    private function createConnectionMock()
-    {
-        return $this->createMock(Connection::class);
-    }
-
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|Statement
-     */
-    private function createStatementMock()
-    {
-        return $this->createMock(Statement::class);
-    }
-
     /**
      * @return \PHPUnit_Framework_MockObject_MockObject|DbalContext
      */
     private function createContextMock()
     {
         return $this->createMock(DbalContext::class);
     }
-
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|QueryBuilder
-     */
-    private function createQueryBuilderMock()
-    {
-        return $this->createMock(QueryBuilder::class);
-    }
-
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|AbstractPlatform
-     */
-    private function createPlatformMock()
-    {
-        return $this->createMock(AbstractPlatform::class);
-    }
 }
 
 class InvalidMessage implements PsrMessage

pkg/dbal/Tests/DbalMessageTest.php

@@ -49,6 +49,13 @@ public function testShouldSetCorrelationIdAsHeader()
         $this->assertSame(['correlation_id' => 'theCorrelationId'], $message->getHeaders());
     }
 
+    public function testShouldSetPublishedAtToNullInConstructor()
+    {
+        $message = new DbalMessage();
+
+        $this->assertNull($message->getPublishedAt());
+    }
+
     public function testShouldSetMessageIdAsHeader()
     {
         $message = new DbalMessage();
@@ -72,4 +79,13 @@ public function testShouldSetReplyToAsHeader()
 
         $this->assertSame(['reply_to' => 'theReply'], $message->getHeaders());
     }
+
+    public function testShouldAllowGetPreviouslySetPublishedAtTime()
+    {
+        $message = new DbalMessage();
+
+        $message->setPublishedAt(123);
+
+        $this->assertSame(123, $message->getPublishedAt());
+    }
 }

pkg/dbal/Tests/DbalProducerTest.php

@@ -2,7 +2,6 @@
 
 namespace Enqueue\Dbal\Tests;
 
-use Doctrine\DBAL\Connection;
 use Enqueue\Dbal\DbalContext;
 use Enqueue\Dbal\DbalDestination;
 use Enqueue\Dbal\DbalMessage;
@@ -60,14 +59,6 @@ private function createContextMock()
     {
         return $this->createMock(DbalContext::class);
     }
-
-    /**
-     * @return \PHPUnit_Framework_MockObject_MockObject|Connection
-     */
-    private function createConnectionMock()
-    {
-        return $this->createMock(Connection::class);
-    }
 }
 
 class NotSupportedDestination1 implements PsrDestination

pkg/dbal/Tests/Functional/DbalConsumerTest.php

@@ -0,0 +1,104 @@
+<?php
+
+namespace Enqueue\Dbal\Tests\Functional;
+
+use Enqueue\Dbal\DbalContext;
+use Enqueue\Dbal\DbalMessage;
+use Enqueue\Dbal\Tests\Spec\CreateDbalContextTrait;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * @group functional
+ */
+class DbalConsumerTest extends TestCase
+{
+    use CreateDbalContextTrait;
+
+    /**
+     * @var DbalContext
+     */
+    private $context;
+
+    public function setUp()
+    {
+        $this->context = $this->createDbalContext();
+    }
+
+    protected function tearDown()
+    {
+        if ($this->context) {
+            $this->context->close();
+        }
+
+        parent::tearDown();
+    }
+
+    public function testShouldSetPublishedAtDateToReceivedMessage()
+    {
+        $context = $this->context;
+        $queue = $context->createQueue(__METHOD__);
+
+        $consumer = $context->createConsumer($queue);
+
+        // guard
+        $this->assertNull($consumer->receiveNoWait());
+
+        $time = (int) (microtime(true) * 10000);
+
+        $expectedBody = __CLASS__.$time;
+
+        $producer = $context->createProducer();
+
+        $message = $context->createMessage($expectedBody);
+        $message->setPublishedAt($time);
+        $producer->send($queue, $message);
+
+        $message = $consumer->receive(8000); // 8 sec
+
+        $this->assertInstanceOf(DbalMessage::class, $message);
+        $consumer->acknowledge($message);
+        $this->assertSame($expectedBody, $message->getBody());
+        $this->assertSame($time, $message->getPublishedAt());
+    }
+
+    public function testShouldOrderMessagesWithSamePriorityByPublishedAtDate()
+    {
+        $context = $this->context;
+        $queue = $context->createQueue(__METHOD__);
+
+        $consumer = $context->createConsumer($queue);
+
+        // guard
+        $this->assertNull($consumer->receiveNoWait());
+
+        $time = (int) (microtime(true) * 10000);
+        $olderTime = $time - 10000;
+
+        $expectedPriority5Body = __CLASS__.'_priority5_'.$time;
+        $expectedPriority5BodyOlderTime = __CLASS__.'_priority5_'.$olderTime;
+
+        $producer = $context->createProducer();
+
+        $message = $context->createMessage($expectedPriority5Body);
+        $message->setPriority(5);
+        $message->setPublishedAt($time);
+        $producer->send($queue, $message);
+
+        $message = $context->createMessage($expectedPriority5BodyOlderTime);
+        $message->setPriority(5);
+        $message->setPublishedAt($olderTime);
+        $producer->send($queue, $message);
+
+        $message = $consumer->receive(8000); // 8 sec
+
+        $this->assertInstanceOf(DbalMessage::class, $message);
+        $consumer->acknowledge($message);
+        $this->assertSame($expectedPriority5BodyOlderTime, $message->getBody());
+
+        $message = $consumer->receive(8000); // 8 sec
+
+        $this->assertInstanceOf(DbalMessage::class, $message);
+        $consumer->acknowledge($message);
+        $this->assertSame($expectedPriority5Body, $message->getBody());
+    }
+}

pkg/dbal/Tests/Spec/DbalSendAndReceivePriorityMessagesFromQueueTest.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Dbal\Tests\Spec;
 
+use Enqueue\Dbal\DbalContext;
 use Enqueue\Dbal\DbalMessage;
 use Interop\Queue\PsrContext;
 use Interop\Queue\Spec\SendAndReceivePriorityMessagesFromQueueSpec;
@@ -13,6 +14,15 @@ class DbalSendAndReceivePriorityMessagesFromQueueTest extends SendAndReceivePrio
 {
     use CreateDbalContextTrait;
 
+    private $publishedAt;
+
+    public function setUp()
+    {
+        parent::setUp();
+
+        $this->publishedAt = (int) (microtime(true) * 10000);
+    }
+
     /**
      * @return PsrContext
      */
@@ -24,13 +34,17 @@ protected function createContext()
     /**
      * {@inheritdoc}
      *
+     * @param DbalContext $context
+     *
      * @return DbalMessage
      */
-    protected function createMessage(PsrContext $context, $priority)
+    protected function createMessage(PsrContext $context, $body)
     {
         /** @var DbalMessage $message */
-        $message = $context->createMessage('priority'.$priority);
-        $message->setPriority($priority);
+        $message = parent::createMessage($context, $body);
+
+        // in order to test priorities correctly we have to make sure the messages were sent in the same time.
+        $message->setPublishedAt($this->publishedAt);
 
         return $message;
     }

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.4@dev",
+        "queue-interop/queue-spec": "^0.5.5@dev",
         "symfony/dependency-injection": "^2.8|^3|^4",
         "symfony/config": "^2.8|^3|^4"
     },