[Messenger] fix Redis support on 32b arch

Author: nicolas-grekas

Tests/Transport/RedisExt/ConnectionTest.php

@@ -220,6 +220,7 @@ public function testGetAfterReject()
     {
         $redis = new \Redis();
         $connection = Connection::fromDsn('redis://localhost/messenger-rejectthenget', [], $redis);
+        $connection->cleanup();
 
         $connection->add('1', []);
         $connection->add('2', []);
@@ -230,20 +231,38 @@ public function testGetAfterReject()
         $connection = Connection::fromDsn('redis://localhost/messenger-rejectthenget');
         $this->assertNotNull($connection->get());
 
-        $redis->del('messenger-rejectthenget');
+        $connection->cleanup();
     }
 
     public function testGetNonBlocking()
     {
         $redis = new \Redis();
 
         $connection = Connection::fromDsn('redis://localhost/messenger-getnonblocking', [], $redis);
+        $connection->cleanup();
 
         $this->assertNull($connection->get()); // no message, should return null immediately
         $connection->add('1', []);
         $this->assertNotEmpty($message = $connection->get());
         $connection->reject($message['id']);
-        $redis->del('messenger-getnonblocking');
+
+        $connection->cleanup();
+    }
+
+    public function testGetDelayed()
+    {
+        $redis = new \Redis();
+
+        $connection = Connection::fromDsn('redis://localhost/messenger-delayed', [], $redis);
+        $connection->cleanup();
+
+        $connection->add('1', [], 100);
+        $this->assertNull($connection->get());
+        usleep(300000);
+        $this->assertNotEmpty($message = $connection->get());
+        $connection->reject($message['id']);
+
+        $connection->cleanup();
     }
 
     public function testJsonError()

Transport/RedisExt/Connection.php

@@ -141,33 +141,29 @@ public function get(): ?array
         if ($this->autoSetup) {
             $this->setup();
         }
+        $now = microtime();
+        $now = substr($now, 11).substr($now, 2, 3);
 
-        try {
-            $queuedMessageCount = $this->connection->zcount($this->queue, 0, $this->getCurrentTimeInMilliseconds());
-        } catch (\RedisException $e) {
-            throw new TransportException($e->getMessage(), 0, $e);
-        }
+        $queuedMessageCount = $this->rawCommand('ZCOUNT', 0, $now);
 
-        if ($queuedMessageCount) {
-            for ($i = 0; $i < $queuedMessageCount; ++$i) {
-                try {
-                    $queuedMessages = $this->connection->zpopmin($this->queue, 1);
-                } catch (\RedisException $e) {
-                    throw new TransportException($e->getMessage(), 0, $e);
-                }
+        while ($queuedMessageCount--) {
+            if (![$queuedMessage, $expiry] = $this->rawCommand('ZPOPMIN', 1)) {
+                break;
+            }
+
+            if (\strlen($expiry) === \strlen($now) ? $expiry > $now : \strlen($expiry) < \strlen($now)) {
+                // if a future-placed message is popped because of a race condition with
+                // another running consumer, the message is readded to the queue
 
-                foreach ($queuedMessages 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()
-                    );
+                if (!$this->rawCommand('ZADD', 'NX', $expiry, $queuedMessage)) {
+                    throw new TransportException('Could not add a message to the redis stream.');
                 }
+
+                break;
             }
+
+            $queuedMessage = json_decode($queuedMessage, true);
+            $this->add($queuedMessage['body'], $queuedMessage['headers'], 0);
         }
 
         $messageId = '>'; // will receive new messages
@@ -255,7 +251,7 @@ public function add(string $body, array $headers, int $delayInMs = 0): void
         }
 
         try {
-            if ($delayInMs > 0) { // the delay could be smaller 0 in a queued message
+            if ($delayInMs > 0) { // the delay is <= 0 for queued messages
                 $message = json_encode([
                     'body' => $body,
                     'headers' => $headers,
@@ -267,8 +263,18 @@ public function add(string $body, array $headers, int $delayInMs = 0): void
                     throw new TransportException(json_last_error_msg());
                 }
 
-                $score = $this->getCurrentTimeInMilliseconds() + $delayInMs;
-                $added = $this->connection->zadd($this->queue, ['NX'], $score, $message);
+                $now = explode(' ', microtime(), 2);
+                $now[0] = str_pad($delayInMs + substr($now[0], 2, 3), 3, '0', \STR_PAD_LEFT);
+                if (3 < \strlen($now[0])) {
+                    $now[1] += substr($now[0], 0, -3);
+                    $now[0] = substr($now[0], -3);
+
+                    if (\is_float($now[1])) {
+                        throw new TransportException("Message delay is too big: {$delayInMs}ms.");
+                    }
+                }
+
+                $added = $this->rawCommand('ZADD', 'NX', $now[1].$now[0], $message);
             } else {
                 $message = json_encode([
                     'body' => $body,
@@ -316,14 +322,30 @@ public function setup(): void
         $this->autoSetup = false;
     }
 
-    private function getCurrentTimeInMilliseconds(): int
-    {
-        return (int) (microtime(true) * 1000);
-    }
-
     public function cleanup(): void
     {
         $this->connection->del($this->stream);
         $this->connection->del($this->queue);
     }
+
+    /**
+     * @return mixed
+     */
+    private function rawCommand(string $command, ...$arguments)
+    {
+        try {
+            $result = $this->connection->rawCommand($command, $this->queue, ...$arguments);
+        } catch (\RedisException $e) {
+            throw new TransportException($e->getMessage(), 0, $e);
+        }
+
+        if (false === $result) {
+            if ($error = $this->connection->getLastError() ?: null) {
+                $this->connection->clearLastError();
+            }
+            throw new TransportException($error ?? sprintf('Could not run "%s" on Redis queue.', $command));
+        }
+
+        return $result;
+    }
 }