bug #31545 [Messenger] Fix redis Connection::get() should be non blocking by default (chalasr)

fabpot · fabpot · commit 6fb05dc8dc2d · 2019-05-21T06:50:56.000+02:00
This PR was merged into the 4.3 branch. Discussion ---------- [Messenger] Fix redis Connection::get() should be non blocking by default | Q | A | ------------- | --- | Branch? | 4.3 | Bug fix? | yes | New feature? | no | BC breaks? | no | Deprecations? | no | Tests pass? | yes | Fixed tickets | n/a | License | MIT | Doc PR | todo The `\Redis::xreadgroup()` method waits until a message arrives or the specified timeout is reached before returning, which means that `RedisExt\Connection::get()` is blocking. That's inconsistent with other transports which all returns immediately in case there is no message, for instance the AMQP transport uses `\Amqp::get()` instead of `\Amqp::consume()` for this reason. It also short-circuits the worker's stop logic: both the `--time-limit` option of the `messenger:consume` command and the `messenger:stop-workers` don't work with the redis transport. This returns early in case the message count is 0 and no blocking timeout has been configured. Commits ------- 229502a [Messenger] Make redis Connection::get() non blocking by default
diff --git a/src/Symfony/Component/Messenger/Tests/Transport/RedisExt/ConnectionTest.php b/src/Symfony/Component/Messenger/Tests/Transport/RedisExt/ConnectionTest.php
@@ -46,9 +46,9 @@ public function testFromDsnWithOptions()
                 'host' => 'localhost',
                 'port' => 6379,
             ], [
-                'blocking_timeout' => 30,
+                'serializer' => 2,
             ]),
-            Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['blocking_timeout' => 30])
+            Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['serializer' => 2])
         );
     }
 
@@ -59,9 +59,9 @@ public function testFromDsnWithQueryOptions()
                 'host' => 'localhost',
                 'port' => 6379,
             ], [
-                'blocking_timeout' => 30,
+                'serializer' => 2,
             ]),
-            Connection::fromDsn('redis://localhost/queue/group1/consumer1?blocking_timeout=30')
+            Connection::fromDsn('redis://localhost/queue/group1/consumer1?serializer=2')
         );
     }
 
@@ -134,16 +134,20 @@ public function testGetAfterReject()
         $redis->del('messenger-rejectthenget');
     }
 
-    public function testBlockingTimeout()
+    public function testGetNonBlocking()
     {
         $redis = new \Redis();
-        $connection = Connection::fromDsn('redis://localhost/messenger-blockingtimeout', ['blocking_timeout' => 1], $redis);
+
+        $connection = Connection::fromDsn('redis://localhost/messenger-getnonblocking', [], $redis);
         try {
             $connection->setup();
         } catch (TransportException $e) {
         }
 
-        $this->assertNull($connection->get());
-        $redis->del('messenger-blockingtimeout');
+        $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');
     }
 }
diff --git a/src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php b/src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php
@@ -31,7 +31,6 @@ class Connection
     private $stream;
     private $group;
     private $consumer;
-    private $blockingTimeout;
     private $couldHavePendingMessages = true;
 
     public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)
@@ -42,7 +41,6 @@ public function __construct(array $configuration, array $connectionCredentials =
         $this->stream = $configuration['stream'] ?? '' ?: 'messages';
         $this->group = $configuration['group'] ?? '' ?: 'symfony';
         $this->consumer = $configuration['consumer'] ?? '' ?: 'consumer';
-        $this->blockingTimeout = $redisOptions['blocking_timeout'] ?? null;
     }
 
     public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $redis = null): self
@@ -83,8 +81,7 @@ public function get(): ?array
                 $this->group,
                 $this->consumer,
                 [$this->stream => $messageId],
-                1,
-                $this->blockingTimeout
+                1
             );
         } catch (\RedisException $e) {
         }
@@ -142,7 +139,7 @@ public function reject(string $id): void
         }
     }
 
-    public function add(string $body, array $headers)
+    public function add(string $body, array $headers): void
     {
         $e = null;
         try {

Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,6 @@ class Connection`
`31`	`31`	`private $stream;`
`32`	`32`	`private $group;`
`33`	`33`	`private $consumer;`
`34`		`- private $blockingTimeout;`
`35`	`34`	`private $couldHavePendingMessages = true;`
`36`	`35`
`37`	`36`	`public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)`
`@@ -42,7 +41,6 @@ public function __construct(array $configuration, array $connectionCredentials =`
`42`	`41`	`$this->stream = $configuration['stream'] ?? '' ?: 'messages';`
`43`	`42`	`$this->group = $configuration['group'] ?? '' ?: 'symfony';`
`44`	`43`	`$this->consumer = $configuration['consumer'] ?? '' ?: 'consumer';`
`45`		`- $this->blockingTimeout = $redisOptions['blocking_timeout'] ?? null;`
`46`	`44`	`}`
`47`	`45`
`48`	`46`	`public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $redis = null): self`
`@@ -83,8 +81,7 @@ public function get(): ?array`
`83`	`81`	`$this->group,`
`84`	`82`	`$this->consumer,`
`85`	`83`	`[$this->stream => $messageId],`
`86`		`- 1,`
`87`		`- $this->blockingTimeout`
	`84`	`+ 1`
`88`	`85`	`);`
`89`	`86`	`} catch (\RedisException $e) {`
`90`	`87`	`}`
`@@ -142,7 +139,7 @@ public function reject(string $id): void`
`142`	`139`	`}`
`143`	`140`	`}`
`144`	`141`
`145`		`- public function add(string $body, array $headers)`
	`142`	`+ public function add(string $body, array $headers): void`
`146`	`143`	`{`
`147`	`144`	`$e = null;`
`148`	`145`	`try {`