Add handling for delayed message to redis transport

Author: alexander-schranz

src/Symfony/Component/Messenger/Tests/Transport/RedisExt/RedisExtIntegrationTest.php

@@ -31,7 +31,7 @@ protected function setUp(): void
 
         $this->redis = new \Redis();
         $this->connection = Connection::fromDsn(getenv('MESSENGER_REDIS_DSN'), [], $this->redis);
-        $this->clearRedis();
+        $this->connection->cleanup();
         $this->connection->setup();
     }
 
@@ -55,11 +55,14 @@ public function testGetTheFirstAvailableMessage()
         $this->assertEquals(['type' => DummyMessage::class], $encoded['headers']);
     }
 
-    private function clearRedis()
+    public function testConnectionSendAndGetDelayed()
     {
-        $parsedUrl = parse_url(getenv('MESSENGER_REDIS_DSN'));
-        $pathParts = explode('/', $parsedUrl['path'] ?? '');
-        $stream = $pathParts[1] ?? 'symfony';
-        $this->redis->del($stream);
+        $this->connection->add('{"message": "Hi"}', ['type' => DummyMessage::class], 1500);
+        $encoded = $this->connection->get();
+        $this->assertEquals(null, $encoded);
+        sleep(2);
+        $encoded = $this->connection->get();
+        $this->assertEquals('{"message": "Hi"}', $encoded['body']);
+        $this->assertEquals(['type' => DummyMessage::class], $encoded['headers']);
     }
 }

src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php

@@ -36,6 +36,7 @@ class Connection
 
     private $connection;
     private $stream;
+    private $queue;
     private $group;
     private $consumer;
     private $autoSetup;
@@ -59,6 +60,7 @@ public function __construct(array $configuration, array $connectionCredentials =
         $this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
         $this->group = $configuration['group'] ?? self::DEFAULT_OPTIONS['group'];
         $this->consumer = $configuration['consumer'] ?? self::DEFAULT_OPTIONS['consumer'];
+        $this->queue = $this->stream.'__queue';
         $this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
         $this->maxEntries = $configuration['stream_max_entries'] ?? self::DEFAULT_OPTIONS['stream_max_entries'];
     }
@@ -112,6 +114,24 @@ public function get(): ?array
             $this->setup();
         }
 
+        $queuedMessageCount = $this->connection->zcount($this->queue, 0, $this->getCurrentTimeInMilliseconds());
+
+        if ($queuedMessageCount) {
+            for ($i = 0; $i < $queuedMessageCount; ++$i) {
+                foreach ($this->connection->zpopmin($this->queue, 1) as $queuedMessage => $time) {
+                    $queuedMessage = json_decode($queuedMessage, true);
+                    // if a futured placed message is actually popped because of a race condition with
+                    // another running message consumer, the message is readded to the queue by add function
+                    // else its just added stream and will be available for all stream consumers
+                    $this->add(
+                        $queuedMessage['body'],
+                        $queuedMessage['headers'],
+                        $time - $this->getCurrentTimeInMilliseconds()
+                    );
+                }
+            }
+        }
+
         $messageId = '>'; // will receive new messages
 
         if ($this->couldHavePendingMessages) {
@@ -191,24 +211,37 @@ public function reject(string $id): void
         }
     }
 
-    public function add(string $body, array $headers): void
+    /**
+     * @param int $delay The delay in milliseconds
+     */
+    public function add(string $body, array $headers, int $delay = 0): void
     {
         if ($this->autoSetup) {
             $this->setup();
         }
 
         try {
-            if ($this->maxEntries) {
-                $added = $this->connection->xadd($this->stream, '*', ['message' => json_encode(
-                    ['body' => $body, 'headers' => $headers]
-                )], $this->maxEntries, true);
+            $message = json_encode([
+                'body' => $body,
+                'headers' => $headers,
+                'uniqid' => uniqid('', true), // Entry need to be unique in the sorted set
+            ]);
+
+            if ($delay > 0) { // the delay could be smaller 0 in a queued message
+                $score = (int) ($this->getCurrentTimeInMilliseconds() + $delay);
+                $added = $this->connection->zadd($this->queue, ['NX'], $score, $message);
             } else {
-                $added = $this->connection->xadd($this->stream, '*', ['message' => json_encode(
-                    ['body' => $body, 'headers' => $headers]
-                )]);
+                if ($this->maxEntries) {
+                    $added = $this->connection->xadd($this->stream, '*', ['message' => $message], $this->maxEntries, true);
+                } else {
+                    $added = $this->connection->xadd($this->stream, '*', ['message' => $message]);
+                }
             }
         } catch (\RedisException $e) {
-            throw new TransportException($e->getMessage(), 0, $e);
+            if ($error = $this->connection->getLastError() ?: null) {
+                $this->connection->clearLastError();
+            }
+            throw new TransportException($error ?? $e->getMessage(), 0, $e);
         }
 
         if (!$added) {
@@ -234,4 +267,18 @@ public function setup(): void
 
         $this->autoSetup = false;
     }
+
+    private function getCurrentTimeInMilliseconds(): int
+    {
+        return (int) (microtime(true) * 1000);
+    }
+
+    /**
+     * @internal
+     */
+    public function cleanup()
+    {
+        $this->connection->del($this->stream);
+        $this->connection->del($this->queue);
+    }
 }

src/Symfony/Component/Messenger/Transport/RedisExt/RedisSender.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger\Transport\RedisExt;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Stamp\DelayStamp;
 use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 
@@ -37,7 +38,11 @@ public function send(Envelope $envelope): Envelope
     {
         $encodedMessage = $this->serializer->encode($envelope);
 
-        $this->connection->add($encodedMessage['body'], $encodedMessage['headers'] ?? []);
+        /** @var DelayStamp|null $delayStamp */
+        $delayStamp = $envelope->last(DelayStamp::class);
+        $delay = null !== $delayStamp ? $delayStamp->getDelay() : 0;
+
+        $this->connection->add($encodedMessage['body'], $encodedMessage['headers'] ?? [], $delay);
 
         return $envelope;
     }