Moved common logic to trait

rosamarsky · rosamarsky · commit 10f507d0e26c · 2018-10-30T16:26:01.000+02:00
diff --git a/pkg/dbal/DbalConsumer.php b/pkg/dbal/DbalConsumer.php
@@ -15,7 +15,8 @@
 
 class DbalConsumer implements Consumer
 {
-    use ConsumerPollingTrait;
+    use ConsumerPollingTrait,
+        DbalConsumerHelperTrait;
 
     /**
      * @var DbalContext
@@ -48,6 +49,16 @@ public function __construct(DbalContext $context, DbalDestination $queue)
         $this->redeliveryDelay = 1200000;
     }
 
+    public function getContext(): DbalContext
+    {
+        return $this->context;
+    }
+
+    public function getConnection(): Connection
+    {
+        return $this->dbal;
+    }
+
     /**
      * Get interval between retry failed messages in milliseconds.
      */
@@ -78,47 +89,27 @@ public function receiveNoWait(): ?Message
     {
         $this->redeliverMessages();
 
-        $this->dbal->beginTransaction();
+        $this->getConnection()->beginTransaction();
         try {
-            $now = (int) time();
-            $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
             $deliveryId = (string) Uuid::uuid1();
 
             // get top message from the queue
-            $message = $this->fetchMessage($now);
+            $message = $this->fetchMessage([$this->queue->getQueueName()]);
 
             if (null == $message) {
-                $this->dbal->commit();
+                $this->getConnection()->commit();
 
                 return null;
             }
 
-            // mark message as delivered to consumer
-            $this->dbal->createQueryBuilder()
-                ->update($this->context->getTableName())
-                ->set('delivery_id', ':deliveryId')
-                ->set('redeliver_after', ':redeliverAfter')
-                ->andWhere('id = :id')
-                ->setParameter('id', $message['id'], Type::GUID)
-                ->setParameter('deliveryId', $deliveryId, Type::STRING)
-                ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
-                ->execute()
-            ;
-
-            $dbalMessage = $this->dbal->createQueryBuilder()
-                ->select('*')
-                ->from($this->context->getTableName())
-                ->andWhere('delivery_id = :deliveryId')
-                ->setParameter('deliveryId', $deliveryId, Type::STRING)
-                ->setMaxResults(1)
-                ->execute()
-                ->fetch()
-            ;
-
-            $this->dbal->commit();
+            $this->markMessageAsDeliveredToConsumer($message, $deliveryId);
+
+            $dbalMessage = $this->getMessageByDeliveryId($deliveryId);
+
+            $this->getConnection()->commit();
 
             if ($dbalMessage['redelivered'] || empty($dbalMessage['time_to_live']) || $dbalMessage['time_to_live'] > time()) {
-                return $this->context->convertMessage($dbalMessage);
+                return $this->getContext()->convertMessage($dbalMessage);
             }
 
             return null;
@@ -145,7 +136,7 @@ public function reject(Message $message, bool $requeue = false): void
         InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);
 
         if ($requeue) {
-            $this->context->createProducer()->send($this->queue, $message);
+            $this->getContext()->createProducer()->send($this->queue, $message);
 
             return;
         }
@@ -155,37 +146,17 @@ public function reject(Message $message, bool $requeue = false): void
 
     private function deleteMessage(?string $deliveryId): void
     {
-        $this->dbal->delete(
-            $this->context->getTableName(),
+        $this->getConnection()->delete(
+            $this->getContext()->getTableName(),
             ['delivery_id' => $deliveryId],
             ['delivery_id' => Type::STRING]
         );
     }
 
-    private function fetchMessage(int $now): ?array
-    {
-        $result = $this->dbal->createQueryBuilder()
-            ->select('*')
-            ->from($this->context->getTableName())
-            ->andWhere('delivery_id IS NULL')
-            ->andWhere('queue = :queue')
-            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
-            ->addOrderBy('priority', 'desc')
-            ->addOrderBy('published_at', 'asc')
-            ->setParameter('queue', $this->queue->getQueueName(), Type::STRING)
-            ->setParameter('delayedUntil', $now, Type::BIGINT)
-            ->setMaxResults(1)
-            ->execute()
-            ->fetch()
-        ;
-
-        return $result ?: null;
-    }
-
     private function redeliverMessages(): void
     {
-        $this->dbal->createQueryBuilder()
-            ->update($this->context->getTableName())
+        $this->getConnection()->createQueryBuilder()
+            ->update($this->getContext()->getTableName())
             ->set('delivery_id', ':deliveryId')
             ->set('redelivered', ':redelivered')
             ->andWhere('delivery_id IS NOT NULL')
diff --git a/pkg/dbal/DbalConsumerHelperTrait.php b/pkg/dbal/DbalConsumerHelperTrait.php
@@ -0,0 +1,67 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Dbal;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Types\Type;
+
+trait DbalConsumerHelperTrait
+{
+    abstract public function getContext(): DbalContext;
+
+    abstract public function getConnection(): Connection;
+
+    protected function fetchMessage(array $queues): ?array
+    {
+        $now = time();
+
+        $result = $this->getConnection()->createQueryBuilder()
+            ->select('*')
+            ->from($this->getContext()->getTableName())
+            ->andWhere('delivery_id IS NULL')
+            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
+            ->andWhere('queue IN (:queues)')
+            ->addOrderBy('priority', 'desc')
+            ->addOrderBy('published_at', 'asc')
+            ->setParameter('delayedUntil', $now, \Doctrine\DBAL\ParameterType::INTEGER)
+            ->setParameter('queues', array_values($queues), \Doctrine\DBAL\Connection::PARAM_STR_ARRAY)
+            ->setMaxResults(1)
+            ->execute()
+            ->fetch()
+        ;
+
+        return $result ?: null;
+    }
+
+    protected function markMessageAsDeliveredToConsumer(array $message, string $deliveryId): void
+    {
+        $now = time();
+        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
+
+        $this->getConnection()->createQueryBuilder()
+            ->andWhere('id = :id')
+            ->update($this->getContext()->getTableName())
+            ->set('delivery_id', ':deliveryId')
+            ->set('redeliver_after', ':redeliverAfter')
+            ->setParameter('id', $message['id'], Type::GUID)
+            ->setParameter('deliveryId', $deliveryId, Type::STRING)
+            ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
+            ->execute()
+        ;
+    }
+
+    protected function getMessageByDeliveryId(string $deliveryId): array
+    {
+        return $this->getConnection()->createQueryBuilder()
+            ->select('*')
+            ->from($this->getContext()->getTableName())
+            ->andWhere('delivery_id = :deliveryId')
+            ->setParameter('deliveryId', $deliveryId, Type::STRING)
+            ->setMaxResults(1)
+            ->execute()
+            ->fetch()
+        ;
+    }
+}
diff --git a/pkg/dbal/DbalContext.php b/pkg/dbal/DbalContext.php
@@ -129,7 +129,13 @@ public function close(): void
 
     public function createSubscriptionConsumer(): SubscriptionConsumer
     {
-        return new DbalSubscriptionConsumer($this);
+        $consumer = new DbalSubscriptionConsumer($this);
+
+        if (isset($this->config['redelivery_delay'])) {
+            $consumer->setRedeliveryDelay($this->config['redelivery_delay']);
+        }
+
+        return $consumer;
     }
 
     /**
diff --git a/pkg/dbal/DbalSubscriptionConsumer.php b/pkg/dbal/DbalSubscriptionConsumer.php
@@ -4,13 +4,15 @@
 
 namespace Enqueue\Dbal;
 
-use Doctrine\DBAL\Types\Type;
+use Doctrine\DBAL\Connection;
 use Interop\Queue\Consumer;
 use Interop\Queue\SubscriptionConsumer;
 use Ramsey\Uuid\Uuid;
 
 class DbalSubscriptionConsumer implements SubscriptionConsumer
 {
+    use DbalConsumerHelperTrait;
+
     /**
      * @var DbalContext
      */
@@ -47,6 +49,16 @@ public function __construct(DbalContext $context)
         $this->redeliveryDelay = 1200000;
     }
 
+    public function getContext(): DbalContext
+    {
+        return $this->context;
+    }
+
+    public function getConnection(): Connection
+    {
+        return $this->dbal;
+    }
+
     /**
      * Get interval between retry failed messages in milliseconds.
      */
@@ -55,6 +67,11 @@ public function getRedeliveryDelay(): int
         return $this->redeliveryDelay;
     }
 
+    public function setRedeliveryDelay(int $redeliveryDelay): void
+    {
+        $this->redeliveryDelay = $redeliveryDelay;
+    }
+
     public function consume(int $timeout = 0): void
     {
         if (empty($this->subscribers)) {
@@ -63,7 +80,6 @@ public function consume(int $timeout = 0): void
 
         $now = time();
         $timeout /= 1000;
-        $redeliveryDelay = $this->getRedeliveryDelay() / 1000; // milliseconds to seconds
         $deliveryId = (string) Uuid::uuid1();
 
         $queueNames = [];
@@ -80,29 +96,11 @@ public function consume(int $timeout = 0): void
             $message = $this->fetchMessage($currentQueueNames);
 
             if ($message) {
-                // mark message as delivered to consumer
-                $this->dbal->createQueryBuilder()
-                    ->update($this->context->getTableName())
-                    ->set('delivery_id', ':deliveryId')
-                    ->set('redeliver_after', ':redeliverAfter')
-                    ->andWhere('id = :id')
-                    ->setParameter('id', $message['id'], Type::GUID)
-                    ->setParameter('deliveryId', $deliveryId, Type::STRING)
-                    ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
-                    ->execute()
-                ;
-
-                $message = $this->dbal->createQueryBuilder()
-                    ->select('*')
-                    ->from($this->context->getTableName())
-                    ->andWhere('delivery_id = :deliveryId')
-                    ->setParameter('deliveryId', $deliveryId, Type::STRING)
-                    ->setMaxResults(1)
-                    ->execute()
-                    ->fetch()
-                ;
-
-                $dbalMessage = $this->context->convertMessage($message);
+                $this->markMessageAsDeliveredToConsumer($message, $deliveryId);
+
+                $message = $this->getMessageByDeliveryId($deliveryId);
+
+                $dbalMessage = $this->getContext()->convertMessage($message);
 
                 /**
                  * @var DbalConsumer
@@ -174,24 +172,4 @@ public function unsubscribeAll(): void
     {
         $this->subscribers = [];
     }
-
-    private function fetchMessage(array $queues): ?array
-    {
-        $result = $this->dbal->createQueryBuilder()
-            ->select('*')
-            ->from($this->context->getTableName())
-            ->andWhere('delivery_id IS NULL')
-            ->andWhere('queue IN (:queues)')
-            ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil')
-            ->addOrderBy('priority', 'desc')
-            ->addOrderBy('published_at', 'asc')
-            ->setParameter('queues', array_keys($queues), \Doctrine\DBAL\Connection::PARAM_STR_ARRAY)
-            ->setParameter('delayedUntil', time(), \Doctrine\DBAL\ParameterType::INTEGER)
-            ->setMaxResults(1)
-            ->execute()
-            ->fetch()
-        ;
-
-        return $result ?: null;
-    }
 }

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,13 @@ public function close(): void`
`129`	`129`
`130`	`130`	`public function createSubscriptionConsumer(): SubscriptionConsumer`
`131`	`131`	`{`
`132`		`- return new DbalSubscriptionConsumer($this);`
	`132`	`+ $consumer = new DbalSubscriptionConsumer($this);`
	`133`	`+`
	`134`	`+ if (isset($this->config['redelivery_delay'])) {`
	`135`	`+ $consumer->setRedeliveryDelay($this->config['redelivery_delay']);`
	`136`	`+ }`
	`137`	`+`
	`138`	`+ return $consumer;`
`133`	`139`	`}`
`134`	`140`
`135`	`141`	`/**`