Merge branch 'php-enqueue:master' into master

Author: mordilion

.github/workflows/ci.yml

@@ -120,7 +120,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        php: ['7.3', '8.0'] # same as in the container
+        php: ['7.4', '8.0'] # same as in the container
         symfony_version: ['4.4.*', '5.2.*']
         dependencies: ['--prefer-lowest', '--prefer-dist']
         rdkafka_action: ['exclude-group', 'group']
@@ -159,5 +159,7 @@ jobs:
       - run: sed -i 's/525568/16777471/' vendor/kwn/php-rdkafka-stubs/stubs/constants.php
 
       - run: bin/dev -b
+        env:
+          PHP_VERSION: ${{ matrix.php }}
 
       - run: bin/test.sh --${{ matrix.rdkafka_action }}=rdkafka

CHANGELOG.md

@@ -1,5 +1,28 @@
 # Change Log
 
+## [0.10.13](https://github.com/php-enqueue/enqueue-dev/tree/0.10.13) (2021-08-25)
+[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.10.12...0.10.13)
+
+**Merged pull requests:**
+
+- \[SNSQS\] added possibility to send message attributes using snsqs transport [\#1195](https://github.com/php-enqueue/enqueue-dev/pull/1195) ([onatskyy](https://github.com/onatskyy))
+- Add in missing arg [\#1194](https://github.com/php-enqueue/enqueue-dev/pull/1194) ([gdsmith](https://github.com/gdsmith))
+- \#1190 add index on delivery\_id to prevent slow queries [\#1191](https://github.com/php-enqueue/enqueue-dev/pull/1191) ([commercewerft](https://github.com/commercewerft))
+- Add setTopicArn methods to SnsContext and SnsQsContext [\#1189](https://github.com/php-enqueue/enqueue-dev/pull/1189) ([gdsmith](https://github.com/gdsmith))
+
+## [0.10.11](https://github.com/php-enqueue/enqueue-dev/tree/0.10.11) (2021-04-28)
+[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.10.10...0.10.11)
+
+**Merged pull requests:**
+
+- Perform at least once delivery when rejecting with requeue [\#1165](https://github.com/php-enqueue/enqueue-dev/pull/1165) ([dgafka](https://github.com/dgafka))
+- Fix dbal delivery delay to always keep integer value [\#1161](https://github.com/php-enqueue/enqueue-dev/pull/1161) ([dgafka](https://github.com/dgafka))
+- Add SqsConsumer methods to SnsQsConsumer [\#1160](https://github.com/php-enqueue/enqueue-dev/pull/1160) ([gdsmith](https://github.com/gdsmith))
+- add subscription\_interval as config for dbal subscription consumer [\#1159](https://github.com/php-enqueue/enqueue-dev/pull/1159) ([mordilion](https://github.com/mordilion))
+- register worker callback only once, move to constructor [\#1157](https://github.com/php-enqueue/enqueue-dev/pull/1157) ([cturbelin](https://github.com/cturbelin))
+- Try to change doctrine/orm version for supporting 2.8 \(PHP 8 support\). [\#1155](https://github.com/php-enqueue/enqueue-dev/pull/1155) ([GothShoot](https://github.com/GothShoot))
+- sns context - fallback for not breaking BC with 10.10 previous versions [\#1149](https://github.com/php-enqueue/enqueue-dev/pull/1149) ([bafor](https://github.com/bafor))
+
 ## [0.10.10](https://github.com/php-enqueue/enqueue-dev/tree/0.10.10) (2021-03-24)
 [Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.10.9...0.10.10)
 

composer.json

@@ -25,10 +25,10 @@
     "queue-interop/amqp-interop": "^0.8.2",
     "queue-interop/queue-interop": "^0.8.1",
     "bunny/bunny": "^0.4|^0.5",
-    "php-amqplib/php-amqplib": "^2.12.1",
-    "doctrine/dbal": "^2.12",
+    "php-amqplib/php-amqplib": "^3.0",
+    "doctrine/dbal": "^2.12|^3.1",
     "ramsey/uuid": "^3.5|^4",
-    "psr/log": "^1.1",
+    "psr/log": "^1.1 || ^2.0 || ^3.0",
     "psr/container": "^1",
     "makasim/temp-file": "^0.2",
     "google/cloud-pubsub": "^1.4.3",
@@ -42,15 +42,16 @@
     "php-http/client-common": "^2.2.1",
     "richardfullmer/rabbitmq-management-api": "^2.1.1",
     "predis/predis": "^1.1",
-    "thruway/client": "^0.5",
+    "thruway/client": "^0.5.5",
     "thruway/pawl-transport": "^0.5.1",
     "influxdb/influxdb-php": "^1.14",
     "datadog/php-datadogstatsd": "^1.3",
     "guzzlehttp/guzzle": "^7.0.1",
     "php-http/discovery": "^1.13",
     "voryx/thruway-common": "^1.0.1",
-    "react/dns": "^1.0",
-    "react/event-loop": "^1.0"
+    "react/dns": "^1.4",
+    "react/event-loop": "^1.2",
+    "react/promise": "^2.8"
   },
   "require-dev": {
     "ext-pcntl": "*",

docker-compose.yml

@@ -3,11 +3,11 @@ version: '2'
 services:
   dev:
     # when image publishing gets sorted:
-    # image: enqueue/dev:${PHP_VERSION}
+#    image: enqueue/dev:${PHP_VERSION:-7.4}
     build:
       context: docker
       args:
-        PHP_VERSION: "${PHP_VERSION:-7.3}"
+        PHP_VERSION: "${PHP_VERSION:-7.4}"
     depends_on:
       - rabbitmq
       - mysql

docker/Dockerfile

@@ -1,4 +1,4 @@
-ARG PHP_VERSION=7.3
+ARG PHP_VERSION=7.4
 FROM makasim/nginx-php-fpm:${PHP_VERSION}-all-exts
 
 ARG PHP_VERSION

pkg/dbal/DbalConsumer.php

@@ -100,14 +100,14 @@ public function reject(Message $message, bool $requeue = false): void
     {
         InvalidMessageException::assertMessageInstanceOf($message, DbalMessage::class);
 
-        $this->acknowledge($message);
-
         if ($requeue) {
             $message = clone $message;
             $message->setRedelivered(false);
 
             $this->getContext()->createProducer()->send($this->queue, $message);
         }
+
+        $this->acknowledge($message);
     }
 
     protected function getContext(): DbalContext

pkg/dbal/DbalConsumerHelperTrait.php

@@ -6,7 +6,6 @@
 
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Exception\RetryableException;
-use Doctrine\DBAL\Types\Type;
 use Ramsey\Uuid\Uuid;
 
 trait DbalConsumerHelperTrait
@@ -39,7 +38,7 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
             ->addOrderBy('priority', 'asc')
             ->addOrderBy('published_at', 'asc')
             ->setParameter('queues', $queues, Connection::PARAM_STR_ARRAY)
-            ->setParameter('delayedUntil', $now, Type::INTEGER)
+            ->setParameter('delayedUntil', $now, DbalType::INTEGER)
             ->setMaxResults(1);
 
         $update = $this->getConnection()->createQueryBuilder()
@@ -48,8 +47,8 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
             ->set('redeliver_after', ':redeliverAfter')
             ->andWhere('id = :messageId')
             ->andWhere('delivery_id IS NULL')
-            ->setParameter('deliveryId', $deliveryId, Type::GUID)
-            ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
+            ->setParameter('deliveryId', $deliveryId, DbalType::GUID)
+            ->setParameter('redeliverAfter', $now + $redeliveryDelay, DbalType::BIGINT)
         ;
 
         while (microtime(true) < $endAt) {
@@ -60,14 +59,14 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessa
                 }
 
                 $update
-                    ->setParameter('messageId', $result['id'], Type::GUID);
+                    ->setParameter('messageId', $result['id'], DbalType::GUID);
 
                 if ($update->execute()) {
                     $deliveredMessage = $this->getConnection()->createQueryBuilder()
                         ->select('*')
                         ->from($this->getContext()->getTableName())
                         ->andWhere('delivery_id = :deliveryId')
-                        ->setParameter('deliveryId', $deliveryId, Type::GUID)
+                        ->setParameter('deliveryId', $deliveryId, DbalType::GUID)
                         ->setMaxResults(1)
                         ->execute()
                         ->fetch();
@@ -103,9 +102,9 @@ protected function redeliverMessages(): void
             ->set('redelivered', ':redelivered')
             ->andWhere('redeliver_after < :now')
             ->andWhere('delivery_id IS NOT NULL')
-            ->setParameter(':now', time(), Type::BIGINT)
-            ->setParameter('deliveryId', null, Type::GUID)
-            ->setParameter('redelivered', true, Type::BOOLEAN)
+            ->setParameter(':now', time(), DbalType::BIGINT)
+            ->setParameter('deliveryId', null, DbalType::GUID)
+            ->setParameter('redelivered', true, DbalType::BOOLEAN)
         ;
 
         try {
@@ -131,8 +130,8 @@ protected function removeExpiredMessages(): void
             ->andWhere('delivery_id IS NULL')
             ->andWhere('redelivered = :redelivered')
 
-            ->setParameter(':now', time(), Type::BIGINT)
-            ->setParameter('redelivered', false, Type::BOOLEAN)
+            ->setParameter(':now', time(), DbalType::BIGINT)
+            ->setParameter('redelivered', false, DbalType::BOOLEAN)
         ;
 
         try {
@@ -153,7 +152,7 @@ private function deleteMessage(string $deliveryId): void
         $this->getConnection()->delete(
             $this->getContext()->getTableName(),
             ['delivery_id' => $deliveryId],
-            ['delivery_id' => Type::GUID]
+            ['delivery_id' => DbalType::GUID]
         );
     }
 }

pkg/dbal/DbalContext.php

@@ -6,7 +6,6 @@
 
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Schema\Table;
-use Doctrine\DBAL\Types\Type;
 use Interop\Queue\Consumer;
 use Interop\Queue\Context;
 use Interop\Queue\Destination;
@@ -39,26 +38,21 @@ class DbalContext implements Context
      * Callable must return instance of Doctrine\DBAL\Connection once called.
      *
      * @param Connection|callable $connection
-     * @param array               $config
      */
     public function __construct($connection, array $config = [])
     {
         $this->config = array_replace([
             'table_name' => 'enqueue',
             'polling_interval' => null,
-            'subscription_interval' => null,
+            'subscription_polling_interval' => null,
         ], $config);
 
         if ($connection instanceof Connection) {
             $this->connection = $connection;
         } elseif (is_callable($connection)) {
             $this->connectionFactory = $connection;
         } else {
-            throw new \InvalidArgumentException(sprintf(
-                'The connection argument must be either %s or callable that returns %s.',
-                Connection::class,
-                Connection::class
-            ));
+            throw new \InvalidArgumentException(sprintf('The connection argument must be either %s or callable that returns %s.', Connection::class, Connection::class));
         }
     }
 
@@ -136,8 +130,8 @@ public function createSubscriptionConsumer(): SubscriptionConsumer
             $consumer->setRedeliveryDelay($this->config['redelivery_delay']);
         }
 
-        if (isset($this->config['subscription_interval'])) {
-            $consumer->setSubscriptionInterval($this->config['subscription_interval']);
+        if (isset($this->config['subscription_polling_interval'])) {
+            $consumer->setPollingInterval($this->config['subscription_polling_interval']);
         }
 
         return $consumer;
@@ -188,7 +182,7 @@ public function purgeQueue(Queue $queue): void
         $this->getDbalConnection()->delete(
             $this->getTableName(),
             ['queue' => $queue->getQueueName()],
-            ['queue' => Type::STRING]
+            ['queue' => DbalType::STRING]
         );
     }
 
@@ -207,10 +201,7 @@ public function getDbalConnection(): Connection
         if (false == $this->connection) {
             $connection = call_user_func($this->connectionFactory);
             if (false == $connection instanceof Connection) {
-                throw new \LogicException(sprintf(
-                    'The factory must return instance of Doctrine\DBAL\Connection. It returns %s',
-                    is_object($connection) ? get_class($connection) : gettype($connection)
-                ));
+                throw new \LogicException(sprintf('The factory must return instance of Doctrine\DBAL\Connection. It returns %s', is_object($connection) ? get_class($connection) : gettype($connection)));
             }
 
             $this->connection = $connection;
@@ -229,24 +220,25 @@ public function createDataBaseTable(): void
 
         $table = new Table($this->getTableName());
 
-        $table->addColumn('id', Type::GUID, ['length' => 16, 'fixed' => true]);
-        $table->addColumn('published_at', Type::BIGINT);
-        $table->addColumn('body', Type::TEXT, ['notnull' => false]);
-        $table->addColumn('headers', Type::TEXT, ['notnull' => false]);
-        $table->addColumn('properties', Type::TEXT, ['notnull' => false]);
-        $table->addColumn('redelivered', Type::BOOLEAN, ['notnull' => false]);
-        $table->addColumn('queue', Type::STRING);
-        $table->addColumn('priority', Type::SMALLINT, ['notnull' => false]);
-        $table->addColumn('delayed_until', Type::BIGINT, ['notnull' => false]);
-        $table->addColumn('time_to_live', Type::BIGINT, ['notnull' => false]);
-        $table->addColumn('delivery_id', Type::GUID, ['length' => 16, 'fixed' => true, 'notnull' => false]);
-        $table->addColumn('redeliver_after', Type::BIGINT, ['notnull' => false]);
+        $table->addColumn('id', DbalType::GUID, ['length' => 16, 'fixed' => true]);
+        $table->addColumn('published_at', DbalType::BIGINT);
+        $table->addColumn('body', DbalType::TEXT, ['notnull' => false]);
+        $table->addColumn('headers', DbalType::TEXT, ['notnull' => false]);
+        $table->addColumn('properties', DbalType::TEXT, ['notnull' => false]);
+        $table->addColumn('redelivered', DbalType::BOOLEAN, ['notnull' => false]);
+        $table->addColumn('queue', DbalType::STRING);
+        $table->addColumn('priority', DbalType::SMALLINT, ['notnull' => false]);
+        $table->addColumn('delayed_until', DbalType::BIGINT, ['notnull' => false]);
+        $table->addColumn('time_to_live', DbalType::BIGINT, ['notnull' => false]);
+        $table->addColumn('delivery_id', DbalType::GUID, ['length' => 16, 'fixed' => true, 'notnull' => false]);
+        $table->addColumn('redeliver_after', DbalType::BIGINT, ['notnull' => false]);
 
         $table->setPrimaryKey(['id']);
         $table->addIndex(['priority', 'published_at', 'queue', 'delivery_id', 'delayed_until', 'id']);
 
         $table->addIndex(['redeliver_after', 'delivery_id']);
         $table->addIndex(['time_to_live', 'delivery_id']);
+        $table->addIndex(['delivery_id']);
 
         $sm->createTable($table);
     }

pkg/dbal/DbalProducer.php

@@ -4,7 +4,6 @@
 
 namespace Enqueue\Dbal;
 
-use Doctrine\DBAL\Types\Type;
 use Interop\Queue\Destination;
 use Interop\Queue\Exception\Exception;
 use Interop\Queue\Exception\InvalidDestinationException;
@@ -35,9 +34,6 @@ class DbalProducer implements Producer
      */
     private $context;
 
-    /**
-     * @param DbalContext $context
-     */
     public function __construct(DbalContext $context)
     {
         $this->context = $context;
@@ -85,49 +81,43 @@ public function send(Destination $destination, Message $message): void
         $delay = $message->getDeliveryDelay();
         if ($delay) {
             if (!is_int($delay)) {
-                throw new \LogicException(sprintf(
-                    'Delay must be integer but got: "%s"',
-                    is_object($delay) ? get_class($delay) : gettype($delay)
-                ));
+                throw new \LogicException(sprintf('Delay must be integer but got: "%s"', is_object($delay) ? get_class($delay) : gettype($delay)));
             }
 
             if ($delay <= 0) {
                 throw new \LogicException(sprintf('Delay must be positive integer but got: "%s"', $delay));
             }
 
-            $dbalMessage['delayed_until'] = time() + (int) $delay / 1000;
+            $dbalMessage['delayed_until'] = time() + (int) ($delay / 1000);
         }
 
         $timeToLive = $message->getTimeToLive();
         if ($timeToLive) {
             if (!is_int($timeToLive)) {
-                throw new \LogicException(sprintf(
-                    'TimeToLive must be integer but got: "%s"',
-                    is_object($timeToLive) ? get_class($timeToLive) : gettype($timeToLive)
-                ));
+                throw new \LogicException(sprintf('TimeToLive must be integer but got: "%s"', is_object($timeToLive) ? get_class($timeToLive) : gettype($timeToLive)));
             }
 
             if ($timeToLive <= 0) {
                 throw new \LogicException(sprintf('TimeToLive must be positive integer but got: "%s"', $timeToLive));
             }
 
-            $dbalMessage['time_to_live'] = time() + (int) $timeToLive / 1000;
+            $dbalMessage['time_to_live'] = time() + (int) ($timeToLive / 1000);
         }
 
         try {
             $rowsAffected = $this->context->getDbalConnection()->insert($this->context->getTableName(), $dbalMessage, [
-                'id' => Type::GUID,
-                'published_at' => Type::INTEGER,
-                'body' => Type::TEXT,
-                'headers' => Type::TEXT,
-                'properties' => Type::TEXT,
-                'priority' => Type::SMALLINT,
-                'queue' => Type::STRING,
-                'time_to_live' => Type::INTEGER,
-                'delayed_until' => Type::INTEGER,
-                'redelivered' => Type::SMALLINT,
-                'delivery_id' => Type::STRING,
-                'redeliver_after' => Type::BIGINT,
+                'id' => DbalType::GUID,
+                'published_at' => DbalType::INTEGER,
+                'body' => DbalType::TEXT,
+                'headers' => DbalType::TEXT,
+                'properties' => DbalType::TEXT,
+                'priority' => DbalType::SMALLINT,
+                'queue' => DbalType::STRING,
+                'time_to_live' => DbalType::INTEGER,
+                'delayed_until' => DbalType::INTEGER,
+                'redelivered' => DbalType::SMALLINT,
+                'delivery_id' => DbalType::STRING,
+                'redeliver_after' => DbalType::BIGINT,
             ]);
 
             if (1 !== $rowsAffected) {