* * For the full copyright and license information, please view the LICENSE * file that was distributed with this source code. */ namespace Symfony\Component\Messenger\Bridge\Redis\Transport; use Symfony\Component\Messenger\Exception\InvalidArgumentException; use Symfony\Component\Messenger\Exception\LogicException; use Symfony\Component\Messenger\Exception\TransportException; /** * A Redis connection. * * @author Alexander Schranz * @author Antoine Bluchet * @author Robin Chalas * * @internal * @final */ class Connection { private const DEFAULT_OPTIONS = [ 'stream' => 'messages', 'group' => 'symfony', 'consumer' => 'consumer', 'auto_setup' => true, 'delete_after_ack' => false, 'delete_after_reject' => true, 'stream_max_entries' => 0, // any value higher than 0 defines an approximate maximum number of stream entries 'dbindex' => 0, 'tls' => false, 'redeliver_timeout' => 3600, // Timeout before redeliver messages still in pending state (seconds) 'claim_interval' => 60000, // Interval by which pending/abandoned messages should be checked 'lazy' => false, 'auth' => null, 'serializer' => \Redis::SERIALIZER_PHP, ]; private $connection; private $stream; private $queue; private $group; private $consumer; private $autoSetup; private $maxEntries; private $redeliverTimeout; private $nextClaim = 0.0; private $claimInterval; private $deleteAfterAck; private $deleteAfterReject; private $couldHavePendingMessages = true; /** * @param \Redis|\RedisCluster|null $redis */ public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], $redis = null) { if (version_compare(phpversion('redis'), '4.3.0', '<')) { throw new LogicException('The redis transport requires php-redis 4.3.0 or higher.'); } $host = $connectionCredentials['host'] ?? '127.0.0.1'; $port = $connectionCredentials['port'] ?? 6379; $serializer = $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP; $dbIndex = $configuration['dbindex'] ?? self::DEFAULT_OPTIONS['dbindex']; $auth = $connectionCredentials['auth'] ?? null; if ('' === $auth) { $auth = null; } $lazy = $configuration['lazy'] ?? self::DEFAULT_OPTIONS['lazy']; if (\is_array($host) || $redis instanceof \RedisCluster) { $hosts = \is_string($host) ? [$host.':'.$port] : $host; // Always ensure we have an array $initializer = static function ($redis) use ($hosts, $auth, $serializer) { return self::initializeRedisCluster($redis, $hosts, $auth, $serializer); }; $redis = $lazy ? new RedisClusterProxy($redis, $initializer) : $initializer($redis); } else { $redis = $redis ?? new \Redis(); $initializer = static function ($redis) use ($host, $port, $auth, $serializer, $dbIndex) { return self::initializeRedis($redis, $host, $port, $auth, $serializer, $dbIndex); }; $redis = $lazy ? new RedisProxy($redis, $initializer) : $initializer($redis); } $this->connection = $redis; foreach (['stream', 'group', 'consumer'] as $key) { if (isset($configuration[$key]) && '' === $configuration[$key]) { throw new InvalidArgumentException(sprintf('"%s" should be configured, got an empty string.', $key)); } } $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']; $this->deleteAfterAck = $configuration['delete_after_ack'] ?? self::DEFAULT_OPTIONS['delete_after_ack']; $this->deleteAfterReject = $configuration['delete_after_reject'] ?? self::DEFAULT_OPTIONS['delete_after_reject']; $this->redeliverTimeout = ($configuration['redeliver_timeout'] ?? self::DEFAULT_OPTIONS['redeliver_timeout']) * 1000; $this->claimInterval = ($configuration['claim_interval'] ?? self::DEFAULT_OPTIONS['claim_interval']) / 1000; } /** * @param string|string[]|null $auth */ private static function initializeRedis(\Redis $redis, string $host, int $port, $auth, int $serializer, int $dbIndex): \Redis { $redis->connect($host, $port); $redis->setOption(\Redis::OPT_SERIALIZER, $serializer); if (null !== $auth && !$redis->auth($auth)) { throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError()); } if ($dbIndex && !$redis->select($dbIndex)) { throw new InvalidArgumentException('Redis connection failed: '.$redis->getLastError()); } return $redis; } /** * @param string|string[]|null $auth */ private static function initializeRedisCluster(?\RedisCluster $redis, array $hosts, $auth, int $serializer): \RedisCluster { if (null === $redis) { $redis = new \RedisCluster(null, $hosts, 0.0, 0.0, false, $auth); } $redis->setOption(\Redis::OPT_SERIALIZER, $serializer); return $redis; } /** * @param \Redis|\RedisCluster|null $redis */ public static function fromDsn(string $dsn, array $redisOptions = [], $redis = null): self { if (false === strpos($dsn, ',')) { $parsedUrl = self::parseDsn($dsn, $redisOptions); } else { $dsns = explode(',', $dsn); $parsedUrls = array_map(function ($dsn) use (&$redisOptions) { return self::parseDsn($dsn, $redisOptions); }, $dsns); // Merge all the URLs, the last one overrides the previous ones $parsedUrl = array_merge(...$parsedUrls); // Regroup all the hosts in an array interpretable by RedisCluster $parsedUrl['host'] = array_map(function ($parsedUrl, $dsn) { if (!isset($parsedUrl['host'])) { throw new InvalidArgumentException(sprintf('Missing host in DSN part "%s", it must be defined when using Redis Cluster.', $dsn)); } return $parsedUrl['host'].':'.($parsedUrl['port'] ?? 6379); }, $parsedUrls, $dsns); } self::validateOptions($redisOptions); $autoSetup = null; if (\array_key_exists('auto_setup', $redisOptions)) { $autoSetup = filter_var($redisOptions['auto_setup'], \FILTER_VALIDATE_BOOLEAN); unset($redisOptions['auto_setup']); } $maxEntries = null; if (\array_key_exists('stream_max_entries', $redisOptions)) { $maxEntries = filter_var($redisOptions['stream_max_entries'], \FILTER_VALIDATE_INT); unset($redisOptions['stream_max_entries']); } $deleteAfterAck = null; if (\array_key_exists('delete_after_ack', $redisOptions)) { $deleteAfterAck = filter_var($redisOptions['delete_after_ack'], \FILTER_VALIDATE_BOOLEAN); unset($redisOptions['delete_after_ack']); } else { 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.'); } $deleteAfterReject = null; if (\array_key_exists('delete_after_reject', $redisOptions)) { $deleteAfterReject = filter_var($redisOptions['delete_after_reject'], \FILTER_VALIDATE_BOOLEAN); unset($redisOptions['delete_after_reject']); } $dbIndex = null; if (\array_key_exists('dbindex', $redisOptions)) { $dbIndex = filter_var($redisOptions['dbindex'], \FILTER_VALIDATE_INT); unset($redisOptions['dbindex']); } $tls = 'rediss' === $parsedUrl['scheme']; if (\array_key_exists('tls', $redisOptions)) { trigger_deprecation('symfony/redis-messenger', '5.3', 'Providing "tls" parameter is deprecated, use "rediss://" DSN scheme instead'); $tls = filter_var($redisOptions['tls'], \FILTER_VALIDATE_BOOLEAN); unset($redisOptions['tls']); } $redeliverTimeout = null; if (\array_key_exists('redeliver_timeout', $redisOptions)) { $redeliverTimeout = filter_var($redisOptions['redeliver_timeout'], \FILTER_VALIDATE_INT); unset($redisOptions['redeliver_timeout']); } $claimInterval = null; if (\array_key_exists('claim_interval', $redisOptions)) { $claimInterval = filter_var($redisOptions['claim_interval'], \FILTER_VALIDATE_INT); unset($redisOptions['claim_interval']); } $configuration = [ 'stream' => $redisOptions['stream'] ?? null, 'group' => $redisOptions['group'] ?? null, 'consumer' => $redisOptions['consumer'] ?? null, 'lazy' => $redisOptions['lazy'] ?? self::DEFAULT_OPTIONS['lazy'], 'auto_setup' => $autoSetup, 'stream_max_entries' => $maxEntries, 'delete_after_ack' => $deleteAfterAck, 'delete_after_reject' => $deleteAfterReject, 'dbindex' => $dbIndex, 'redeliver_timeout' => $redeliverTimeout, 'claim_interval' => $claimInterval, ]; if (isset($parsedUrl['host'])) { $pass = '' !== ($parsedUrl['pass'] ?? '') ? urldecode($parsedUrl['pass']) : null; $user = '' !== ($parsedUrl['user'] ?? '') ? urldecode($parsedUrl['user']) : null; $connectionCredentials = [ 'host' => $parsedUrl['host'] ?? '127.0.0.1', 'port' => $parsedUrl['port'] ?? 6379, // See: https://github.com/phpredis/phpredis/#auth 'auth' => $redisOptions['auth'] ?? (null !== $pass && null !== $user ? [$user, $pass] : ($pass ?? $user)), ]; $pathParts = explode('/', rtrim($parsedUrl['path'] ?? '', '/')); $configuration['stream'] = $pathParts[1] ?? $configuration['stream']; $configuration['group'] = $pathParts[2] ?? $configuration['group']; $configuration['consumer'] = $pathParts[3] ?? $configuration['consumer']; if ($tls) { $connectionCredentials['host'] = 'tls://'.$connectionCredentials['host']; } } else { $connectionCredentials = [ 'host' => $parsedUrl['path'], 'port' => 0, ]; } return new self($configuration, $connectionCredentials, $redisOptions, $redis); } private static function parseDsn(string $dsn, array &$redisOptions): array { $url = $dsn; $scheme = 0 === strpos($dsn, 'rediss:') ? 'rediss' : 'redis'; if (preg_match('#^'.$scheme.':///([^:@])+$#', $dsn)) { $url = str_replace($scheme.':', 'file:', $dsn); } if (false === $parsedUrl = parse_url($url)) { throw new InvalidArgumentException(sprintf('The given Redis DSN "%s" is invalid.', $dsn)); } if (isset($parsedUrl['query'])) { parse_str($parsedUrl['query'], $dsnOptions); $redisOptions = array_merge($redisOptions, $dsnOptions); } return $parsedUrl; } private static function validateOptions(array $options): void { $availableOptions = array_keys(self::DEFAULT_OPTIONS); if (0 < \count($invalidOptions = array_diff(array_keys($options), $availableOptions))) { trigger_deprecation('symfony/messenger', '5.1', 'Invalid option(s) "%s" passed to the Redis Messenger transport. Passing invalid options is deprecated.', implode('", "', $invalidOptions)); } } private function claimOldPendingMessages() { try { // This could soon be optimized with https://github.com/antirez/redis/issues/5212 or // https://github.com/antirez/redis/issues/6256 $pendingMessages = $this->connection->xpending($this->stream, $this->group, '-', '+', 1); } catch (\RedisException $e) { throw new TransportException($e->getMessage(), 0, $e); } $claimableIds = []; foreach ($pendingMessages as $pendingMessage) { if ($pendingMessage[1] === $this->consumer) { $this->couldHavePendingMessages = true; return; } if ($pendingMessage[2] >= $this->redeliverTimeout) { $claimableIds[] = $pendingMessage[0]; } } if (\count($claimableIds) > 0) { try { $this->connection->xclaim( $this->stream, $this->group, $this->consumer, $this->redeliverTimeout, $claimableIds, ['JUSTID'] ); $this->couldHavePendingMessages = true; } catch (\RedisException $e) { throw new TransportException($e->getMessage(), 0, $e); } } $this->nextClaim = microtime(true) + $this->claimInterval; } public function get(): ?array { if ($this->autoSetup) { $this->setup(); } $now = microtime(); $now = substr($now, 11).substr($now, 2, 3); $queuedMessageCount = $this->rawCommand('ZCOUNT', 0, $now); while ($queuedMessageCount--) { if (!$message = $this->rawCommand('ZPOPMIN', 1)) { break; } [$queuedMessage, $expiry] = $message; 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 if (!$this->rawCommand('ZADD', 'NX', $expiry, $queuedMessage)) { throw new TransportException('Could not add a message to the redis stream.'); } break; } $decodedQueuedMessage = json_decode($queuedMessage, true); $this->add(\array_key_exists('body', $decodedQueuedMessage) ? $decodedQueuedMessage['body'] : $queuedMessage, $decodedQueuedMessage['headers'] ?? [], 0); } if (!$this->couldHavePendingMessages && $this->nextClaim <= microtime(true)) { $this->claimOldPendingMessages(); } $messageId = '>'; // will receive new messages if ($this->couldHavePendingMessages) { $messageId = '0'; // will receive consumers pending messages } try { $messages = $this->connection->xreadgroup( $this->group, $this->consumer, [$this->stream => $messageId], 1 ); } catch (\RedisException $e) { throw new TransportException($e->getMessage(), 0, $e); } if (false === $messages) { if ($error = $this->connection->getLastError() ?: null) { $this->connection->clearLastError(); } throw new TransportException($error ?? 'Could not read messages from the redis stream.'); } if ($this->couldHavePendingMessages && empty($messages[$this->stream])) { $this->couldHavePendingMessages = false; // No pending messages so get a new one return $this->get(); } foreach ($messages[$this->stream] ?? [] as $key => $message) { return [ 'id' => $key, 'data' => $message, ]; } return null; } public function ack(string $id): void { try { $acknowledged = $this->connection->xack($this->stream, $this->group, [$id]); if ($this->deleteAfterAck) { $acknowledged = $this->connection->xdel($this->stream, [$id]); } } catch (\RedisException $e) { throw new TransportException($e->getMessage(), 0, $e); } if (!$acknowledged) { if ($error = $this->connection->getLastError() ?: null) { $this->connection->clearLastError(); } throw new TransportException($error ?? sprintf('Could not acknowledge redis message "%s".', $id)); } } public function reject(string $id): void { try { $deleted = $this->connection->xack($this->stream, $this->group, [$id]); if ($this->deleteAfterReject) { $deleted = $this->connection->xdel($this->stream, [$id]) && $deleted; } } catch (\RedisException $e) { throw new TransportException($e->getMessage(), 0, $e); } if (!$deleted) { if ($error = $this->connection->getLastError() ?: null) { $this->connection->clearLastError(); } throw new TransportException($error ?? sprintf('Could not delete message "%s" from the redis stream.', $id)); } } public function add(string $body, array $headers, int $delayInMs = 0): void { if ($this->autoSetup) { $this->setup(); } try { if ($delayInMs > 0) { // the delay is <= 0 for queued messages $message = json_encode([ 'body' => $body, 'headers' => $headers, // Entry need to be unique in the sorted set else it would only be added once to the delayed messages queue 'uniqid' => uniqid('', true), ]); if (false === $message) { throw new TransportException(json_last_error_msg()); } $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, 'headers' => $headers, ]); if (false === $message) { throw new TransportException(json_last_error_msg()); } 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) { if ($error = $this->connection->getLastError() ?: null) { $this->connection->clearLastError(); } throw new TransportException($error ?? $e->getMessage(), 0, $e); } if (!$added) { if ($error = $this->connection->getLastError() ?: null) { $this->connection->clearLastError(); } throw new TransportException($error ?? 'Could not add a message to the redis stream.'); } } public function setup(): void { try { $this->connection->xgroup('CREATE', $this->stream, $this->group, 0, true); } catch (\RedisException $e) { throw new TransportException($e->getMessage(), 0, $e); } // group might already exist, ignore if ($this->connection->getLastError()) { $this->connection->clearLastError(); } if ($this->deleteAfterAck || $this->deleteAfterReject) { $groups = $this->connection->xinfo('GROUPS', $this->stream); if ( // support for Redis extension version 5+ (\is_array($groups) && 1 < \count($groups)) // support for Redis extension version 4.x || (\is_string($groups) && substr_count($groups, '"name"')) ) { 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)); } } $this->autoSetup = false; } private function getCurrentTimeInMilliseconds(): int { return (int) (microtime(true) * 1000); } public function cleanup(): void { static $unlink = true; if ($unlink) { try { $unlink = false !== $this->connection->unlink($this->stream, $this->queue); } catch (\Throwable $e) { $unlink = false; } } if (!$unlink) { $this->connection->del($this->stream, $this->queue); } } /** * @return mixed */ private function rawCommand(string $command, ...$arguments) { try { if ($this->connection instanceof \RedisCluster || $this->connection instanceof RedisClusterProxy) { $result = $this->connection->rawCommand($this->queue, $command, $this->queue, ...$arguments); } else { $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; } } if (!class_exists(\Symfony\Component\Messenger\Transport\RedisExt\Connection::class, false)) { class_alias(Connection::class, \Symfony\Component\Messenger\Transport\RedisExt\Connection::class); }