vendor/symfony/redis-messenger/Transport/Connection.php line 169

Open in your IDE?
  1. <?php
  3. /*
  4.  * This file is part of the Symfony package.
  5.  *
  6.  * (c) Fabien Potencier <fabien@symfony.com>
  7.  *
  8.  * For the full copyright and license information, please view the LICENSE
  9.  * file that was distributed with this source code.
  10.  */
  12. namespace Symfony\Component\Messenger\Bridge\Redis\Transport;
  14. use Symfony\Component\Messenger\Exception\InvalidArgumentException;
  15. use Symfony\Component\Messenger\Exception\LogicException;
  16. use Symfony\Component\Messenger\Exception\TransportException;
  18. /**
  19.  * A Redis connection.
  20.  *
  21.  * @author Alexander Schranz <alexander@sulu.io>
  22.  * @author Antoine Bluchet <soyuka@gmail.com>
  23.  * @author Robin Chalas <robin.chalas@gmail.com>
  24.  *
  25.  * @internal
  26.  *
  27.  * @final
  28.  */
  29. class Connection
  30. {
  31.     private const DEFAULT_OPTIONS = [
  32.         'stream' => 'messages',
  33.         'group' => 'symfony',
  34.         'consumer' => 'consumer',
  35.         'auto_setup' => true,
  36.         'delete_after_ack' => false,
  37.         'delete_after_reject' => true,
  38.         'stream_max_entries' => 0, // any value higher than 0 defines an approximate maximum number of stream entries
  39.         'dbindex' => 0,
  40.         'tls' => false,
  41.         'redeliver_timeout' => 3600, // Timeout before redeliver messages still in pending state (seconds)
  42.         'claim_interval' => 60000, // Interval by which pending/abandoned messages should be checked
  43.         'lazy' => false,
  44.         'auth' => null,
  45.         'serializer' => \Redis::SERIALIZER_PHP,
  46.     ];
  48.     private $connection;
  49.     private $stream;
  50.     private $queue;
  51.     private $group;
  52.     private $consumer;
  53.     private $autoSetup;
  54.     private $maxEntries;
  55.     private $redeliverTimeout;
  56.     private $nextClaim = 0.0;
  57.     private $claimInterval;
  58.     private $deleteAfterAck;
  59.     private $deleteAfterReject;
  60.     private $couldHavePendingMessages = true;
  62.     /**
  63.      * @param \Redis|\RedisCluster|null $redis
  64.      */
  65.     public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], $redis = null)
  66.     {
  67.         if (version_compare(phpversion('redis'), '4.3.0', '<')) {
  68.             throw new LogicException('The redis transport requires php-redis 4.3.0 or higher.');
  69.         }
  71.         $host = $connectionCredentials['host'] ?? '127.0.0.1';
  72.         $port = $connectionCredentials['port'] ?? 6379;
  73.         $serializer = $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP;
  74.         $dbIndex = $configuration['dbindex'] ?? self::DEFAULT_OPTIONS['dbindex'];
  75.         $auth = $connectionCredentials['auth'] ?? null;
  76.         if ('' === $auth) {
  77.             $auth = null;
  78.         }
  80.         $lazy = $configuration['lazy'] ?? self::DEFAULT_OPTIONS['lazy'];
  81.         if (\is_array($host) || $redis instanceof \RedisCluster) {
  82.             $hosts = \is_string($host) ? [$host.':'.$port] : $host; // Always ensure we have an array
  83.             $initializer = static function ($redis) use ($hosts, $auth, $serializer) {
  84.                 return self::initializeRedisCluster($redis, $hosts, $auth, $serializer);
  85.             };
  86.             $redis = $lazy ? new RedisClusterProxy($redis, $initializer) : $initializer($redis);
  87.         } else {
  88.             $redis = $redis ?? new \Redis();
  89.             $initializer = static function ($redis) use ($host, $port, $auth, $serializer, $dbIndex) {
  90.                 return self::initializeRedis($redis, $host, $port, $auth, $serializer, $dbIndex);
  91.             };
  92.             $redis = $lazy ? new RedisProxy($redis, $initializer) : $initializer($redis);
  93.         }
  95.         $this->connection = $redis;
  97.         foreach (['stream', 'group', 'consumer'] as $key) {
  98.             if (isset($configuration[$key]) && '' === $configuration[$key]) {
  99.                 throw new InvalidArgumentException(sprintf('"%s" should be configured, got an empty string.', $key));
  100.             }
  101.         }
  103.         $this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
  104.         $this->group = $configuration['group'] ?? self::DEFAULT_OPTIONS['group'];
  105.         $this->consumer = $configuration['consumer'] ?? self::DEFAULT_OPTIONS['consumer'];
  106.         $this->queue = $this->stream.'__queue';
  107.         $this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
  108.         $this->maxEntries = $configuration['stream_max_entries'] ?? self::DEFAULT_OPTIONS['stream_max_entries'];
  109.         $this->deleteAfterAck = $configuration['delete_after_ack'] ?? self::DEFAULT_OPTIONS['delete_after_ack'];
  110.         $this->deleteAfterReject = $configuration['delete_after_reject'] ?? self::DEFAULT_OPTIONS['delete_after_reject'];
  111.         $this->redeliverTimeout = ($configuration['redeliver_timeout'] ?? self::DEFAULT_OPTIONS['redeliver_timeout']) * 1000;
  112.         $this->claimInterval = ($configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval']) / 1000;
  113.     }
  115.     /**
  116.      * @param string|string[]|null $auth
  117.      */
  118.     private static function initializeRedis(\Redis $redis, string $host, int $port, $auth, int $serializer, int $dbIndex): \Redis
  119.     {
  120.         if ($redis->isConnected()) {
  121.             return $redis;
  122.         }
  124.         @$redis->connect($host, $port);
  126.         $error = null;
  127.         set_error_handler(function ($type, $msg) use (&$error) { $error = $msg; });
  129.         try {
  130.             $isConnected = $redis->isConnected();
  131.         } finally {
  132.             restore_error_handler();
  133.         }
  135.         if (!$isConnected) {
  136.             throw new InvalidArgumentException('Redis connection failed: '.(preg_match('/^Redis::p?connect\(\): (.*)/', $error ?? $redis->getLastError() ?? '', $matches) ? \sprintf(' (%s)', $matches[1]) : ''));
  137.         }
  139.         $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);
  141.         if (null !== $auth && !$redis->auth($auth)) {
  142.             throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
  143.         }
  145.         if ($dbIndex && !$redis->select($dbIndex)) {
  146.             throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError());
  147.         }
  149.         return $redis;
  150.     }
  152.     /**
  153.      * @param string|string[]|null $auth
  154.      */
  155.     private static function initializeRedisCluster(?\RedisCluster $redis, array $hosts, $auth, int $serializer): \RedisCluster
  156.     {
  157.         if (null === $redis) {
  158.             $redis = new \RedisCluster(null, $hosts, 0.0, 0.0, false, $auth);
  159.         }
  161.         $redis->setOption(\Redis::OPT_SERIALIZER, $serializer);
  163.         return $redis;
  164.     }
  166.     /**
  167.      * @param \Redis|\RedisCluster|null $redis
  168.      */
  169.     public static function fromDsn(string $dsn, array $redisOptions = [], $redis = null): self
  170.     {
  171.         if (false === strpos($dsn, ',')) {
  172.             $params = self::parseDsn($dsn, $redisOptions);
  173.         } else {
  174.             $dsns = explode(',', $dsn);
  175.             $parsedUrls = array_map(function ($dsn) use (&$redisOptions) {
  176.                 return self::parseDsn($dsn, $redisOptions);
  177.             }, $dsns);
  179.             // Merge all the URLs, the last one overrides the previous ones
  180.             $params = array_merge(...$parsedUrls);
  182.             // Regroup all the hosts in an array interpretable by RedisCluster
  183.             $params['host'] = array_map(function ($parsedUrl) {
  184.                 if (!isset($parsedUrl['host'])) {
  185.                     throw new InvalidArgumentException('Missing host in DSN, it must be defined when using Redis Cluster.');
  186.                 }
  188.                 return $parsedUrl['host'].':'.($parsedUrl['port'] ?? 6379);
  189.             }, $parsedUrls, $dsns);
  190.         }
  192.         self::validateOptions($redisOptions);
  194.         $autoSetup = null;
  195.         if (\array_key_exists('auto_setup', $redisOptions)) {
  196.             $autoSetup = filter_var($redisOptions['auto_setup'], \FILTER_VALIDATE_BOOLEAN);
  197.             unset($redisOptions['auto_setup']);
  198.         }
  200.         $maxEntries = null;
  201.         if (\array_key_exists('stream_max_entries', $redisOptions)) {
  202.             $maxEntries = filter_var($redisOptions['stream_max_entries'], \FILTER_VALIDATE_INT);
  203.             unset($redisOptions['stream_max_entries']);
  204.         }
  206.         $deleteAfterAck = null;
  207.         if (\array_key_exists('delete_after_ack', $redisOptions)) {
  208.             $deleteAfterAck = filter_var($redisOptions['delete_after_ack'], \FILTER_VALIDATE_BOOLEAN);
  209.             unset($redisOptions['delete_after_ack']);
  210.         } else {
  211.             trigger_deprecation('symfony/redis-messenger', '5.4', 'Not setting the "delete_after_ack" boolean option explicitly is deprecated, its default value will change to true in 6.0.');
  212.         }
  214.         $deleteAfterReject = null;
  215.         if (\array_key_exists('delete_after_reject', $redisOptions)) {
  216.             $deleteAfterReject = filter_var($redisOptions['delete_after_reject'], \FILTER_VALIDATE_BOOLEAN);
  217.             unset($redisOptions['delete_after_reject']);
  218.         }
  220.         $dbIndex = null;
  221.         if (\array_key_exists('dbindex', $redisOptions)) {
  222.             $dbIndex = filter_var($redisOptions['dbindex'], \FILTER_VALIDATE_INT);
  223.             unset($redisOptions['dbindex']);
  224.         }
  226.         $tls = 'rediss' === $params['scheme'];
  227.         if (\array_key_exists('tls', $redisOptions)) {
  228.             trigger_deprecation('symfony/redis-messenger', '5.3', 'Providing "tls" parameter is deprecated, use "rediss://" DSN scheme instead');
  229.             $tls = filter_var($redisOptions['tls'], \FILTER_VALIDATE_BOOLEAN);
  230.             unset($redisOptions['tls']);
  231.         }
  233.         $redeliverTimeout = null;
  234.         if (\array_key_exists('redeliver_timeout', $redisOptions)) {
  235.             $redeliverTimeout = filter_var($redisOptions['redeliver_timeout'], \FILTER_VALIDATE_INT);
  236.             unset($redisOptions['redeliver_timeout']);
  237.         }
  239.         $claimInterval = null;
  240.         if (\array_key_exists('claim_interval', $redisOptions)) {
  241.             $claimInterval = filter_var($redisOptions['claim_interval'], \FILTER_VALIDATE_INT);
  242.             unset($redisOptions['claim_interval']);
  243.         }
  245.         $configuration = [
  246.             'stream' => $redisOptions['stream'] ?? null,
  247.             'group' => $redisOptions['group'] ?? null,
  248.             'consumer' => $redisOptions['consumer'] ?? null,
  249.             'lazy' => $redisOptions['lazy'] ?? self::DEFAULT_OPTIONS['lazy'],
  250.             'auto_setup' => $autoSetup,
  251.             'stream_max_entries' => $maxEntries,
  252.             'delete_after_ack' => $deleteAfterAck,
  253.             'delete_after_reject' => $deleteAfterReject,
  254.             'dbindex' => $dbIndex,
  255.             'redeliver_timeout' => $redeliverTimeout,
  256.             'claim_interval' => $claimInterval,
  257.         ];
  259.         if (isset($params['host'])) {
  260.             $user = isset($params['user']) && '' !== $params['user'] ? rawurldecode($params['user']) : null;
  261.             $pass = isset($params['pass']) && '' !== $params['pass'] ? rawurldecode($params['pass']) : null;
  262.             $connectionCredentials = [
  263.                 'host' => $params['host'],
  264.                 'port' => $params['port'] ?? 6379,
  265.                 // See: https://github.com/phpredis/phpredis/#auth
  266.                 'auth' => $redisOptions['auth'] ?? (null !== $pass && null !== $user ? [$user, $pass] : ($pass ?? $user)),
  267.             ];
  269.             $pathParts = explode('/', rtrim($params['path'] ?? '', '/'));
  271.             $configuration['stream'] = $pathParts[1] ?? $configuration['stream'];
  272.             $configuration['group'] = $pathParts[2] ?? $configuration['group'];
  273.             $configuration['consumer'] = $pathParts[3] ?? $configuration['consumer'];
  274.             if ($tls) {
  275.                 $connectionCredentials['host'] = 'tls://'.$connectionCredentials['host'];
  276.             }
  277.         } else {
  278.             $connectionCredentials = [
  279.                 'host' => $params['path'],
  280.                 'port' => 0,
  281.             ];
  282.         }
  284.         return new self($configuration, $connectionCredentials, $redisOptions, $redis);
  285.     }
  287.     private static function parseDsn(string $dsn, array &$redisOptions): array
  288.     {
  289.         $url = $dsn;
  290.         $scheme = 0 === strpos($dsn, 'rediss:') ? 'rediss' : 'redis';
  292.         if (preg_match('#^'.$scheme.':///([^:@])+$#', $dsn)) {
  293.             $url = str_replace($scheme.':', 'file:', $dsn);
  294.         }
  296.         if (false === $params = parse_url($url)) {
  297.             throw new InvalidArgumentException('The given Redis DSN is invalid.');
  298.         }
  299.         if (isset($params['query'])) {
  300.             parse_str($params['query'], $dsnOptions);
  301.             $redisOptions = array_merge($redisOptions, $dsnOptions);
  302.         }
  304.         return $params;
  305.     }
  307.     private static function validateOptions(array $options): void
  308.     {
  309.         $availableOptions = array_keys(self::DEFAULT_OPTIONS);
  311.         if (0 < \count($invalidOptions = array_diff(array_keys($options), $availableOptions))) {
  312.             trigger_deprecation('symfony/messenger', '5.1', 'Invalid option(s) "%s" passed to the Redis Messenger transport. Passing invalid options is deprecated.', implode('", "', $invalidOptions));
  313.         }
  314.     }
  316.     private function claimOldPendingMessages()
  317.     {
  318.         try {
  319.             // This could soon be optimized with https://github.com/antirez/redis/issues/5212 or
  320.             // https://github.com/antirez/redis/issues/6256
  321.             $pendingMessages = $this->connection->xpending($this->stream, $this->group, '-', '+', 1) ?: [];
  322.         } catch (\RedisException $e) {
  323.             throw new TransportException($e->getMessage(), 0, $e);
  324.         }
  326.         $claimableIds = [];
  327.         foreach ($pendingMessages as $pendingMessage) {
  328.             if ($pendingMessage[1] === $this->consumer) {
  329.                 $this->couldHavePendingMessages = true;
  331.                 return;
  332.             }
  334.             if ($pendingMessage[2] >= $this->redeliverTimeout) {
  335.                 $claimableIds[] = $pendingMessage[0];
  336.             }
  337.         }
  339.         if (\count($claimableIds) > 0) {
  340.             try {
  341.                 $this->connection->xclaim(
  342.                     $this->stream,
  343.                     $this->group,
  344.                     $this->consumer,
  345.                     $this->redeliverTimeout,
  346.                     $claimableIds,
  347.                     ['JUSTID']
  348.                 );
  350.                 $this->couldHavePendingMessages = true;
  351.             } catch (\RedisException $e) {
  352.                 throw new TransportException($e->getMessage(), 0, $e);
  353.             }
  354.         }
  356.         $this->nextClaim = microtime(true) + $this->claimInterval;
  357.     }
  359.     public function get(): ?array
  360.     {
  361.         if ($this->autoSetup) {
  362.             $this->setup();
  363.         }
  364.         $now = microtime();
  365.         $now = substr($now, 11).substr($now, 2, 3);
  367.         $queuedMessageCount = $this->rawCommand('ZCOUNT', 0, $now) ?? 0;
  369.         while ($queuedMessageCount--) {
  370.             if (!$message = $this->rawCommand('ZPOPMIN', 1)) {
  371.                 break;
  372.             }
  374.             [$queuedMessage, $expiry] = $message;
  376.             if (\strlen($expiry) === \strlen($now) ? $expiry > $now : \strlen($expiry) < \strlen($now)) {
  377.                 // if a future-placed message is popped because of a race condition with
  378.                 // another running consumer, the message is readded to the queue
  380.                 if (!$this->rawCommand('ZADD', 'NX', $expiry, $queuedMessage)) {
  381.                     throw new TransportException('Could not add a message to the redis stream.');
  382.                 }
  384.                 break;
  385.             }
  387.             $decodedQueuedMessage = json_decode($queuedMessage, true);
  388.             $this->add(\array_key_exists('body', $decodedQueuedMessage) ? $decodedQueuedMessage['body'] : $queuedMessage, $decodedQueuedMessage['headers'] ?? [], 0);
  389.         }
  391.         if (!$this->couldHavePendingMessages && $this->nextClaim <= microtime(true)) {
  392.             $this->claimOldPendingMessages();
  393.         }
  395.         $messageId = '>'; // will receive new messages
  397.         if ($this->couldHavePendingMessages) {
  398.             $messageId = '0'; // will receive consumers pending messages
  399.         }
  401.         try {
  402.             $messages = $this->connection->xreadgroup(
  403.                 $this->group,
  404.                 $this->consumer,
  405.                 [$this->stream => $messageId],
  406.                 1,
  407.                 1
  408.             );
  409.         } catch (\RedisException $e) {
  410.             throw new TransportException($e->getMessage(), 0, $e);
  411.         }
  413.         if (false === $messages) {
  414.             if ($error = $this->connection->getLastError() ?: null) {
  415.                 $this->connection->clearLastError();
  416.             }
  418.             throw new TransportException($error ?? 'Could not read messages from the redis stream.');
  419.         }
  421.         if ($this->couldHavePendingMessages && empty($messages[$this->stream])) {
  422.             $this->couldHavePendingMessages = false;
  424.             // No pending messages so get a new one
  425.             return $this->get();
  426.         }
  428.         foreach ($messages[$this->stream] ?? [] as $key => $message) {
  429.             return [
  430.                 'id' => $key,
  431.                 'data' => $message,
  432.             ];
  433.         }
  435.         return null;
  436.     }
  438.     public function ack(string $id): void
  439.     {
  440.         try {
  441.             $acknowledged = $this->connection->xack($this->stream, $this->group, [$id]);
  442.             if ($this->deleteAfterAck) {
  443.                 $acknowledged = $this->connection->xdel($this->stream, [$id]);
  444.             }
  445.         } catch (\RedisException $e) {
  446.             throw new TransportException($e->getMessage(), 0, $e);
  447.         }
  449.         if (!$acknowledged) {
  450.             if ($error = $this->connection->getLastError() ?: null) {
  451.                 $this->connection->clearLastError();
  452.             }
  453.             throw new TransportException($error ?? sprintf('Could not acknowledge redis message "%s".', $id));
  454.         }
  455.     }
  457.     public function reject(string $id): void
  458.     {
  459.         try {
  460.             $deleted = $this->connection->xack($this->stream, $this->group, [$id]);
  461.             if ($this->deleteAfterReject) {
  462.                 $deleted = $this->connection->xdel($this->stream, [$id]) && $deleted;
  463.             }
  464.         } catch (\RedisException $e) {
  465.             throw new TransportException($e->getMessage(), 0, $e);
  466.         }
  468.         if (!$deleted) {
  469.             if ($error = $this->connection->getLastError() ?: null) {
  470.                 $this->connection->clearLastError();
  471.             }
  472.             throw new TransportException($error ?? sprintf('Could not delete message "%s" from the redis stream.', $id));
  473.         }
  474.     }
  476.     public function add(string $body, array $headers, int $delayInMs = 0): void
  477.     {
  478.         if ($this->autoSetup) {
  479.             $this->setup();
  480.         }
  482.         try {
  483.             if ($delayInMs > 0) { // the delay is <= 0 for queued messages
  484.                 $message = json_encode([
  485.                     'body' => $body,
  486.                     'headers' => $headers,
  487.                     // Entry need to be unique in the sorted set else it would only be added once to the delayed messages queue
  488.                     'uniqid' => uniqid('', true),
  489.                 ]);
  491.                 if (false === $message) {
  492.                     throw new TransportException(json_last_error_msg());
  493.                 }
  495.                 $now = explode(' ', microtime(), 2);
  496.                 $now[0] = str_pad($delayInMs + substr($now[0], 2, 3), 3, '0', \STR_PAD_LEFT);
  497.                 if (3 < \strlen($now[0])) {
  498.                     $now[1] += substr($now[0], 0, -3);
  499.                     $now[0] = substr($now[0], -3);
  501.                     if (\is_float($now[1])) {
  502.                         throw new TransportException("Message delay is too big: {$delayInMs}ms.");
  503.                     }
  504.                 }
  506.                 $added = $this->rawCommand('ZADD', 'NX', $now[1].$now[0], $message);
  507.             } else {
  508.                 $message = json_encode([
  509.                     'body' => $body,
  510.                     'headers' => $headers,
  511.                 ]);
  513.                 if (false === $message) {
  514.                     throw new TransportException(json_last_error_msg());
  515.                 }
  517.                 if ($this->maxEntries) {
  518.                     $added = $this->connection->xadd($this->stream, '*', ['message' => $message], $this->maxEntries, true);
  519.                 } else {
  520.                     $added = $this->connection->xadd($this->stream, '*', ['message' => $message]);
  521.                 }
  522.             }
  523.         } catch (\RedisException $e) {
  524.             if ($error = $this->connection->getLastError() ?: null) {
  525.                 $this->connection->clearLastError();
  526.             }
  527.             throw new TransportException($error ?? $e->getMessage(), 0, $e);
  528.         }
  530.         if (!$added) {
  531.             if ($error = $this->connection->getLastError() ?: null) {
  532.                 $this->connection->clearLastError();
  533.             }
  534.             throw new TransportException($error ?? 'Could not add a message to the redis stream.');
  535.         }
  536.     }
  538.     public function setup(): void
  539.     {
  540.         try {
  541.             $this->connection->xgroup('CREATE', $this->stream, $this->group, 0, true);
  542.         } catch (\RedisException $e) {
  543.             throw new TransportException($e->getMessage(), 0, $e);
  544.         }
  546.         // group might already exist, ignore
  547.         if ($this->connection->getLastError()) {
  548.             $this->connection->clearLastError();
  549.         }
  551.         if ($this->deleteAfterAck || $this->deleteAfterReject) {
  552.             $groups = $this->connection->xinfo('GROUPS', $this->stream);
  553.             if (
  554.                 // support for Redis extension version 5+
  555.                 (\is_array($groups) && 1 < \count($groups))
  556.                 // support for Redis extension version 4.x
  557.                 || (\is_string($groups) && substr_count($groups, '"name"'))
  558.             ) {
  559.                 throw new LogicException(sprintf('More than one group exists for stream "%s", delete_after_ack and delete_after_reject cannot be enabled as it risks deleting messages before all groups could consume them.', $this->stream));
  560.             }
  561.         }
  563.         $this->autoSetup = false;
  564.     }
  566.     private function getCurrentTimeInMilliseconds(): int
  567.     {
  568.         return (int) (microtime(true) * 1000);
  569.     }
  571.     public function cleanup(): void
  572.     {
  573.         static $unlink = true;
  575.         if ($unlink) {
  576.             try {
  577.                 $unlink = false !== $this->connection->unlink($this->stream, $this->queue);
  578.             } catch (\Throwable $e) {
  579.                 $unlink = false;
  580.             }
  581.         }
  583.         if (!$unlink) {
  584.             $this->connection->del($this->stream, $this->queue);
  585.         }
  586.     }
  588.     /**
  589.      * @return mixed
  590.      */
  591.     private function rawCommand(string $command, ...$arguments)
  592.     {
  593.         try {
  594.             if ($this->connection instanceof \RedisCluster || $this->connection instanceof RedisClusterProxy) {
  595.                 $result = $this->connection->rawCommand($this->queue, $command, $this->queue, ...$arguments);
  596.             } else {
  597.                 $result = $this->connection->rawCommand($command, $this->queue, ...$arguments);
  598.             }
  599.         } catch (\RedisException $e) {
  600.             throw new TransportException($e->getMessage(), 0, $e);
  601.         }
  603.         if (false === $result) {
  604.             if ($error = $this->connection->getLastError() ?: null) {
  605.                 $this->connection->clearLastError();
  606.             }
  607.             throw new TransportException($error ?? sprintf('Could not run "%s" on Redis queue.', $command));
  608.         }
  610.         return $result;
  611.     }
  612. }
  614. if (!class_exists(\Symfony\Component\Messenger\Transport\RedisExt\Connection::class, false)) {
  615.     class_alias(Connection::class, \Symfony\Component\Messenger\Transport\RedisExt\Connection::class);
  616. }