bug #34017 [Messenger] Fix ignored options in redis transport (chalasr)

This PR was merged into the 4.3 branch.

Discussion
----------

[Messenger] Fix ignored options in redis transport

| Q             | A
| ------------- | ---
| Branch?       | 4.3
| Bug fix?      | yes
| New feature?  | no
| Deprecations? | no
| Tickets       | Fix #33425
| License       | MIT
| Doc PR        | -

Also fixes redis authentication failure handling (inline with invalid db index handling, borrowed from symfony/cache).
/cc @alexander-schranz

Commits
-------

c83ff94c37 [Messenger] Fix ignored options in redis transport

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

@@ -57,13 +57,8 @@ class ConnectionTest extends TestCase
     public function testFromDsnWithOptions()
     {
         $this->assertEquals(
-            new Connection(['stream' => 'queue', 'group' => 'group1', 'consumer' => 'consumer1', 'auto_setup' => false], [
+            Connection::fromDsn('redis://localhost', ['stream' => 'queue', 'group' => 'group1', 'consumer' => 'consumer1', 'auto_setup' => false, 'serializer' => 2]),
-                'host' => 'localhost',
+            Connection::fromDsn('redis://localhost/queue/group1/consumer1?serializer=2&auto_setup=0')
-                'port' => 6379,
-            ], [
-                'serializer' => 2,
-            ]),
-            Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['serializer' => 2, 'auto_setup' => false])
         );
     }
 
@@ -99,7 +94,21 @@ class ConnectionTest extends TestCase
         $redis = $this->getMockBuilder(\Redis::class)->disableOriginalConstructor()->getMock();
 
         $redis->expects($this->exactly(1))->method('auth')
-            ->with('password');
+            ->with('password')
+            ->willReturn(true);
+
+        Connection::fromDsn('redis://password@localhost/queue', [], $redis);
+    }
+
+    public function testFailedAuth()
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('Redis connection failed');
+        $redis = $this->getMockBuilder(\Redis::class)->disableOriginalConstructor()->getMock();
+
+        $redis->expects($this->exactly(1))->method('auth')
+            ->with('password')
+            ->willReturn(false);
 
         Connection::fromDsn('redis://password@localhost/queue', [], $redis);
     }

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

@@ -20,6 +20,7 @@ use Symfony\Component\Messenger\Exception\TransportException;
  *
  * @author Alexander Schranz <alexander@sulu.io>
  * @author Antoine Bluchet <soyuka@gmail.com>
+ * @author Robin Chalas <robin.chalas@gmail.com>
  *
  * @internal
  * @final
@@ -52,8 +53,8 @@ class Connection
         $this->connection->connect($connectionCredentials['host'] ?? '127.0.0.1', $connectionCredentials['port'] ?? 6379);
         $this->connection->setOption(\Redis::OPT_SERIALIZER, $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP);
 
-        if (isset($connectionCredentials['auth'])) {
+        if (isset($connectionCredentials['auth']) && !$this->connection->auth($connectionCredentials['auth'])) {
-            $this->connection->auth($connectionCredentials['auth']);
+            throw new InvalidArgumentException(sprintf('Redis connection failed: %s', $redis->getLastError()));
         }
 
         $this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
@@ -70,9 +71,9 @@ class Connection
 
         $pathParts = explode('/', $parsedUrl['path'] ?? '');
 
-        $stream = $pathParts[1] ?? null;
+        $stream = $pathParts[1] ?? $redisOptions['stream'] ?? null;
-        $group = $pathParts[2] ?? null;
+        $group = $pathParts[2] ?? $redisOptions['group'] ?? null;
-        $consumer = $pathParts[3] ?? null;
+        $consumer = $pathParts[3] ?? $redisOptions['consumer'] ?? null;
 
         $connectionCredentials = [
             'host' => $parsedUrl['host'] ?? '127.0.0.1',