From 23adb182a04bcd73433c1a77db1323e9402452a3 Mon Sep 17 00:00:00 2001 From: upstream Date: Sat, 10 Oct 2026 11:41:46 +0200 Subject: [PATCH] Upstream predis/predis state of the touched files before 6efe01c49b --- CHANGELOG.md | 21 +- .../Replication/SentinelReplication.php | 855 ++++++ tests/Predis/ClientTest.php | 1803 ++++++++++++ .../Replication/SentinelReplicationTest.php | 2427 +++++++++++++++++ 4 files changed, 5101 insertions(+), 5 deletions(-) create mode 100644 src/Connection/Replication/SentinelReplication.php create mode 100644 tests/Predis/ClientTest.php create mode 100644 tests/Predis/Connection/Replication/SentinelReplicationTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a7669d..44d4a4a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,16 +2,27 @@ ## Unreleased ### Added -- Added optional `CELL_SIZE` argument for `CMS.INITBY(DIM|PROB)` (#1729) -- Added `BLESS *` command family (#1733) -- Added testing with a new HNSW compression (#1737) +- Added support for numbered databases in Valkey clusters (#1696) ### Fixed -- Fixed client iteration with master/slave and Sentinel replication (#923) +- Fixed interrupted pipelines leaving unread replies on reused connections; failed pipelines now discard remaining commands and reset execution state (#1309) +- Fixed Relay Pub/Sub subscriptions rejecting message callbacks and failing when the `prefix` option is set (#1292) +- Fixed Relay returning server errors as error responses and raw error messages without stack traces (#1292) +- Fixed Relay reconnecting after an explicit disconnect (#1292) +- Fixed Relay ACL authentication accepting username and password (#1292) + +## v3.7.0 (2026-10-08) +### Added +- Added `BLESS *` command family (#1733) +- Added optional `CELL_SIZE` argument for `CMS.INITBY(DIM|PROB)` (#1729) + +### Fixed +- Fixed Redis Cluster not retrying slot map discovery on `StreamInitException` (#1717) +- Fixed client iteration when using replication (#1730) - Fixed `*StreamConnection` looping forever when closed in the middle of a bulk reply (#1731) - Fixed cluster CAS transactions returning transaction objects for reads and sending `MULTI` twice (#1735) -- Fixed `MultiExec::multi()` being ignored as the first call of a CAS transaction (#1735) - Fixed cleanup of empty transactions and transaction callbacks that throw PHP errors (#1735) +- Fixed `MultiExec::multi()` being ignored as the first call of a CAS transaction (#1735) - Fixed `spl_object_hash()` deprecation on PHP 8.6 in `ConnectionErrorProof` pipeline (#1738) - Fixed `ConnectionErrorProof` pipeline returning too few responses or throwing `ValueError` (#1739) diff --git a/src/Connection/Replication/SentinelReplication.php b/src/Connection/Replication/SentinelReplication.php new file mode 100644 index 0000000..0ae7af6 --- /dev/null +++ b/src/Connection/Replication/SentinelReplication.php @@ -0,0 +1,855 @@ + + * @author Ville Mattila + */ +class SentinelReplication extends AbstractAggregateConnection implements ReplicationInterface +{ + /** + * @var NodeConnectionInterface + */ + protected $master; + + /** + * @var NodeConnectionInterface[] + */ + protected $slaves = []; + + /** + * @var NodeConnectionInterface[] + */ + protected $pool = []; + + /** + * @var NodeConnectionInterface + */ + protected $current; + + /** + * @var string + */ + protected $service; + + /** + * @var ConnectionFactoryInterface + */ + protected $connectionFactory; + + /** + * @var ReplicationStrategy + */ + protected $strategy; + + /** + * Sentinel connection parameters. + * + * Can contain: + * - String URIs (e.g., "tcp://127.0.0.1:26379") + * - Arrays of connection parameters (e.g., ['host' => '127.0.0.1', 'port' => 26379]) + * - ParametersInterface objects + * - NodeConnectionInterface objects + * + * @var array + */ + protected $sentinels = []; + + /** + * @var int + */ + protected $sentinelIndex = 0; + + /** + * @var NodeConnectionInterface + */ + protected $sentinelConnection; + + /** + * @var float + */ + protected $sentinelTimeout = 0.100; + + /** + * Max number of automatic retries of commands upon server failure. + * + * -1 = unlimited retry attempts + * 0 = no retry attempts (fails immediately) + * n = fail only after n retry attempts + * + * @var int + */ + protected $retryLimit = 20; + + /** + * Time to wait in milliseconds before fetching a new configuration from one + * of the sentinel servers. + * + * @var int + */ + protected $retryWait = 1000; + + /** + * Flag for automatic fetching of available sentinels. + * + * @var bool + */ + protected $updateSentinels = false; + + /** + * @param string $service Name of the service for autodiscovery. + * @param array $sentinels Sentinel servers connection parameters. + * @param ConnectionFactoryInterface $connectionFactory Connection factory instance. + * @param ReplicationStrategy|null $strategy Replication strategy instance. + */ + public function __construct( + $service, + array $sentinels, + ConnectionFactoryInterface $connectionFactory, + ?ReplicationStrategy $strategy = null + ) { + $this->sentinels = $sentinels; + $this->service = $service; + $this->connectionFactory = $connectionFactory; + $this->strategy = $strategy ?: new ReplicationStrategy(); + } + + /** + * Sets a default timeout for connections to sentinels. + * + * When "timeout" is present in the connection parameters of sentinels, its + * value overrides the default sentinel timeout. + * + * @param float $timeout Timeout value. + */ + public function setSentinelTimeout($timeout) + { + $this->sentinelTimeout = (float) $timeout; + } + + /** + * Sets the maximum number of retries for commands upon server failure. + * + * -1 = unlimited retry attempts + * 0 = no retry attempts (fails immediately) + * n = fail only after n retry attempts + * + * @param int $retry Number of retry attempts. + */ + public function setRetryLimit($retry) + { + $this->retryLimit = (int) $retry; + } + + /** + * Sets the time to wait (in milliseconds) before fetching a new configuration + * from one of the sentinels. + * + * @param float $milliseconds Time to wait before the next attempt. + */ + public function setRetryWait($milliseconds) + { + $this->retryWait = (float) $milliseconds; + } + + /** + * Set automatic fetching of available sentinels. + * + * @param bool $update Enable or disable automatic updates. + */ + public function setUpdateSentinels($update) + { + $this->updateSentinels = (bool) $update; + } + + /** + * Resets the current connection. + */ + protected function reset() + { + $this->current = null; + } + + /** + * Wipes the current list of master and slaves nodes. + */ + protected function wipeServerList() + { + $this->reset(); + + $this->master = null; + $this->slaves = []; + $this->pool = []; + } + + /** + * {@inheritdoc} + */ + public function add(NodeConnectionInterface $connection) + { + $parameters = $connection->getParameters(); + $role = $parameters->role; + + if ('master' === $role) { + $this->master = $connection; + } elseif ('sentinel' === $role) { + $this->sentinels[] = $connection; + + // sentinels are not considered part of the pool. + return; + } else { + // everything else is considered a slave. + $this->slaves[] = $connection; + } + + $this->pool[(string) $connection] = $connection; + + $this->reset(); + } + + /** + * {@inheritdoc} + */ + public function remove(NodeConnectionInterface $connection) + { + if ($connection === $this->master) { + $this->master = null; + } elseif (false !== $id = array_search($connection, $this->slaves, true)) { + unset($this->slaves[$id]); + } elseif (false !== $id = array_search($connection, $this->sentinels, true)) { + unset($this->sentinels[$id]); + + return true; + } else { + return false; + } + + unset($this->pool[(string) $connection]); + + $this->reset(); + + return true; + } + + /** + * Creates a new connection to a sentinel server. + * + * @return NodeConnectionInterface + */ + protected function createSentinelConnection($parameters) + { + if ($parameters instanceof NodeConnectionInterface) { + return $parameters; + } + + if (is_string($parameters)) { + $parameters = Parameters::parse($parameters); + } + + if (is_array($parameters)) { + // NOTE: sentinels do not accept SELECT command so we must + // explicitly set it to NULL to avoid problems when using default + // parameters set via client options. + $parameters['database'] = null; + + // don't leak password from between configurations + // https://github.com/predis/predis/pull/807/#discussion_r985764770 + if (!isset($parameters['password'])) { + $parameters['password'] = null; + } + + if (!isset($parameters['timeout'])) { + $parameters['timeout'] = $this->sentinelTimeout; + } + } + + return $this->connectionFactory->create($parameters); + } + + /** + * Returns the current sentinel connection. + * + * If there is no active sentinel connection, a new connection is created. + * + * @return NodeConnectionInterface + */ + public function getSentinelConnection() + { + if (!$this->sentinelConnection) { + if ($this->sentinelIndex >= count($this->sentinels)) { + $this->sentinelIndex = 0; + throw new \Predis\ClientException('No sentinel server available for autodiscovery.'); + } + + $sentinel = $this->sentinels[$this->sentinelIndex]; + ++$this->sentinelIndex; + $this->sentinelConnection = $this->createSentinelConnection($sentinel); + } + + return $this->sentinelConnection; + } + + /** + * Fetches an updated list of sentinels from a sentinel. + */ + public function updateSentinels() + { + SENTINEL_QUERY: { + $sentinel = $this->getSentinelConnection(); + + try { + $payload = $sentinel->executeCommand( + RawCommand::create('SENTINEL', 'sentinels', $this->service) + ); + + $this->sentinels = []; + $this->sentinelIndex = 0; + // NOTE: sentinel server does not return itself, so we add it back. + $this->sentinels[] = $sentinel->getParameters()->toArray(); + + foreach ($payload as $sentinel) { + $this->sentinels[] = [ + 'host' => $sentinel[3], + 'port' => $sentinel[5], + 'role' => 'sentinel', + ]; + } + } catch (ConnectionException|StreamInitException $exception) { + $this->sentinelConnection = null; + + goto SENTINEL_QUERY; + } + } + } + + /** + * Fetches the details for the master and slave servers from a sentinel. + */ + public function querySentinel() + { + $this->wipeServerList(); + + $this->updateSentinels(); + $this->getMaster(); + $this->getSlaves(); + } + + /** + * Handles error responses returned by redis-sentinel. + * + * @param NodeConnectionInterface $sentinel Connection to a sentinel server. + * @param ErrorResponseInterface $error Error response. + */ + private function handleSentinelErrorResponse(NodeConnectionInterface $sentinel, ErrorResponseInterface $error) + { + if ($error->getErrorType() === 'IDONTKNOW') { + throw new ConnectionException($sentinel, $error->getMessage()); + } + throw new ServerException($error->getMessage()); + } + + /** + * Fetches the details for the master server from a sentinel. + * + * @param NodeConnectionInterface $sentinel Connection to a sentinel server. + * @param string $service Name of the service. + * + * @return array + */ + protected function querySentinelForMaster(NodeConnectionInterface $sentinel, $service) + { + $payload = $sentinel->executeCommand( + RawCommand::create('SENTINEL', 'get-master-addr-by-name', $service) + ); + + if ($payload === null) { + throw new ServerException('ERR No such master with that name'); + } + + if ($payload instanceof ErrorResponseInterface) { + $this->handleSentinelErrorResponse($sentinel, $payload); + } + + return [ + 'host' => $payload[0], + 'port' => $payload[1], + 'role' => 'master', + ]; + } + + /** + * Fetches the details for the slave servers from a sentinel. + * + * @param NodeConnectionInterface $sentinel Connection to a sentinel server. + * @param string $service Name of the service. + * + * @return array + */ + protected function querySentinelForSlaves(NodeConnectionInterface $sentinel, $service) + { + $slaves = []; + + $payload = $sentinel->executeCommand( + RawCommand::create('SENTINEL', 'slaves', $service) + ); + + if ($payload instanceof ErrorResponseInterface) { + $this->handleSentinelErrorResponse($sentinel, $payload); + } + + foreach ($payload as $slave) { + if ($slave !== [] && !is_string(key($slave))) { + $slave = CommandUtility::arrayToDictionary($slave, null, false); + } + + $flags = explode(',', $slave['flags']); + + if (array_intersect($flags, ['s_down', 'o_down', 'disconnected'])) { + continue; + } + + // ensure `master-link-status` is ok + if (isset($slave['master-link-status']) && $slave['master-link-status'] === 'err') { + continue; + } + + $slaves[] = [ + 'host' => $slave['ip'], + 'port' => $slave['port'], + 'role' => 'slave', + ]; + } + + return $slaves; + } + + /** + * {@inheritdoc} + */ + public function getCurrent() + { + return $this->current; + } + + /** + * {@inheritdoc} + */ + public function getMaster() + { + if ($this->master) { + return $this->master; + } + + if ($this->updateSentinels) { + $this->updateSentinels(); + } + + SENTINEL_QUERY: { + $sentinel = $this->getSentinelConnection(); + + try { + $masterParameters = $this->querySentinelForMaster($sentinel, $this->service); + $masterConnection = $this->connectionFactory->create($masterParameters); + + $this->add($masterConnection); + } catch (ConnectionException|StreamInitException $exception) { + $this->sentinelConnection = null; + + goto SENTINEL_QUERY; + } + } + + return $masterConnection; + } + + /** + * {@inheritdoc} + */ + public function getSlaves() + { + if ($this->slaves) { + return array_values($this->slaves); + } + + if ($this->updateSentinels) { + $this->updateSentinels(); + } + + SENTINEL_QUERY: { + $sentinel = $this->getSentinelConnection(); + + try { + $slavesParameters = $this->querySentinelForSlaves($sentinel, $this->service); + + foreach ($slavesParameters as $slaveParameters) { + $this->add($this->connectionFactory->create($slaveParameters)); + } + } catch (ConnectionException|StreamInitException $exception) { + $this->sentinelConnection = null; + + goto SENTINEL_QUERY; + } + } + + return array_values($this->slaves); + } + + /** + * Returns a random slave. + * + * @return NodeConnectionInterface|null + */ + protected function pickSlave() + { + $slaves = $this->getSlaves(); + + return $slaves + ? $slaves[rand(1, count($slaves)) - 1] + : null; + } + + /** + * Returns the connection instance in charge for the given command. + * + * @param CommandInterface $command Command instance. + * + * @return NodeConnectionInterface + */ + private function getConnectionInternal(CommandInterface $command) + { + if (!$this->current) { + if ($this->strategy->isReadOperation($command) && $slave = $this->pickSlave()) { + $this->current = $slave; + } else { + $this->current = $this->getMaster(); + } + + return $this->current; + } + + if ($this->current === $this->master) { + return $this->current; + } + + if (!$this->strategy->isReadOperation($command)) { + $this->current = $this->getMaster(); + } + + return $this->current; + } + + /** + * Asserts that the specified connection matches an expected role. + * + * @param NodeConnectionInterface $connection Connection to a redis server. + * @param string $role Expected role of the server ("master", "slave" or "sentinel"). + * + * @throws RoleException|ConnectionException + */ + protected function assertConnectionRole(NodeConnectionInterface $connection, $role) + { + $role = strtolower($role); + $retry = $connection->getParameters()->retry; + $actualRole = $retry->callWithRetry(static function () use ($connection) { + return $connection->executeCommand(RawCommand::create('ROLE')); + }); + + if ($actualRole instanceof Error) { + throw new ConnectionException($connection, $actualRole->getMessage()); + } + + if ($role !== $actualRole[0]) { + throw new RoleException($connection, "Expected $role but got $actualRole[0] [$connection]"); + } + } + + /** + * {@inheritdoc} + */ + public function getConnectionByCommand(CommandInterface $command) + { + $connection = $this->getConnectionInternal($command); + + if (!$connection->isConnected()) { + // When we do not have any available slave in the pool we can expect + // read-only operations to hit the master server. + $expectedRole = $this->strategy->isReadOperation($command) && $this->slaves ? 'slave' : 'master'; + $this->assertConnectionRole($connection, $expectedRole); + } + + return $connection; + } + + /** + * {@inheritdoc} + */ + public function getConnectionById($id) + { + return $this->pool[$id] ?? null; + } + + /** + * Returns a connection by its role. + * + * @param string $role Connection role (`master`, `slave` or `sentinel`) + * + * @return NodeConnectionInterface|null + */ + public function getConnectionByRole($role) + { + if ($role === 'master') { + return $this->getMaster(); + } elseif ($role === 'slave') { + return $this->pickSlave(); + } elseif ($role === 'sentinel') { + return $this->getSentinelConnection(); + } + + return null; + } + + /** + * Switches the internal connection in use by the backend. + * + * Sentinel connections are not considered as part of the pool, meaning that + * trying to switch to a sentinel will throw an exception. + * + * @param NodeConnectionInterface $connection Connection instance in the pool. + */ + public function switchTo(NodeConnectionInterface $connection) + { + if ($connection && $connection === $this->current) { + return; + } + + if ($connection !== $this->master && !in_array($connection, $this->slaves, true)) { + throw new InvalidArgumentException('Invalid connection or connection not found.'); + } + + $connection->connect(); + + if ($this->current) { + $this->current->disconnect(); + } + + $this->current = $connection; + } + + /** + * {@inheritdoc} + */ + public function switchToMaster() + { + $connection = $this->getConnectionByRole('master'); + $this->switchTo($connection); + } + + /** + * {@inheritdoc} + */ + public function switchToSlave() + { + $connection = $this->getConnectionByRole('slave'); + $this->switchTo($connection); + } + + /** + * {@inheritdoc} + */ + public function isConnected() + { + return $this->current ? $this->current->isConnected() : false; + } + + /** + * {@inheritdoc} + */ + public function connect() + { + if (!$this->current) { + if (!$this->current = $this->pickSlave()) { + $this->current = $this->getMaster(); + } + } + + $this->current->connect(); + } + + /** + * {@inheritdoc} + */ + public function disconnect() + { + foreach ($this->pool as $connection) { + $connection->disconnect(); + } + } + + /** + * Retries the execution of a command upon server failure after asking a new + * configuration to one of the sentinels. + * + * @param CommandInterface $command Command instance. + * @param string $method Actual method. + * + * @return mixed + */ + private function retryCommandOnFailure(CommandInterface $command, $method) + { + $parameters = $this->getParameters(); + + if ($parameters->isDisabledRetry() || $this->connectionFactory instanceof RelayFactory) { + // Override default parameters, for backward-compatibility + // with current behaviour + $retry = new Retry( + new ExponentialBackoff($this->retryWait * 1000, -1), + $this->retryLimit + ); + } else { + $retry = $parameters->retry; + } + $retry->updateCatchableExceptions([CommunicationException::class]); + + $doCallback = function () use ($method, $command) { + $response = $this->getConnectionByCommand($command)->{$method}($command); + + if ($response instanceof Error && $response->getErrorType() === 'LOADING') { + throw new ConnectionException($this->current, $response->getMessage()); + } + + return $response; + }; + + $failCallback = function (Throwable $exception) { + $this->wipeServerList(); + + if ($exception instanceof CommunicationException) { + $exception->getConnection()->disconnect(); + } + }; + + return $retry->callWithRetry($doCallback, $failCallback); + } + + /** + * {@inheritdoc} + */ + public function writeRequest(CommandInterface $command) + { + $this->retryCommandOnFailure($command, __FUNCTION__); + } + + /** + * {@inheritdoc} + */ + public function readResponse(CommandInterface $command) + { + return $this->retryCommandOnFailure($command, __FUNCTION__); + } + + /** + * {@inheritdoc} + */ + public function executeCommand(CommandInterface $command) + { + return $this->retryCommandOnFailure($command, __FUNCTION__); + } + + /** + * Returns the underlying replication strategy. + * + * @return ReplicationStrategy + */ + public function getReplicationStrategy() + { + return $this->strategy; + } + + /** + * {@inheritdoc} + */ + public function __sleep() + { + return [ + 'master', 'slaves', 'pool', 'service', 'sentinels', 'connectionFactory', 'strategy', + ]; + } + + /** + * {@inheritdoc} + */ + public function getParameters(): ?ParametersInterface + { + if (isset($this->master)) { + return $this->master->getParameters(); + } + + if (!empty($this->slaves)) { + return $this->slaves[0]->getParameters(); + } + + if (!empty($this->sentinels)) { + $sentinel = $this->sentinels[0]; + + // Handle string URIs (e.g., "tcp://127.0.0.1:26379") + if (is_string($sentinel)) { + return new Parameters(Parameters::parse($sentinel)); + } + + // After querySentinels(), sentinels array contains plain arrays instead of connection objects + if (is_array($sentinel)) { + return new Parameters($sentinel); + } + + if ($sentinel instanceof ParametersInterface) { + return $sentinel; + } + + return $sentinel->getParameters(); + } + + return null; + } + + /** + * @return string + */ + public function __toString() + { + return implode(',', array_keys($this->pool)); + } +} diff --git a/tests/Predis/ClientTest.php b/tests/Predis/ClientTest.php new file mode 100644 index 0000000..024fe05 --- /dev/null +++ b/tests/Predis/ClientTest.php @@ -0,0 +1,1803 @@ +getConnection(); + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $connection); + + $parameters = $connection->getParameters(); + $this->assertSame($parameters->host, '127.0.0.1'); + $this->assertSame($parameters->port, 6379); + + $options = $client->getOptions(); + $this->assertSame($options->commands, $client->getCommandFactory()); + + $this->assertFalse($client->isConnected()); + } + + /** + * @group disconnected + */ + public function testConstructorWithNullArgument(): void + { + $client = new Client(null); + + /** @var NodeConnectionInterface */ + $connection = $client->getConnection(); + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $connection); + + $parameters = $connection->getParameters(); + $this->assertSame($parameters->host, '127.0.0.1'); + $this->assertSame($parameters->port, 6379); + + $options = $client->getOptions(); + $this->assertSame($options->commands, $client->getCommandFactory()); + + $this->assertFalse($client->isConnected()); + } + + /** + * @group disconnected + */ + public function testConstructorWithNullAndNullArguments(): void + { + $client = new Client(null, null); + + /** @var NodeConnectionInterface */ + $connection = $client->getConnection(); + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $connection); + + $parameters = $connection->getParameters(); + $this->assertSame($parameters->host, '127.0.0.1'); + $this->assertSame($parameters->port, 6379); + + $options = $client->getOptions(); + $this->assertSame($options->commands, $client->getCommandFactory()); + + $this->assertFalse($client->isConnected()); + } + + /** + * @group disconnected + */ + public function testConstructorWithArrayArgument(): void + { + $client = new Client($arg1 = ['host' => 'localhost', 'port' => 7000]); + + /** @var NodeConnectionInterface */ + $connection = $client->getConnection(); + $parameters = $connection->getParameters(); + + $this->assertSame($parameters->host, $arg1['host']); + $this->assertSame($parameters->port, $arg1['port']); + } + + /** + * @group disconnected + */ + public function testConstructorThrowsExceptionWithArrayOfParametersArgumentAndMissingOption(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Array of connection parameters requires `cluster`, `replication` or `aggregate` client option'); + + $arg1 = [ + ['host' => 'localhost', 'port' => 7000], + ['host' => 'localhost', 'port' => 7001], + ]; + + $client = new Client($arg1); + } + + /** + * @group disconnected + */ + public function testConstructorWithArrayOfArrayArgumentAndClusterOption(): void + { + $arg1 = [ + ['host' => 'localhost', 'port' => 7000], + ['host' => 'localhost', 'port' => 7001], + ]; + + $client = new Client($arg1, [ + 'aggregate' => $this->getAggregateInitializer($arg1), + ]); + + $this->assertInstanceOf('Predis\Connection\AggregateConnectionInterface', $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithStringArgument(): void + { + $client = new Client('tcp://localhost:7000'); + + /** @var NodeConnectionInterface */ + $connection = $client->getConnection(); + $parameters = $connection->getParameters(); + + $this->assertSame($parameters->host, 'localhost'); + $this->assertSame($parameters->port, 7000); + } + + /** + * @group disconnected + */ + public function testConstructorWithArrayOfStringArgument(): void + { + $arg1 = ['tcp://localhost:7000', 'tcp://localhost:7001']; + + $client = new Client($arg1, [ + 'aggregate' => $this->getAggregateInitializer($arg1), + ]); + + $this->assertInstanceOf('Predis\Connection\AggregateConnectionInterface', $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithArrayOfConnectionsArgument(): void + { + $arg1 = [ + $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(), + $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(), + ]; + + $client = new Client($arg1, [ + 'aggregate' => $this->getAggregateInitializer($arg1), + ]); + + $this->assertInstanceOf('Predis\Connection\AggregateConnectionInterface', $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithConnectionArgument(): void + { + $factory = new Factory(); + $connection = $factory->create('tcp://localhost:7000'); + + $client = new Client($connection); + + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $client->getConnection()); + $this->assertSame($connection, $client->getConnection()); + + /** @var NodeConnectionInterface */ + $connection = $client->getConnection(); + $parameters = $connection->getParameters(); + + $this->assertSame($parameters->host, 'localhost'); + $this->assertSame($parameters->port, 7000); + } + + /** + * @group disconnected + */ + public function testConstructorWithClusterArgument(): void + { + $cluster = new Connection\Cluster\PredisCluster(new Parameters()); + + $factory = new Factory(); + $cluster->add($factory->create('tcp://localhost:7000')); + $cluster->add($factory->create('tcp://localhost:7001')); + + $client = new Client($cluster); + + $this->assertInstanceOf('Predis\Connection\Cluster\ClusterInterface', $client->getConnection()); + $this->assertSame($cluster, $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithReplicationArgument(): void + { + $replication = new MasterSlaveReplication(); + + $factory = new Factory(); + $replication->add($factory->create('tcp://host1?alias=master')); + $replication->add($factory->create('tcp://host2?alias=slave')); + + $client = new Client($replication); + + $this->assertInstanceOf('Predis\Connection\Replication\ReplicationInterface', $client->getConnection()); + $this->assertSame($replication, $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithCallableArgument(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->once()) + ->method('__invoke') + ->with($this->isInstanceOf('Predis\Configuration\OptionsInterface')) + ->willReturn($connection); + + $client = new Client($callable); + + $this->assertSame($connection, $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithCallableConnectionInitializerThrowsExceptionOnInvalidReturnType(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Callable parameters must return a valid connection'); + + $wrongType = $this->getMockBuilder('stdClass')->getMock(); + + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->once()) + ->method('__invoke') + ->with($this->isInstanceOf('Predis\Configuration\OptionsInterface')) + ->willReturn($wrongType); + + new Client($callable); + } + + /** + * @group disconnected + */ + public function testConstructorWithNullAndArrayArgument(): void + { + $connections = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + + $arg2 = ['prefix' => 'prefix:', 'connections' => $connections]; + $client = new Client(null, $arg2); + + /** @var CommandFactory */ + $commands = $client->getCommandFactory(); + $this->assertInstanceOf('Predis\Command\FactoryInterface', $commands); + + /** @var KeyPrefixProcessor */ + $processor = $commands->getProcessor(); + $this->assertInstanceOf('Predis\Command\Processor\KeyPrefixProcessor', $processor); + $this->assertSame('prefix:', $processor->getPrefix()); + } + + /** + * @group disconnected + */ + public function testConstructorWithArrayAndOptionReplication(): void + { + $arg1 = ['tcp://127.0.0.1:6379?role=master', 'tcp://127.0.0.1:6380?role=slave']; + $arg2 = ['replication' => 'predis']; + $client = new Client($arg1, $arg2); + + /** @var MasterSlaveReplication */ + $connection = $client->getConnection(); + + $this->assertInstanceOf('Predis\Connection\Replication\ReplicationInterface', $connection); + $this->assertSame('127.0.0.1:6379', (string) $connection->getConnectionByRole('master')); + $this->assertSame('127.0.0.1:6380', (string) $connection->getConnectionByRole('slave')); + } + + /** + * @group disconnected + */ + public function testClusterOptionHasPrecedenceOverReplicationOptionAndAggregateOption(): void + { + $arg1 = ['tcp://host1', 'tcp://host2']; + + $connection = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface')->getMock(); + + $fncluster = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $fncluster + ->expects($this->once()) + ->method('__invoke') + ->with( + $arg1, + $this->isInstanceOf('Predis\Configuration\OptionsInterface'), + $this->isInstanceOf('Predis\Configuration\OptionInterface') + ) + ->willReturn($connection); + + $fnreplication = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $fnreplication + ->expects($this->never()) + ->method('__invoke'); + + $fnaggregate = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $fnaggregate + ->expects($this->never()) + ->method('__invoke'); + + $arg2 = [ + 'cluster' => $fncluster, + 'replication' => $fnreplication, + 'aggregate' => $fnaggregate, + ]; + + $client = new Client($arg1, $arg2); + + $this->assertSame($connection, $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testReplicationOptionHasPrecedenceOverAggregateOption(): void + { + $arg1 = ['tcp://host1', 'tcp://host2']; + + $connection = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface')->getMock(); + + $fnreplication = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $fnreplication + ->expects($this->once()) + ->method('__invoke') + ->with( + $arg1, + $this->isInstanceOf('Predis\Configuration\OptionsInterface'), + $this->isInstanceOf('Predis\Configuration\OptionInterface') + ) + ->willReturn($connection); + + $fnaggregate = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $fnaggregate + ->expects($this->never()) + ->method('__invoke'); + + $arg2 = [ + 'replication' => $fnreplication, + 'aggregate' => $fnaggregate, + ]; + + $client = new Client($arg1, $arg2); + } + + /** + * @group disconnected + */ + public function testAggregateOptionDoesNotTriggerAggregationInClient(): void + { + $arg1 = ['tcp://host1', 'tcp://host2']; + + $connections = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $connections + ->expects($this->never()) + ->method('create'); + + $connection = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->getMock(); + $connection + ->expects($this->never()) + ->method('add'); + + $fnaggregate = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $fnaggregate + ->expects($this->once()) + ->method('__invoke') + ->with( + $arg1, + $this->isInstanceOf('Predis\Configuration\OptionsInterface'), + $this->isInstanceOf('Predis\Configuration\OptionInterface') + ) + ->willReturn($connection); + + $arg2 = ['aggregate' => $fnaggregate, 'connections' => $connections]; + + $client = new Client($arg1, $arg2); + + $this->assertSame($connection, $client->getConnection()); + } + + /** + * @group disconnected + */ + public function testConstructorWithInvalidArgumentType(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Invalid type for connection parameters'); + + $client = new Client(new stdClass()); + } + + /** + * @group disconnected + */ + public function testConstructorWithInvalidOptionType(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Invalid type for client options'); + + $client = new Client('tcp://host1', new stdClass()); + } + + /** + * @group disconnected + */ + public function testConnectAndDisconnect(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('connect'); + $connection + ->expects($this->once()) + ->method('disconnect'); + + $client = new Client($connection); + $client->connect(); + $client->disconnect(); + } + + /** + * @group disconnected + */ + public function testIsConnectedChecksConnectionState(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('isConnected'); + + $client = new Client($connection); + $client->isConnected(); + } + + /** + * @group disconnected + */ + public function testQuitIsAliasForDisconnect(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('disconnect'); + + $client = new Client($connection); + $client->quit(); + } + + /** + * @group disconnected + */ + public function testCreatesNewCommandUsingSpecifiedCommandFactory(): void + { + $ping = $this->getCommandFactory()->create('ping', []); + + $commands = $this->getMockBuilder('Predis\Command\FactoryInterface')->getMock(); + $commands + ->expects($this->once()) + ->method('create') + ->with('ping', []) + ->willReturn($ping); + + $client = new Client(null, ['commands' => $commands]); + $this->assertSame($ping, $client->createCommand('ping', [])); + } + + /** + * @group disconnected + */ + public function testExecuteCommandReturnsParsedResponses(): void + { + $commands = $this->getCommandFactory(); + + $ping = $commands->create('ping', []); + $hgetall = $commands->create('hgetall', ['metavars', 'foo', 'hoge']); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->exactly(2)) + ->method('executeCommand') + ->withConsecutive( + [$ping], + [$hgetall] + ) + ->willReturnOnConsecutiveCalls( + new Response\Status('PONG'), + ['foo', 'bar', 'hoge', 'piyo'] + ); + + $connection + ->expects($this->exactly(2)) + ->method('getParameters') + ->willReturn(new Parameters(['protocol' => 2])); + + $client = new Client($connection); + + $this->assertEquals('PONG', $client->executeCommand($ping)); + $this->assertSame(['foo' => 'bar', 'hoge' => 'piyo'], $client->executeCommand($hgetall)); + } + + /** + * @group disconnected + */ + public function testExecuteCommandReturnsResp3ParsedResponses(): void + { + $commands = $this->getCommandFactory(); + + $ping = $commands->create('ping', []); + $get = $commands->create('get', []); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->exactly(2)) + ->method('executeCommand') + ->withConsecutive( + [$ping], + [$get] + ) + ->willReturnOnConsecutiveCalls( + new Response\Status('PONG'), + [] + ); + + $connection + ->expects($this->exactly(2)) + ->method('getParameters') + ->willReturn(new Parameters(['protocol' => 3])); + + $client = new Client($connection); + + $this->assertEquals('PONG', $client->executeCommand($ping)); + $this->assertSame([], $client->executeCommand($get)); + } + + /** + * @group disconnected + */ + public function testExecuteCommandThrowsExceptionOnRedisError(): void + { + $this->expectException('Predis\Response\ServerException'); + $this->expectExceptionMessage('Operation against a key holding the wrong kind of value'); + + $ping = $this->getCommandFactory()->create('ping', []); + $expectedResponse = new Response\Error('ERR Operation against a key holding the wrong kind of value'); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->willReturn($expectedResponse); + $connection + ->expects($this->once()) + ->method('getParameters') + ->willReturn(new Parameters()); + + $client = new Client($connection); + $client->executeCommand($ping); + } + + /** + * @group disconnected + */ + public function testExecuteCommandReturnsErrorResponseOnRedisError(): void + { + $ping = $this->getCommandFactory()->create('ping', []); + $expectedResponse = new Response\Error('ERR Operation against a key holding the wrong kind of value'); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->willReturn($expectedResponse); + $connection + ->expects($this->once()) + ->method('getParameters') + ->willReturn(new Parameters()); + + $client = new Client($connection, ['exceptions' => false]); + $response = $client->executeCommand($ping); + + $this->assertSame($response, $expectedResponse); + } + + /** + * @group disconnected + */ + public function testCallingRedisCommandExecutesInstanceOfCommand(): void + { + $ping = $this->getCommandFactory()->create('ping', []); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isInstanceOf('Predis\Command\Redis\PING')) + ->willReturn('PONG'); + + $connection + ->expects($this->once()) + ->method('getParameters') + ->willReturn(new Parameters(['protocol' => 2])); + + $commands = $this->getMockBuilder('Predis\Command\FactoryInterface')->getMock(); + $commands + ->expects($this->once()) + ->method('create') + ->with('ping', []) + ->willReturn($ping); + + $options = ['commands' => $commands]; + + /** @var ClientInterface */ + $client = $this->getMockBuilder('Predis\Client') + ->onlyMethods([]) + ->setConstructorArgs([$connection, $options]) + ->getMock(); + + $this->assertEquals('PONG', $client->ping()); + } + + /** + * @group disconnected + */ + public function testCallingRedisCommandThrowsExceptionOnServerError(): void + { + $this->expectException('Predis\Response\ServerException'); + $this->expectExceptionMessage('Operation against a key holding the wrong kind of value'); + + $expectedResponse = new Response\Error('ERR Operation against a key holding the wrong kind of value'); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('PING')) + ->willReturn($expectedResponse); + + $connection + ->expects($this->once()) + ->method('getParameters') + ->willReturn(new Parameters()); + + $client = new Client($connection); + $client->ping(); + } + + /** + * @group disconnected + */ + public function testCallingRedisCommandReturnsErrorResponseOnRedisError(): void + { + $expectedResponse = new Response\Error('ERR Operation against a key holding the wrong kind of value'); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('PING')) + ->willReturn($expectedResponse); + $connection + ->expects($this->once()) + ->method('getParameters') + ->willReturn(new Parameters()); + + $client = new Client($connection, ['exceptions' => false]); + $response = $client->ping(); + + $this->assertSame($response, $expectedResponse); + } + + /** + * @group disconnected + */ + public function testRawCommand(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->exactly(3)) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SET', ['foo', 'bar'])], + [$this->isRedisCommand('GET', ['foo'])], + [$this->isRedisCommand('PING')] + ) + ->willReturnOnConsecutiveCalls( + new Response\Status('OK'), + 'bar', + 'PONG' + ); + + $client = new Client($connection); + + $this->assertSame('OK', $client->executeRaw(['SET', 'foo', 'bar'])); + $this->assertSame('bar', $client->executeRaw(['GET', 'foo'])); + + $error = true; // $error is always populated by reference. + $this->assertSame('PONG', $client->executeRaw(['PING'], $error)); + $this->assertFalse($error); + } + + /** + * @group disconnected + */ + public function testRawCommandNeverAppliesPrefix(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->exactly(2)) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SET', ['foo', 'bar'])], + [$this->isRedisCommand('GET', ['foo'])] + ) + ->willReturnOnConsecutiveCalls( + new Response\Status('OK'), + 'bar' + ); + + $client = new Client($connection, ['prefix' => 'predis:']); + + $this->assertSame('OK', $client->executeRaw(['SET', 'foo', 'bar'])); + $this->assertSame('bar', $client->executeRaw(['GET', 'foo'])); + } + + /** + * @group disconnected + */ + public function testRawCommandNeverThrowsExceptions(): void + { + $message = 'ERR Mock error response'; + $response = new Response\Error($message); + + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('PING')) + ->willReturn($response); + + $client = new Client($connection, ['exceptions' => true]); + + $this->assertSame($message, $client->executeRaw(['PING'], $error)); + $this->assertTrue($error); + } + + /** + * @group disconnected + */ + public function testThrowsExceptionOnNonRegisteredRedisCommand(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('Command `INVALIDCOMMAND` is not a registered Redis command'); + + $client = new Client(); + $client->invalidCommand(); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodReturnsInstanceOfSubclass(): void + { + /** @var Client */ + $client = $this->getMockBuilder('Predis\Client') + ->onlyMethods([]) + ->setConstructorArgs([ + ['tcp://host1?alias=node01', 'tcp://host2?alias=node02'], + ['cluster' => 'predis'], + ]) + ->setMockClassName('SubclassedClient') + ->getMock(); + + $this->assertInstanceOf('SubclassedClient', $client->getClientBy('alias', 'node02')); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodSupportsSelectingConnectionById(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->onlyMethods(['getConnectionById']) + ->getMockForAbstractClass(); + $aggregate + ->expects($this->once()) + ->method('getConnectionById') + ->with('127.0.0.1:6379') + ->willReturn($connection); + + $client = new Client($aggregate); + $nodeClient = $client->getClientBy('id', '127.0.0.1:6379'); + + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame($client->getOptions(), $nodeClient->getOptions()); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodThrowsExceptionSelectingConnectionByUnknownId(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Cannot find a connection by id matching `127.0.0.1:7000`'); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface')->getMock(); + $aggregate + ->expects($this->once()) + ->method('getConnectionById') + ->with('127.0.0.1:7000') + ->willReturn(null); + + $client = new Client($aggregate); + $client->getClientBy('id', '127.0.0.1:7000'); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodSupportsSelectingConnectionByAlias(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->addMethods(['getConnectionByAlias']) + ->getMockForAbstractClass(); + $aggregate + ->expects($this->once()) + ->method('getConnectionByAlias') + ->with('myalias') + ->willReturn($connection); + + $client = new Client($aggregate); + $nodeClient = $client->getClientBy('alias', 'myalias'); + + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame($client->getOptions(), $nodeClient->getOptions()); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodSupportsSelectingConnectionByKey(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->addMethods(['getConnectionByKey']) + ->getMockForAbstractClass(); + $aggregate + ->expects($this->once()) + ->method('getConnectionByKey') + ->with('key:1') + ->willReturn($connection); + + $client = new Client($aggregate); + $nodeClient = $client->getClientBy('key', 'key:1'); + + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame($client->getOptions(), $nodeClient->getOptions()); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodSupportsSelectingConnectionBySlot(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->addMethods(['getConnectionBySlot']) + ->getMockForAbstractClass(); + $aggregate + ->expects($this->once()) + ->method('getConnectionBySlot') + ->with(5460) + ->willReturn($connection); + + $client = new Client($aggregate); + $nodeClient = $client->getClientBy('slot', 5460); + + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame($client->getOptions(), $nodeClient->getOptions()); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodSupportsSelectingConnectionByRole(): void + { + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->addMethods(['getConnectionByRole']) + ->getMockForAbstractClass(); + $aggregate + ->expects($this->once()) + ->method('getConnectionByRole') + ->with('master') + ->willReturn($connection); + + $client = new Client($aggregate); + $nodeClient = $client->getClientBy('role', 'master'); + + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame($client->getOptions(), $nodeClient->getOptions()); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodSupportsSelectingConnectionByCommand(): void + { + $command = RawCommand::create('GET', 'key'); + $connection = $this->getMockBuilder('Predis\Connection\ConnectionInterface')->getMock(); + + $aggregate = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface') + ->onlyMethods(['getConnectionByCommand']) + ->getMockForAbstractClass(); + $aggregate + ->expects($this->once()) + ->method('getConnectionByCommand') + ->with($command) + ->willReturn($connection); + + $client = new Client($aggregate); + $nodeClient = $client->getClientBy('command', $command); + + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame($client->getOptions(), $nodeClient->getOptions()); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodThrowsExceptionWhenSelectingConnectionByUnknownType(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Invalid selector type: `unknown`'); + + $client = new Client('tcp://127.0.0.1?alias=node01'); + + $client->getClientBy('unknown', 'test'); + } + + /** + * @group disconnected + */ + public function testGetClientByMethodThrowsExceptionWhenConnectionDoesNotSupportSelectorType(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Selecting connection by id is not supported by Predis\Connection\StreamConnection'); + + $client = new Client('tcp://127.0.0.1?alias=node01'); + + $client->getClientBy('id', 'node01'); + } + + /** + * @group disconnected + */ + public function testPipelineWithoutArgumentsReturnsPipeline(): void + { + $client = new Client(); + + $this->assertInstanceOf('Predis\Pipeline\Pipeline', $client->pipeline()); + } + + /** + * @group disconnected + */ + public function testPipelineWithArrayReturnsPipeline(): void + { + $client = new Client(); + + $this->assertInstanceOf('Predis\Pipeline\Pipeline', $client->pipeline([])); + $this->assertInstanceOf('Predis\Pipeline\Atomic', $client->pipeline(['atomic' => true])); + $this->assertInstanceOf('Predis\Pipeline\FireAndForget', $client->pipeline(['fire-and-forget' => true])); + } + + /** + * @group disconnected + */ + public function testPipelineWithCallableExecutesPipeline(): void + { + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->once()) + ->method('__invoke') + ->with($this->isInstanceOf('Predis\Pipeline\Pipeline')); + + $client = new Client(); + $client->pipeline($callable); + } + + /** + * @group disconnected + */ + public function testPubSubLoopWithoutArgumentsReturnsPubSubConsumer(): void + { + $client = new Client(); + + $this->assertInstanceOf('Predis\Consumer\PubSub\Consumer', $client->pubSubLoop()); + } + + /** + * @group disconnected + */ + public function testPubSubLoopWithArrayReturnsPubSubConsumerWithOptions(): void + { + $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(); + $options = ['subscribe' => 'channel']; + + $client = new Client($connection); + + $this->assertInstanceOf('Predis\Consumer\PubSub\Consumer', $pubsub = $client->pubSubLoop($options)); + + $reflection = new ReflectionProperty($pubsub, 'options'); + $reflection->setAccessible(true); + + $this->assertSame($options, $reflection->getValue($pubsub)); + } + + /** + * @group disconnected + */ + public function testPubSubLoopWithArrayAndCallableExecutesPubSub(): void + { + // NOTE: we use a subscribe count of 0 in the message payload to trick + // the context and forcing it to be closed since there are no more + // active subscriptions. + $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('read') + ->willReturn(['subscribe', 'channel', 0]); + + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->once()) + ->method('__invoke'); + + $client = new Client($connection); + $this->assertNull($client->pubSubLoop(['subscribe' => 'channel'], $callable)); + } + + /** + * @group disconnected + */ + public function testPubSubLoopWithCallableReturningFalseStopsPubSubConsumer(): void + { + $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(); + $connection + ->expects($this->exactly(2)) + ->method('read') + ->willReturnOnConsecutiveCalls( + ['subscribe', 'channel', 1], + ['unsubscribe', 'channel', 0] + ); + $connection + ->expects($this->exactly(2)) + ->method('writeRequest') + ->withConsecutive( + [$this->isRedisCommand('SUBSCRIBE')], + [$this->isRedisCommand('UNSUBSCRIBE')] + ); + + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->exactly(2)) + ->method('__invoke') + ->withConsecutive( + [ + $this->isInstanceOf('Predis\Consumer\PubSub\Consumer'), + (object) ['kind' => 'subscribe', 'channel' => 'channel', 'payload' => 1], + ], + [ + $this->isInstanceOf('Predis\Consumer\PubSub\Consumer'), + (object) ['kind' => 'unsubscribe', 'channel' => 'channel', 'payload' => 0], + ] + ) + ->willReturnOnConsecutiveCalls( + false, + null // <-- this value would be ignored as it is the callback to UNSUBSCRIBE + ); + + $client = new Client($connection); + + $this->assertNull($client->pubSubLoop(['subscribe' => 'channel'], $callable)); + } + + /** + * @group disconnected + */ + public function testTransactionWithoutArgumentsReturnsMultiExec(): void + { + $client = new Client(); + + $this->assertInstanceOf('Predis\Transaction\MultiExec', $client->transaction()); + } + + /** + * @group disconnected + */ + public function testTransactionWithArrayReturnsMultiExecTransactionWithOptions(): void + { + $options = ['cas' => true, 'retry' => 3]; + + $client = new Client(); + + $this->assertInstanceOf('Predis\Transaction\MultiExec', $tx = $client->transaction($options)); + + // I hate this part but reflection is the easiest way in this case. + $property = new ReflectionProperty($tx, 'modeCAS'); + $property->setAccessible(true); + $this->assertSame($options['cas'], $property->getValue($tx)); + + $property = new ReflectionProperty($tx, 'attempts'); + $property->setAccessible(true); + $this->assertSame($options['retry'], $property->getValue($tx)); + } + + /** + * @group disconnected + */ + public function testTransactionWithArrayAndCallableExecutesMultiExec(): void + { + // We use CAS here as we don't care about the actual MULTI/EXEC context. + $options = ['cas' => true, 'retry' => 3]; + + $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('executeCommand') + ->willReturn(new Response\Status('QUEUED')); + $connection + ->expects($this->any()) + ->method('getParameters') + ->willReturn(new Parameters()); + + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->once()) + ->method('__invoke') + ->willReturnCallback(static function ($tx) { $tx->ping(); }); + + $client = new Client($connection); + $client->transaction($options, $callable); + } + + /** + * @group disconnected + */ + public function testMonitorReturnsMonitorConsumer(): void + { + $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(); + $connection + ->expects($this->once()) + ->method('getParameters') + ->willReturn(new Parameters(['protocol' => 2])); + + $client = new Client($connection); + + $this->assertInstanceOf('Predis\Monitor\Consumer', $monitor = $client->monitor()); + } + + /** + * @group disconnected + */ + public function testClientResendScriptCommandUsingEvalOnNoScriptErrors(): void + { + $luaScriptBody = 'return redis.call(\'exists\', KEYS[1])'; + + $command = $this->getMockForAbstractClass('Predis\Command\ScriptCommand', [], '', true, true, true, ['parseResponse']); + $command + ->expects($this->once()) + ->method('getScript') + ->willReturn($luaScriptBody); + $command + ->expects($this->once()) + ->method('parseResponse') + ->with('OK') + ->willReturn(true); + + $connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock(); + $connection + ->expects($this->exactly(2)) + ->method('executeCommand') + ->withConsecutive( + [$command], + [$this->isRedisCommand('EVAL', [$luaScriptBody])] + ) + ->willReturnOnConsecutiveCalls( + new Response\Error('NOSCRIPT'), + 'OK' + ); + + $connection + ->expects($this->exactly(2)) + ->method('getParameters') + ->willReturn(new Parameters(['protocol' => 2])); + + $client = new Client($connection); + + $this->assertTrue($client->executeCommand($command)); + } + + /** + * @group disconnected + */ + public function testGetIteratorWithTraversableConnections(): void + { + $connection1 = $this->getMockConnection('tcp://127.0.0.1:6381'); + $connection2 = $this->getMockConnection('tcp://127.0.0.1:6382'); + $connection3 = $this->getMockConnection('tcp://127.0.0.1:6383'); + + $aggregate = new Connection\Cluster\PredisCluster(new Parameters()); + + $aggregate->add($connection1); + $aggregate->add($connection2); + $aggregate->add($connection3); + + $client = new Client($aggregate); + + /** @var Iterator */ + $iterator = $client->getIterator(); + + $this->assertInstanceOf('\Predis\Client', $nodeClient = $iterator->current()); + $this->assertSame($connection1, $nodeClient->getConnection()); + $this->assertSame('127.0.0.1:6381', $iterator->key()); + + $iterator->next(); + + $this->assertInstanceOf('\Predis\Client', $nodeClient = $iterator->current()); + $this->assertSame($connection2, $nodeClient->getConnection()); + $this->assertSame('127.0.0.1:6382', $iterator->key()); + + $iterator->next(); + + $this->assertInstanceOf('\Predis\Client', $nodeClient = $iterator->current()); + $this->assertSame($connection3, $nodeClient->getConnection()); + $this->assertSame('127.0.0.1:6383', $iterator->key()); + } + + /** + * @group disconnected + */ + public function testGetIteratorWithNonTraversableConnectionNoException(): void + { + $connection = $this->getMockConnection('tcp://127.0.0.1:6381'); + $client = new Client($connection); + + /** @var Iterator */ + $iterator = $client->getIterator(); + + $this->assertInstanceOf('\Predis\Client', $nodeClient = $iterator->current()); + $this->assertSame($connection, $nodeClient->getConnection()); + $this->assertSame('127.0.0.1:6381', $iterator->key()); + } + + /** + * @group disconnected + */ + public function testExecuteCommandRetryCommandOnRetryableException() + { + $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock(); + $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock(); + $parameters = new Parameters([ + 'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3), + ]); + + $mockStream + ->expects($this->atLeast(3)) + ->method('close') + ->withAnyParameters(); + + $mockStream + ->expects($this->exactly(4)) + ->method('write') + ->withAnyParameters() + ->willReturnOnConsecutiveCalls( + $this->throwException(new RuntimeException('', 2)), + $this->throwException(new RuntimeException('', 2)), + $this->throwException(new RuntimeException('', 2)), + 1000 + ); + + $mockStream + ->expects($this->once()) + ->method('read') + ->withAnyParameters() + ->willReturn("+PONG\r\n"); + + $mockStreamFactory + ->expects($this->exactly(4)) + ->method('createStream') + ->withAnyParameters() + ->willReturn($mockStream); + + $connection = new StreamConnection($parameters, $mockStreamFactory); + $client = new Client($connection); + $this->assertEquals('PONG', $client->ping()); + } + + /** + * @group connected + * @requiresRedisVersion >= 7.2.0 + */ + public function testSetClientInfoOnConnection(): void + { + $client = new Client($this->getParameters()); + $info = $client->client('LIST', null, $client->client('ID'))[0]; + + $this->assertSame('predis', $info['lib-name']); + $this->assertSame(Client::VERSION, $info['lib-ver']); + } + + /** + * @group connected + * @requiresRedisVersion >= 5.0.0 + */ + public function testClientsCreateDifferentPersistentConnections(): void + { + $client1 = new Client($this->getParameters(['database' => 14, 'persistent' => true, 'conn_uid' => 1])); + $client2 = new Client($this->getParameters(['database' => 15, 'persistent' => true, 'conn_uid' => 2])); + + $client1->set('foo', 'bar'); + $client2->set('foo', 'baz'); + + $this->assertSame('bar', $client1->get('foo')); + $this->assertSame('baz', $client2->get('foo')); + $this->assertNotSame($client1->client('ID'), $client2->client('ID')); + $client1->disconnect(); + $client2->disconnect(); + } + + /** + * @group connected + * @requiresRedisVersion >= 5.0.0 + */ + public function testClientsCreateSamePersistentConnections(): void + { + $client1 = new Client($this->getParameters(['persistent' => true])); + $client2 = new Client($this->getParameters(['persistent' => true])); + + $client1->set('foo', 'bar'); + $client2->set('foo', 'baz'); + + $this->assertSame('baz', $client2->get('foo')); + $this->assertSame($client1->client('ID'), $client2->client('ID')); + $client1->disconnect(); + } + + /** + * @group connected + * @group cluster + * @requiresRedisVersion >= 2.0.0 + */ + public function testClusterClientsCreateDifferentPersistentConnections(): void + { + $client1 = new Client( + $this->getDefaultParametersArray(), + ['cluster' => 'redis', 'parameters' => ['persistent' => true, 'conn_uid' => 1]] + ); + $client2 = new Client( + $this->getDefaultParametersArray(), + ['cluster' => 'redis', 'parameters' => ['persistent' => true, 'conn_uid' => 2]] + ); + + $client1->set('{shard1}foo', 'bar'); + $client2->set('{shard2}foo', 'baz'); + + $this->assertSame('bar', $client1->get('{shard1}foo')); + $this->assertSame('baz', $client2->get('{shard2}foo')); + $client1->disconnect(); + $client2->disconnect(); + } + + /** + * @group connected + * @group cluster + * @requiresRedisVersion >= 2.0.0 + * @requires PHP >= 7.4 + * @return void + */ + public function testCreatesClusterWithRelayConnection(): void + { + $client = new Client($this->getDefaultParametersArray(), ['cluster' => 'redis', 'connections' => 'relay']); + + $this->assertEquals('OK', $client->set('key', 'value')); + $this->assertSame('value', $client->get('key')); + } + + /** + * @group connected + * @group unprotected + * @requiresRedisVersion >= 2.0.0 + * @return void + */ + public function testClientAuthenticationAgainstUnprotectedServer(): void + { + $client = new Client($this->getParameters()); + + $this->assertEquals('OK', $client->set('key', 'value')); + $this->assertSame('value', $client->get('key')); + + // AUTH doesn't throw exception if no authentication requires. + $clientWithPassword = new Client($this->getParameters( + ['password' => getenv('REDIS_PASSWORD') ?: constant('REDIS_PASSWORD')]) + ); + $this->assertEquals('OK', $clientWithPassword->set('key', 'value')); + $this->assertSame('value', $clientWithPassword->get('key')); + + $this->assertEquals( + 'OK', + $client->acl->setUser( + 'test_user', + 'on', + '>foobar', + 'allcommands', + 'allkeys' + ) + ); + + $clientTestUser = new Client($this->getParameters( + ['username' => 'test_user', 'password' => 'foobar']) + ); + $this->assertEquals('test_user', $clientTestUser->acl->whoami()); + $this->assertEquals('OK', $clientTestUser->set('key', 'value')); + $this->assertSame('value', $clientTestUser->get('key')); + $this->assertEquals(1, $clientTestUser->acl->delUser('test_user')); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 7.0.0 + */ + public function testStandaloneNodeRetryCommandExecutionOnTimeoutException(): void + { + $retries = 0; + $mockDisconnect = static function () use (&$retries) { + $streamConnection = new StreamConnection(new Parameters([ + 'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3), + ])); + $disconnectFunc = [$streamConnection, 'disconnect']; + ++$retries; + $disconnectFunc(); + }; + + $stubConnection = $this->getMockBuilder(StreamConnection::class) + ->setConstructorArgs([new Parameters([ + 'retry' => new Retry(new ExponentialBackoff(100, 1000), 3), + 'read_write_timeout' => 0.1, + ])]) + ->onlyMethods(['disconnect']) + ->getMock(); + + $stubConnection + ->expects($this->exactly(7)) + ->method('disconnect') + ->willReturnCallback($mockDisconnect); + + $stubConnection->addConnectCommand(new RawCommand('auth', ['foobar'])); + + $client = new Client($stubConnection); + + $this->expectException(TimeoutException::class); + + $client->blmpop(3, ['random_key']); + $this->assertEquals(3, $retries); + } + + /** + * @group connected + * @group relay-incompatible + * @return void + * @requiresRedisVersion >= 7.0.0 + */ + public function testStandaloneNodeRetryCommandExecutionOnTimeoutExceptionIntegration(): void + { + // Retry used to wrap callback around, so we can count retries + $retry = new Retry(new ExponentialBackoff(100, 1000), 3); + $retriesCount = 0; + $retryWrapperFunc = static function (callable $do, ?callable $fail = null) use ($retry, &$retriesCount) { + $failWrapperFunc = static function (Exception $e) use (&$retriesCount, $fail) { + ++$retriesCount; + $fail($e); + }; + + return $retry->callWithRetry($do, $failWrapperFunc); + }; + + $mockRetry = $this->getMockBuilder(Retry::class) + ->setConstructorArgs([new ExponentialBackoff(100, 1000), 3]) + ->onlyMethods(['callWithRetry']) + ->getMock(); + + $mockRetry + ->expects($this->any()) + ->method('callWithRetry') + ->willReturnCallback($retryWrapperFunc); + + // Create a real connection with mocked retry and short read_write_timeout + $client = $this->createClient([ + 'retry' => $mockRetry, + 'read_write_timeout' => 0.1, + ]); + + $this->expectException(TimeoutException::class); + + try { + // blmpop with 3 second timeout will exceed the 0.1 second read_write_timeout + // causing TimeoutException to be thrown and retried 3 times before failing + $client->blmpop(3, ['random_key_that_does_not_exist']); + } finally { + $this->assertGreaterThanOrEqual(3, $retriesCount); + } + } + + /** + * @group connected + * @group cluster + * @return void + * @requiresRedisVersion >= 2.0.0 + */ + public function testClusterRetryCommandExecutionOnTimeoutException(): void + { + $defaultParams = $this->getDefaultParametersArray(); + $parsedParams = []; + + foreach ($defaultParams as $param) { + $parsedParam = Parameters::parse($param); + $parsedParam['retry'] = new Retry(new ExponentialBackoff(1000, 10000), 3); + $parsedParam['read_write_timeout'] = 0.1; + $parsedParams[] = Parameters::create($parsedParam); + } + + $retries = 0; + $mockDisconnect = static function () use (&$retries) { + $streamConnection = new StreamConnection(new Parameters([ + 'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3), + ])); + $disconnectFunc = [$streamConnection, 'disconnect']; + ++$retries; + $disconnectFunc(); + }; + + $stubConnection1 = $this->getMockBuilder(StreamConnection::class) + ->setConstructorArgs([$parsedParams[0]]) + ->onlyMethods(['disconnect']) + ->getMock(); + + $stubConnection1 + ->expects($this->any()) + ->method('disconnect') + ->willReturnCallback($mockDisconnect); + + $stubConnection1->addConnectCommand(new RawCommand('auth', [$parsedParams[0]->password])); + + $stubConnection2 = $this->getMockBuilder(StreamConnection::class) + ->setConstructorArgs([$parsedParams[1]]) + ->onlyMethods(['disconnect']) + ->getMock(); + + $stubConnection2 + ->expects($this->any()) + ->method('disconnect') + ->willReturnCallback($mockDisconnect); + + $stubConnection2->addConnectCommand(new RawCommand('auth', [$parsedParams[1]->password])); + + $stubConnection3 = $this->getMockBuilder(StreamConnection::class) + ->setConstructorArgs([$parsedParams[2]]) + ->onlyMethods(['disconnect']) + ->getMock(); + + $stubConnection3 + ->expects($this->any()) + ->method('disconnect') + ->willReturnCallback($mockDisconnect); + + $stubConnection3->addConnectCommand(new RawCommand('auth', [$parsedParams[2]->password])); + + $mockFactory = $this->getMockBuilder(Factory::class)->getMock(); + $clusterConnection = new RedisCluster($mockFactory, $parsedParams[0]); + + $clusterConnection->add($stubConnection1); + $clusterConnection->add($stubConnection2); + $clusterConnection->add($stubConnection3); + + $client = new Client($clusterConnection); + + $this->expectException(TimeoutException::class); + + $client->blpop(['random_key'], 3); + $this->assertEquals(3, $retries); + } + + /** + * @group connected + * @group sentinel + * @group relay-incompatible + * @requiresRedisVersion >= 7.0.0 + */ + public function testConnectToSentinelAndExecuteCommands(): void + { + $client = $this->createClient(); + + $this->assertEquals('PONG', $client->ping()); + + $client->set('sentinel_test_key', 'sentinel_test_value'); + $this->assertEquals('sentinel_test_value', $client->get('sentinel_test_key')); + } + + /** + * @group connected + * @group sentinel + * @group relay-incompatible + * @requiresRedisVersion >= 7.0.0 + */ + public function testSentinelReplicationConnection(): void + { + $client = $this->createClient(); + $connection = $client->getConnection(); + + $this->assertInstanceOf(SentinelReplication::class, $connection); + + // Verify we can get master connection + $master = $connection->getMaster(); + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $master); + + // Verify master is connected + $this->assertTrue($master->isConnected()); + } + + /** + * @group connected + * @group sentinel + * @group relay-incompatible + * @requiresRedisVersion >= 7.0.0 + */ + public function testSentinelReplicationConnectionWithResp3(): void + { + $client = $this->createClient(['protocol' => 3]); + $connection = $client->getConnection(); + + $this->assertInstanceOf(SentinelReplication::class, $connection); + + // SENTINEL slaves returns a map under RESP3 - calling getSlaves() + // exercises the parsing path that previously read fields by + // positional index. We do not assert on cardinality because the + // CI topology may not include replicas; we only require that the + // call completes without protocol-incompatibility errors. + $slaves = $connection->getSlaves(); + $this->assertIsArray($slaves); + + foreach ($slaves as $slave) { + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $slave); + } + + $client->set('sentinel_resp3_test_key', 'sentinel_resp3_test_value'); + $this->assertEquals('sentinel_resp3_test_value', $client->get('sentinel_resp3_test_key')); + } + + // ******************************************************************** // + // ---- HELPER METHODS ------------------------------------------------ // + // ******************************************************************** // + + /** + * Returns an URI string representation of the specified connection parameters. + * + * @param array $parameters Array of connection parameters. + * + * @return string URI string. + */ + protected function getParametersString(array $parameters): string + { + $defaults = $this->getDefaultParametersArray(); + + $scheme = $parameters['scheme'] ?? $defaults['scheme']; + $host = $parameters['host'] ?? $defaults['host']; + $port = $parameters['port'] ?? $defaults['port']; + + unset($parameters['scheme'], $parameters['host'], $parameters['port']); + $uriString = "$scheme://$host:$port/?"; + + foreach ($parameters as $k => $v) { + $uriString .= "$k=$v&"; + } + + return $uriString; + } + + /** + * Returns a mock object simulating an aggregate connection initializer. + * + * @param ParametersInterface|array|string $parameters Expected connection parameters + * + * @return callable|MockObject + */ + protected function getAggregateInitializer($parameters) + { + $connection = $this->getMockBuilder('Predis\Connection\AggregateConnectionInterface')->getMock(); + + $callable = $this->getMockBuilder('stdClass') + ->addMethods(['__invoke']) + ->getMock(); + $callable + ->expects($this->once()) + ->method('__invoke') + ->with( + $parameters, + $this->isInstanceOf('Predis\Configuration\OptionsInterface'), + $this->isInstanceOf('Predis\Configuration\OptionInterface') + ) + ->willReturn($connection); + + return $callable; + } +} diff --git a/tests/Predis/Connection/Replication/SentinelReplicationTest.php b/tests/Predis/Connection/Replication/SentinelReplicationTest.php new file mode 100644 index 0000000..159b2ae --- /dev/null +++ b/tests/Predis/Connection/Replication/SentinelReplicationTest.php @@ -0,0 +1,2427 @@ +getMockBuilder(Connection\FactoryInterface::class)->getMock(); + $factory->expects($this->never())->method('create'); + + $client = new Client([ + 'tcp://127.0.0.1:26379', + 'tcp://127.0.0.1:26380', + ], ['replication' => 'sentinel', 'service' => 'svc', 'connections' => $factory]); + + $iterator = $client->getIterator(); + + $this->assertCount(1, $iterator); + $this->assertSame('', $iterator->key()); + $this->assertSame($client->getConnection(), $iterator->current()->getConnection()); + $this->assertSame($client->getOptions(), $iterator->current()->getOptions()); + $this->assertFalse($client->isConnected()); + } + + /** + * @group disconnected + */ + public function testClientIterationKeepsReplicationConnectionWithKnownNodes(): void + { + $replication = $this->getReplicationConnection('svc', []); + $replication->add(new StreamConnection(Parameters::create('tcp://127.0.0.1:6379?role=master'))); + $replication->add(new StreamConnection(Parameters::create('tcp://127.0.0.1:6380?role=slave'))); + + $client = new Client($replication); + $iterator = $client->getIterator(); + + $this->assertCount(1, $iterator); + $this->assertSame('127.0.0.1:6379,127.0.0.1:6380', $iterator->key()); + $this->assertSame($replication, $iterator->current()->getConnection()); + $this->assertSame($client->getOptions(), $iterator->current()->getOptions()); + $this->assertFalse($client->isConnected()); + } + + /** + * @group disconnected + */ + public function testMethodGetSentinelConnectionThrowsExceptionOnEmptySentinelsPool(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('No sentinel server available for autodiscovery.'); + + $replication = $this->getReplicationConnection('svc', []); + $replication->getSentinelConnection(); + } + + /** + * @group disconnected + */ + public function testParametersForSentinelConnectionShouldUsePasswordForAuthentication(): void + { + $replication = $this->getReplicationConnection('svc', [ + 'tcp://127.0.0.1:5381?alias=sentinel1&password=secret', + ]); + + $parameters = $replication->getSentinelConnection()->getParameters()->toArray(); + + $this->assertArrayHasKey('password', $parameters, 'Parameter `passwords` was expected to exist in connection parameters'); + } + + /** + * @group disconnected + */ + public function testParametersForSentinelConnectionShouldIgnoreDatabase(): void + { + $replication = $this->getReplicationConnection('svc', [ + 'tcp://127.0.0.1:5381?role=sentinel&database=1&username=myusername&password=password', + ]); + + $parameters = $replication->getSentinelConnection()->getParameters()->toArray(); + + $this->assertArrayNotHasKey('database', $parameters, 'Parameter `database` was expected to not exist in connection parameters'); + $this->assertArrayHasKey('username', $parameters, 'Parameter `username` was expected to not exist in connection parameters'); + $this->assertArrayHasKey('password', $parameters, 'Parameter `password` was expected to not exist in connection parameters'); + } + + /** + * @group disconnected + */ + public function testParametersForSentinelConnectionHaveDefaultTimeout(): void + { + $replication = $this->getReplicationConnection('svc', [ + 'tcp://127.0.0.1:5381?role=sentinel', + ]); + + $parameters = $replication->getSentinelConnection()->getParameters()->toArray(); + + $this->assertArrayHasKey('timeout', $parameters); + $this->assertSame(0.100, $parameters['timeout']); + } + + /** + * @group disconnected + */ + public function testParametersForSentinelConnectionCanOverrideDefaultTimeout(): void + { + $replication = $this->getReplicationConnection('svc', [ + 'tcp://127.0.0.1:5381?role=sentinel&timeout=1', + ]); + + $parameters = $replication + ->getSentinelConnection() + ->getParameters() + ->toArray(); + + $this->assertArrayHasKey('timeout', $parameters); + $this->assertSame('1', $parameters['timeout']); + } + + /** + * @group disconnected + */ + public function testConnectionParametersInstanceForSentinelConnectionIsNotModified(): void + { + $originalParameters = Parameters::create( + 'tcp://127.0.0.1:5381?role=sentinel&database=1&password=secret' + ); + + $replication = $this->getReplicationConnection('svc', [$originalParameters]); + + $parameters = $replication + ->getSentinelConnection() + ->getParameters(); + + $this->assertSame($originalParameters, $parameters); + $this->assertNotNull($parameters->password); + $this->assertNotNull($parameters->database); + } + + /** + * @group disconnected + */ + public function testConnectionParametersInstanceForSentinelConnectionIsNotModifiedEmptyPassword(): void + { + $sentinel1 = Parameters::create('tcp://127.0.0.1:5381?role=sentinel&database=1&password='); + $sentinel2 = Parameters::create('tcp://127.0.0.1:5381?role=sentinel&database=1'); + + $replication1 = $this->getReplicationConnection('svc', [$sentinel1]); + $replication2 = $this->getReplicationConnection('svc', [$sentinel2]); + + $parameters1 = $replication1->getSentinelConnection()->getParameters(); + $parameters2 = $replication2->getSentinelConnection()->getParameters(); + + $this->assertSame($sentinel1, $parameters1); + $this->assertSame($sentinel2, $parameters2); + + $this->assertNull($parameters1->password); + $this->assertNull($parameters2->password); + + $this->assertNotNull($parameters1->database); + $this->assertNotNull($parameters2->database); + } + + /** + * @group disconnected + */ + public function testMethodGetSentinelConnectionReturnsFirstAvailableSentinel(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel&alias=sentinel1'); + $sentinel2 = $this->getMockSentinelConnection('tcp://127.0.0.1:5382?role=sentinel&alias=sentinel2'); + $sentinel3 = $this->getMockSentinelConnection('tcp://127.0.0.1:5383?role=sentinel&alias=sentinel3'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1, $sentinel2, $sentinel3]); + + $this->assertSame($sentinel1, $replication->getSentinelConnection()); + } + + /** + * @group disconnected + */ + public function testMethodAddAttachesMasterOrSlaveNodesToReplication(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + $replication->add($slave2); + + $this->assertSame($master, $replication->getConnectionById('127.0.0.1:6381')); + $this->assertSame($slave1, $replication->getConnectionById('127.0.0.1:6382')); + $this->assertSame($slave2, $replication->getConnectionById('127.0.0.1:6383')); + + $this->assertSame($master, $replication->getMaster()); + $this->assertSame([$slave1, $slave2], $replication->getSlaves()); + } + + /** + * @group disconnected + * @FIXME + */ + public function testMethodRemoveDismissesMasterOrSlaveNodesFromReplication(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + $replication->add($slave2); + + $this->assertTrue($replication->remove($slave1)); + $this->assertTrue($replication->remove($sentinel1)); + + $this->assertSame('127.0.0.1:6381', (string) $replication->getMaster()); + $this->assertCount(1, $slaves = $replication->getSlaves()); + $this->assertSame('127.0.0.1:6383', (string) $slaves[0]); + } + + /** + * @group disconnected + */ + public function testMethodGetConnectionByIdOnEmptyReplication(): void + { + $replication = $this->getReplicationConnection('svc', []); + + $this->assertNull($replication->getConnectionById('127.0.0.1:6381')); + } + + /** + * @group disconnected + */ + public function testMethodGetConnectionByRole(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + + $replication = $this->getReplicationConnection('svc', []); + + $replication->add($master); + $replication->add($slave1); + $replication->add($sentinel1); + + $this->assertSame($sentinel1, $replication->getConnectionByRole('sentinel')); + $this->assertSame($master, $replication->getConnectionByRole('master')); + $this->assertSame($slave1, $replication->getConnectionByRole('slave')); + } + + /** + * @group disconnected + */ + public function testMethodGetConnectionByRoleOnEmptyReplicationForcesSentinelQueries(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->exactly(2)) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['get-master-addr-by-name', 'svc'])], + [$this->isRedisCommand('SENTINEL', ['slaves', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + // SENTINEL get-master-addr-by-name svc + ['127.0.0.1', '6381'], + + // SENTINEL slaves svc + [ + [ + 'name', '127.0.0.1:6382', + 'ip', '127.0.0.1', + 'port', '6382', + 'runid', '112cdebd22924a7d962be496f3a1c4c7c9bad93f', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $this->assertSame($sentinel1, $replication->getConnectionByRole('sentinel')); + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $replication->getConnectionByRole('master')); + $this->assertInstanceOf('Predis\Connection\NodeConnectionInterface', $replication->getConnectionByRole('slave')); + } + + /** + * @group disconnected + */ + public function testMethodGetConnectionByRoleUnknown(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + + $replication = $this->getReplicationConnection('svc', []); + + $replication->add($master); + $replication->add($slave1); + $replication->add($sentinel1); + + $this->assertNull($replication->getConnectionByRole('unknown')); + } + + /** + * @group disconnected + */ + public function testMethodUpdateSentinelsFetchesSentinelNodes(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['sentinels', 'svc'] + )) + ->willReturn( + [ + [ + 'name', '127.0.0.1:5382', + 'ip', '127.0.0.1', + 'port', '5382', + 'runid', 'a113aa7a0d4870a85bb22b4b605fd26eb93ed40e', + 'flags', 'sentinel', + ], + [ + 'name', '127.0.0.1:5383', + 'ip', '127.0.0.1', + 'port', '5383', + 'runid', 'f53b52d281be5cdd4873700c94846af8dbe47209', + 'flags', 'sentinel', + ], + ] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + $replication->updateSentinels(); + + // TODO: sorry for the smell... + $reflection = new ReflectionProperty($replication, 'sentinels'); + $reflection->setAccessible(true); + + $expected = [ + ['host' => '127.0.0.1', 'port' => '5381'], + ['host' => '127.0.0.1', 'port' => '5382'], + ['host' => '127.0.0.1', 'port' => '5383'], + ]; + + $this->assertSame($sentinel1, $replication->getSentinelConnection()); + $this->assertSame($expected, array_intersect_key($expected, $reflection->getValue($replication))); + } + + /** + * @group disconnected + */ + public function testMethodUpdateSentinelsRemovesCurrentSentinelAndRetriesNextOneOnFailure(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel&alias=sentinel1'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['sentinels', 'svc'] + )) + ->willThrowException( + new Connection\ConnectionException($sentinel1, 'Unknown connection error [127.0.0.1:5381]') + ); + + $sentinel2 = $this->getMockSentinelConnection('tcp://127.0.0.1:5382?role=sentinel&alias=sentinel2'); + $sentinel2 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['sentinels', 'svc'] + )) + ->willReturn( + [ + [ + 'name', '127.0.0.1:5383', + 'ip', '127.0.0.1', + 'port', '5383', + 'runid', 'f53b52d281be5cdd4873700c94846af8dbe47209', + 'flags', 'sentinel', + ], + ] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1, $sentinel2]); + $replication->updateSentinels(); + + // TODO: sorry for the smell... + $reflection = new ReflectionProperty($replication, 'sentinels'); + $reflection->setAccessible(true); + + $expected = [ + ['host' => '127.0.0.1', 'port' => '5382'], + ['host' => '127.0.0.1', 'port' => '5383'], + ]; + + $this->assertSame($sentinel2, $replication->getSentinelConnection()); + $this->assertSame($expected, array_intersect_key($expected, $reflection->getValue($replication))); + } + + /** + * @group disconnected + */ + public function testMethodUpdateSentinelsThrowsExceptionOnNoAvailableSentinel(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('No sentinel server available for autodiscovery.'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['sentinels', 'svc'] + )) + ->willThrowException( + new Connection\ConnectionException($sentinel1, 'Unknown connection error [127.0.0.1:5381]') + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + $replication->updateSentinels(); + } + + /** + * @group disconnected + */ + public function testMethodQuerySentinelFetchesMasterNodeSlaveNodesAndSentinelNodes(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel&alias=sentinel1'); + $sentinel1 + ->expects($this->exactly(3)) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['sentinels', 'svc'])], + [$this->isRedisCommand('SENTINEL', ['get-master-addr-by-name', 'svc'])], + [$this->isRedisCommand('SENTINEL', ['slaves', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + // SENTINEL sentinels svc + [ + [ + 'name', '127.0.0.1:5382', + 'ip', '127.0.0.1', + 'port', '5382', + 'runid', 'a113aa7a0d4870a85bb22b4b605fd26eb93ed40e', + 'flags', 'sentinel', + ], + ], + + // SENTINEL get-master-addr-by-name svc + ['127.0.0.1', '6381'], + + // SENTINEL slaves svc + [ + [ + 'name', '127.0.0.1:6382', + 'ip', '127.0.0.1', + 'port', '6382', + 'runid', '112cdebd22924a7d962be496f3a1c4c7c9bad93f', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + [ + 'name', '127.0.0.1:6383', + 'ip', '127.0.0.1', + 'port', '6383', + 'runid', '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $sentinel2 = $this->getMockSentinelConnection('tcp://127.0.0.1:5382?role=sentinel&alias=sentinel2'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + $replication->querySentinel(); + + // TODO: sorry for the smell... + $reflection = new ReflectionProperty($replication, 'sentinels'); + $reflection->setAccessible(true); + + $sentinels = [ + ['host' => '127.0.0.1', 'port' => '5381'], + ['host' => '127.0.0.1', 'port' => '5382'], + ]; + + $this->assertSame($sentinel1, $replication->getSentinelConnection()); + $this->assertSame($sentinels, array_intersect_key($sentinels, $reflection->getValue($replication))); + + $master = $replication->getMaster(); + $slaves = $replication->getSlaves(); + + $this->assertSame('127.0.0.1:6381', (string) $master); + + $this->assertCount(2, $slaves); + $this->assertSame('127.0.0.1:6382', (string) $slaves[0]); + $this->assertSame('127.0.0.1:6383', (string) $slaves[1]); + } + + /** + * @group disconnected + */ + public function testMethodGetMasterAsksSentinelForMasterOnMasterNotSet(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['get-master-addr-by-name', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + ['127.0.0.1', '6381'] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $this->assertSame('127.0.0.1:6381', (string) $replication->getMaster()); + } + + /** + * @group disconnected + */ + public function testMethodGetMasterThrowsExceptionOnNoAvailableSentinels(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('No sentinel server available for autodiscovery.'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willThrowException( + new Connection\ConnectionException($sentinel1, 'Unknown connection error [127.0.0.1:5381]') + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->getMaster(); + } + + /** + * @group disconnected + */ + public function testMethodGetSlavesOnEmptySlavePoolAsksSentinelForSlaves(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['slaves', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + [ + [ + 'name', '127.0.0.1:6382', + 'ip', '127.0.0.1', + 'port', '6382', + 'runid', '112cdebd22924a7d962be496f3a1c4c7c9bad93f', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + [ + 'name', '127.0.0.1:6383', + 'ip', '127.0.0.1', + 'port', '6383', + 'runid', '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $slaves = $replication->getSlaves(); + + $this->assertSame('127.0.0.1:6382', (string) $slaves[0]); + $this->assertSame('127.0.0.1:6383', (string) $slaves[1]); + } + + /** + * @group disconnected + */ + public function testMethodGetSlavesParsesResp3MapResponse(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['slaves', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + [ + [ + 'name' => '127.0.0.1:6382', + 'ip' => '127.0.0.1', + 'port' => '6382', + 'runid' => '112cdebd22924a7d962be496f3a1c4c7c9bad93f', + 'flags' => 'slave', + 'master-host' => '127.0.0.1', + 'master-port' => '6381', + 'master-link-status' => 'ok', + ], + [ + 'name' => '127.0.0.1:6383', + 'ip' => '127.0.0.1', + 'port' => '6383', + 'runid' => '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags' => 's_down,slave', + 'master-host' => '127.0.0.1', + 'master-port' => '6381', + 'master-link-status' => 'ok', + ], + [ + 'name' => '127.0.0.1:6384', + 'ip' => '127.0.0.1', + 'port' => '6384', + 'runid' => '8c0bf1291797fbc5608c07a17da394147dc62818', + 'flags' => 'slave', + 'master-host' => '127.0.0.1', + 'master-port' => '6381', + 'master-link-status' => 'err', + ], + ] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $slaves = $replication->getSlaves(); + + $this->assertCount(1, $slaves); + $this->assertSame('127.0.0.1:6382', (string) $slaves[0]); + } + + /** + * @group disconnected + */ + public function testMethodGetSlavesThrowsExceptionOnNoAvailableSentinels(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('No sentinel server available for autodiscovery.'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['slaves', 'svc'] + )) + ->willThrowException( + new Connection\ConnectionException($sentinel1, 'Unknown connection error [127.0.0.1:5381]') + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->getSlaves(); + } + + /** + * @group disconnected + */ + public function testMethodConnectThrowsExceptionOnConnectWithEmptySentinelsPool(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('No sentinel server available for autodiscovery.'); + + $replication = $this->getReplicationConnection('svc', []); + $replication->connect(); + } + + /** + * @group disconnected + */ + public function testMethodConnectForcesConnectionToSlave(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->never()) + ->method('connect'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->once()) + ->method('connect'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $replication->connect(); + } + + /** + * @group disconnected + */ + public function testMethodConnectOnEmptySlavePoolAsksSentinelForSlavesAndForcesConnectionToSlave(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['slaves', 'svc'] + )) + ->willReturn( + [ + [ + 'name', '127.0.0.1:6382', + 'ip', '127.0.0.1', + 'port', '6382', + 'runid', '112cdebd22924a7d962be496f3a1c4c7c9bad93f', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->never()) + ->method('connect'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->once()) + ->method('connect'); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6382', + 'role' => 'slave', + ]) + ->willReturn($slave1); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->add($master); + + $replication->connect(); + } + + /** + * @group disconnected + */ + public function testMethodConnectOnEmptySlavePoolAsksSentinelForSlavesAndForcesConnectionToMasterIfStillEmpty(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->exactly(2)) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['slaves', 'svc'])], + [$this->isRedisCommand('SENTINEL', ['get-master-addr-by-name', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + [], + ['127.0.0.1', '6381'] + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->once()) + ->method('connect'); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6381', + 'role' => 'master', + ]) + ->willReturn($master); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->connect(); + } + + /** + * @group disconnected + */ + public function testMethodDisconnectForcesDisconnectionOnAllConnectionsInPool(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->never()) + ->method('disconnect'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->once()) + ->method('disconnect'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->once()) + ->method('disconnect'); + + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + $slave2 + ->expects($this->once()) + ->method('disconnect'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + $replication->add($slave2); + + $replication->disconnect(); + } + + /** + * @group disconnected + */ + public function testMethodIsConnectedReturnConnectionStatusOfCurrentConnection(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->exactly(2)) + ->method('isConnected') + ->willReturnOnConsecutiveCalls(true, false); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($slave1); + + $this->assertFalse($replication->isConnected()); + $replication->connect(); + $this->assertTrue($replication->isConnected()); + $replication->getConnectionById('127.0.0.1:6382')->disconnect(); + $this->assertFalse($replication->isConnected()); + } + + /** + * @group disconnected + */ + public function testMethodGetConnectionByIdReturnsConnectionWhenFound(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame($master, $replication->getConnectionById('127.0.0.1:6381')); + $this->assertSame($slave1, $replication->getConnectionById('127.0.0.1:6382')); + $this->assertNull($replication->getConnectionById('127.0.0.1:6383')); + } + + /** + * @group disconnected + */ + public function testMethodSwitchToSelectsCurrentConnection(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->once()) + ->method('connect'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->never()) + ->method('connect'); + + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave2 + ->expects($this->once()) + ->method('connect'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + $replication->add($slave2); + + $replication->switchTo($master); + $this->assertSame($master, $replication->getCurrent()); + + $replication->switchTo($slave2); + $this->assertSame($slave2, $replication->getCurrent()); + } + + /** + * @group disconnected + */ + public function testMethodSwitchToThrowsExceptionOnConnectionNotFound(): void + { + $this->expectException('InvalidArgumentException'); + $this->expectExceptionMessage('Invalid connection or connection not found.'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $replication->switchTo($slave2); + } + + /** + * @group disconnected + */ + public function testMethodSwitchToMasterSelectsCurrentConnectionToMaster(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->once()) + ->method('connect'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->never()) + ->method('connect'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $replication->switchToMaster(); + + $this->assertSame($master, $replication->getCurrent()); + } + + /** + * @group disconnected + */ + public function testMethodSwitchToSlaveSelectsCurrentConnectionToRandomSlave(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->never()) + ->method('connect'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->once()) + ->method('connect'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $replication->switchToSlave(); + + $this->assertSame($slave1, $replication->getCurrent()); + } + + /** + * @group disconnected + */ + public function testGetConnectionByCommandReturnsMasterForWriteCommands(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->exactly(2)) + ->method('isConnected') + ->willReturnOnConsecutiveCalls(false, true); + $master + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('ROLE')] + ) + ->willReturnOnConsecutiveCalls( + ['master', 3129659, [['127.0.0.1', 6382, 3129242]]] + ); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame($master, $replication->getConnectionByCommand( + Command\RawCommand::create('set', 'key', 'value') + )); + + $this->assertSame($master, $replication->getConnectionByCommand( + Command\RawCommand::create('del', 'key') + )); + } + + /** + * @group disconnected + */ + public function testGetConnectionByCommandReturnsSlaveForReadOnlyCommands(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->exactly(2)) + ->method('isConnected') + ->willReturnOnConsecutiveCalls(false, true); + $slave1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('ROLE')] + ) + ->willReturnOnConsecutiveCalls( + ['slave', '127.0.0.1', 9000, 'connected', 3167038] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame($slave1, $replication->getConnectionByCommand( + Command\RawCommand::create('get', 'key') + )); + + $this->assertSame($slave1, $replication->getConnectionByCommand( + Command\RawCommand::create('exists', 'key') + )); + } + + /** + * @group disconnected + */ + public function testGetConnectionByCommandSwitchesToMasterAfterWriteCommand(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->exactly(2)) + ->method('isConnected') + ->willReturnOnConsecutiveCalls(false, true); + $master + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('ROLE')] + ) + ->willReturnOnConsecutiveCalls( + ['master', 3129659, [['127.0.0.1', 6382, 3129242]]] + ); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->once()) + ->method('isConnected') + ->willReturnOnConsecutiveCalls(false); + $slave1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('ROLE')] + ) + ->willReturnOnConsecutiveCalls( + ['slave', '127.0.0.1', 9000, 'connected', 3167038] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame($slave1, $replication->getConnectionByCommand( + Command\RawCommand::create('exists', 'key') + )); + + $this->assertSame($master, $replication->getConnectionByCommand( + Command\RawCommand::create('set', 'key', 'value') + )); + + $this->assertSame($master, $replication->getConnectionByCommand( + Command\RawCommand::create('get', 'key') + )); + } + + /** + * @group disconnected + */ + public function testGetConnectionByCommandThrowsExceptionOnNodeRoleMismatch(): void + { + $this->expectException('Predis\Replication\RoleException'); + $this->expectExceptionMessage('Expected master but got slave [127.0.0.1:6381]'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->once()) + ->method('isConnected') + ->willReturn(false); + $master + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('ROLE')] + ) + ->willReturnOnConsecutiveCalls( + ['slave', '127.0.0.1', 9000, 'connected', 3167038] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + + $replication->getConnectionByCommand(Command\RawCommand::create('del', 'key')); + } + + /** + * @group disconnected + */ + public function testGetConnectionByCommandReturnsMasterForReadOnlyOperationsOnUnavailableSlaves(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['slaves', 'svc'])] + ) + ->willReturnOnConsecutiveCalls( + [ + [ + 'name', '127.0.0.1:6382', + 'ip', '127.0.0.1', + 'port', '6382', + 'runid', '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags', 'slave,s_down,disconnected', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->once()) + ->method('isConnected') + ->willReturn(false); + $master + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('ROLE')] + ) + ->willReturnOnConsecutiveCalls( + ['master', '0', []] + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + + $replication->getConnectionByCommand(Command\RawCommand::create('get', 'key')); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandSendsCommandToNodeAndReturnsResponse(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $cmdGet = Command\RawCommand::create('get', 'key'); + $cmdGetResponse = 'value'; + + $cmdSet = Command\RawCommand::create('set', 'key', 'value'); + $cmdSetResponse = Response\Status::get('OK'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $master + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SET', ['key', $cmdGetResponse])] + ) + ->willReturnOnConsecutiveCalls( + $cmdSetResponse + ); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $slave1 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('GET', ['key'])] + ) + ->willReturnOnConsecutiveCalls( + $cmdGetResponse + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame($cmdGetResponse, $replication->executeCommand($cmdGet)); + $this->assertSame($cmdSetResponse, $replication->executeCommand($cmdSet)); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandRetriesReadOnlyCommandOnNextSlaveOnConnectionException(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['slaves', 'svc'] + )) + ->willReturn( + [ + [ + 'name', '127.0.0.1:6383', + 'ip', '127.0.0.1', + 'port', '6383', + 'runid', '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + + $slave1 + ->expects($this->any()) + ->method('disconnect'); + + $slave1 + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('GET', ['key']) + ) + ->willThrowException( + new Connection\ConnectionException($slave1, 'Unknown connection error [127.0.0.1:6382]') + ); + + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + $slave2 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $slave2 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('GET', ['key'])] + ) + ->willReturnOnConsecutiveCalls( + 'value' + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6383', + 'role' => 'slave', + ]) + ->willReturn($slave2); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->add($master); + $replication->add($slave1); + + self::assertSame('value', $replication->executeCommand( + Command\RawCommand::create('get', 'key') + )); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandRetriesReadOnlyCommandOnNextSlaveOnFailure(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['slaves', 'svc'] + )) + ->willReturn( + [ + [ + 'name', '127.0.0.1:6383', + 'ip', '127.0.0.1', + 'port', '6383', + 'runid', '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $slave1 + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('GET', ['key']) + ) + ->willThrowException( + new Connection\ConnectionException($slave1, 'message') + ); + + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + $slave2 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $slave2 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('GET', ['key'])] + ) + ->willReturnOnConsecutiveCalls( + 'value' + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6383', + 'role' => 'slave', + ]) + ->willReturn($slave2); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame('value', $replication->executeCommand( + Command\RawCommand::create('get', 'key') + )); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandRetriesWriteCommandOnNewMasterOnFailure(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willReturn( + ['127.0.0.1', '6391'] + ); + + $masterOld = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $masterOld + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterOld + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('DEL', ['key']) + ) + ->willThrowException( + new Connection\ConnectionException($masterOld, 'Unknown connection error [127.0.0.1:6381]') + ); + + $masterNew = $this->getMockConnection('tcp://127.0.0.1:6391?role=master'); + $masterNew + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterNew + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('DEL', ['key'])] + ) + ->willReturnOnConsecutiveCalls( + 1 + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6391', + 'role' => 'master', + ]) + ->willReturn($masterNew); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->add($masterOld); + + $this->assertSame(1, $replication->executeCommand( + Command\RawCommand::create('del', 'key') + )); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandRetriesWriteCommandOnNewMasterOnStreamInitException(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willReturn( + ['127.0.0.1', '6391'] + ); + + $masterOld = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $masterOld + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterOld + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('DEL', ['key']) + ) + ->willThrowException( + new StreamInitException('Connection refused [tcp://127.0.0.1:6381]') + ); + + $masterNew = $this->getMockConnection('tcp://127.0.0.1:6391?role=master'); + $masterNew + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterNew + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('DEL', ['key'])] + ) + ->willReturnOnConsecutiveCalls( + 1 + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6391', + 'role' => 'master', + ]) + ->willReturn($masterNew); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->add($masterOld); + + $this->assertSame(1, $replication->executeCommand( + Command\RawCommand::create('del', 'key') + )); + } + + /** + * When every retry also fails with a StreamInitException, the original + * transport error must be surfaced unchanged - not masked by a TypeError in + * the retry fail callback (regression test for issue #1713). + * + * @group disconnected + */ + public function testMethodExecuteCommandSurfacesStreamInitExceptionWhenAllRetriesFail(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willReturn( + ['127.0.0.1', '6391'] + ); + + $masterOld = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $masterOld + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterOld + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('DEL', ['key'])) + ->willThrowException( + new StreamInitException('Connection refused [tcp://127.0.0.1:6381]') + ); + + $masterNew = $this->getMockConnection('tcp://127.0.0.1:6391?role=master'); + $masterNew + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterNew + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('DEL', ['key'])) + ->willThrowException( + new StreamInitException('Connection refused [tcp://127.0.0.1:6391]') + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6391', + 'role' => 'master', + ]) + ->willReturn($masterNew); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + $replication->add($masterOld); + $replication->setRetryLimit(1); + + // The fail callback is invoked with the StreamInitException on the first + // attempt (it must not TypeError), then the second attempt exhausts the + // retry budget and the original StreamInitException is re-thrown. + $this->expectException(StreamInitException::class); + $this->expectExceptionMessage('Connection refused [tcp://127.0.0.1:6391]'); + + $replication->executeCommand(Command\RawCommand::create('del', 'key')); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandThrowsExceptionOnUnknownServiceName(): void + { + $this->expectException('Predis\Response\ServerException'); + $this->expectExceptionMessage('ERR No such master with that name'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willReturn(null); + + $masterOld = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $masterOld + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $masterOld + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('DEL', ['key']) + ) + ->willThrowException( + new Connection\ConnectionException($masterOld, 'Unknown connection error [127.0.0.1:6381]') + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($masterOld); + + $replication->executeCommand( + Command\RawCommand::create('del', 'key') + ); + } + + /** + * @group disconnected + */ + public function testMethodExecuteCommandThrowsExceptionOnConnectionFailureAndNoAvailableSentinels(): void + { + $this->expectException('Predis\ClientException'); + $this->expectExceptionMessage('No sentinel server available for autodiscovery.'); + + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['get-master-addr-by-name', 'svc'] + )) + ->willThrowException( + new Connection\ConnectionException($sentinel1, 'Unknown connection error [127.0.0.1:5381]') + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $master + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('DEL', ['key']) + ) + ->willThrowException( + new Connection\ConnectionException($master, 'Unknown connection error [127.0.0.1:6381]') + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1]); + + $replication->add($master); + + $replication->executeCommand( + Command\RawCommand::create('del', 'key') + ); + } + + /** + * @group disconnected + */ + public function testMethodGetReplicationStrategyReturnsInstance(): void + { + $strategy = new Replication\ReplicationStrategy(); + $factory = new Connection\Factory(); + + $replication = new SentinelReplication( + 'svc', ['tcp://127.0.0.1:5381?role=sentinel'], $factory, $strategy + ); + + $this->assertSame($strategy, $replication->getReplicationStrategy()); + } + + /** + * @group disconnected + */ + public function testMethodSerializeCanSerializeWholeObject(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + + $strategy = new Replication\ReplicationStrategy(); + $factory = new Connection\Factory(); + + $replication = new SentinelReplication('svc', [$sentinel1], $factory, $strategy); + + $replication->add($master); + $replication->add($slave1); + $replication->add($slave2); + + $unserialized = unserialize(serialize($replication)); + + $this->assertEquals($master, $unserialized->getConnectionById('127.0.0.1:6381')); + $this->assertEquals($slave1, $unserialized->getConnectionById('127.0.0.1:6382')); + $this->assertEquals($master, $unserialized->getConnectionById('127.0.0.1:6383')); + $this->assertEquals($strategy, $unserialized->getReplicationStrategy()); + } + + /** + * @group disconnected + */ + public function testMethodGetSentinelConnectionAfterSentinelRestart(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel&alias=sentinel1'); + $sentinel1 + ->expects($this->exactly(2)) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['sentinels', 'svc'] + )) + ->willReturnOnConsecutiveCalls( + $this->throwException(new Connection\ConnectionException($sentinel1, 'Unknown connection error [127.0.0.1:5381]')), + [ + [ + 'name', '127.0.0.1:5382', + 'ip', '127.0.0.1', + 'port', '5382', + 'runid', 'f53b52d281be5cdd4873700c94846af8dbe47209', + 'flags', 'sentinel', + ], + ] + ); + + $sentinel2 = $this->getMockSentinelConnection('tcp://127.0.0.1:5382?role=sentinel&alias=sentinel2'); + $sentinel2 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['sentinels', 'svc'] + )) + ->willThrowException( + new Connection\ConnectionException($sentinel2, 'Unknown connection error [127.0.0.1:5382]') + ); + + $replication = $this->getReplicationConnection('svc', [$sentinel1, $sentinel2]); + try { + $replication->updateSentinels(); + } catch (\Predis\ClientException $exception) { + $this->assertEquals('No sentinel server available for autodiscovery.', $exception->getMessage()); + } + + $replication->updateSentinels(); + } + + /** + * @dataProvider connectionsProvider + * @group disconnected + */ + public function testGetParameters(string $connection): void + { + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $connection = $this->getMockConnection($connection); + + $replication = $this->getReplicationConnection('srv', [$sentinel]); + $replication->add($connection); + + $this->assertSame($connection->getParameters(), $replication->getParameters()); + } + + /** + * @group disconnected + * @return void + */ + public function testGetParametersReturnsSentinelParametersOnEmptyConnectionPool(): void + { + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $replication = $this->getReplicationConnection('srv', [$sentinel]); + + $this->assertSame($sentinel->getParameters(), $replication->getParameters()); + } + + /** + * @group disconnected + * @return void + */ + public function testGetParametersReturnsParametersFromArrayAfterUpdateSentinels(): void + { + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('SENTINEL', ['sentinels', 'srv'])) + ->willReturn([ + [ + 'name', '127.0.0.1:5382', + 'ip', '127.0.0.1', + 'port', '5382', + 'runid', 'a113aa7a0d4870a85bb22b4b605fd26eb93ed40e', + 'flags', 'sentinel', + ], + ]); + + $replication = $this->getReplicationConnection('srv', [$sentinel]); + $replication->updateSentinels(); + + // After updateSentinels(), the sentinels array contains plain arrays + $parameters = $replication->getParameters(); + + $this->assertInstanceOf(Parameters::class, $parameters); + $this->assertEquals('127.0.0.1', $parameters->host); + $this->assertEquals('5381', $parameters->port); + } + + /** + * @group disconnected + * @return void + */ + public function testGetParametersReturnsParametersFromArrayAfterQuerySentinel(): void + { + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel + ->expects($this->exactly(3)) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('SENTINEL', ['sentinels', 'srv'])], + [$this->isRedisCommand('SENTINEL', ['get-master-addr-by-name', 'srv'])], + [$this->isRedisCommand('SENTINEL', ['slaves', 'srv'])] + ) + ->willReturnOnConsecutiveCalls( + // SENTINEL sentinels srv + [ + [ + 'name', '127.0.0.1:5382', + 'ip', '127.0.0.1', + 'port', '5382', + 'runid', 'a113aa7a0d4870a85bb22b4b605fd26eb93ed40e', + 'flags', 'sentinel', + ], + ], + // SENTINEL get-master-addr-by-name srv + ['127.0.0.1', '6381'], + // SENTINEL slaves srv + [] + ); + + $replication = $this->getReplicationConnection('srv', [$sentinel]); + + // Clear master and slaves to test sentinel array scenario + $replication->querySentinel(); + + // Remove master to force getParameters to use sentinels array + $reflectionMaster = new ReflectionProperty($replication, 'master'); + $reflectionMaster->setAccessible(true); + $reflectionMaster->setValue($replication, null); + + $reflectionSlaves = new ReflectionProperty($replication, 'slaves'); + $reflectionSlaves->setAccessible(true); + $reflectionSlaves->setValue($replication, []); + + // After querySentinel(), the sentinels array contains plain arrays + $parameters = $replication->getParameters(); + + $this->assertInstanceOf(Parameters::class, $parameters); + $this->assertEquals('127.0.0.1', $parameters->host); + } + + /** + * @group disconnected + * @return void + */ + public function testGetParametersHandlesParametersInterfaceInSentinelsArray(): void + { + $parametersObject = Parameters::create('tcp://127.0.0.1:5381?role=sentinel'); + + $replication = $this->getReplicationConnection('srv', [$parametersObject]); + + $parameters = $replication->getParameters(); + + $this->assertSame($parametersObject, $parameters); + } + + /** + * @group disconnected + * @return void + */ + public function testGetParametersReturnsNullWhenNoConnectionsAvailable(): void + { + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + $replication = $this->getReplicationConnection('srv', [$sentinel]); + + // Clear all connections + $reflectionSentinels = new ReflectionProperty($replication, 'sentinels'); + $reflectionSentinels->setAccessible(true); + $reflectionSentinels->setValue($replication, []); + + $parameters = $replication->getParameters(); + + $this->assertNull($parameters); + } + + /** + * @group disconnected + * @return void + */ + public function testGetParametersHandlesStringUriInSentinelsArray(): void + { + // Test with string URI (e.g., "tcp://127.0.0.1:26379") + $sentinelUri = 'tcp://127.0.0.1:5381?role=sentinel'; + + $replication = $this->getReplicationConnection('srv', [$sentinelUri]); + + $parameters = $replication->getParameters(); + + $this->assertInstanceOf(ParametersInterface::class, $parameters); + $this->assertSame('127.0.0.1', $parameters->host); + $this->assertSame(5381, $parameters->port); + } + + /** + * @group disconnected + */ + public function testWrite(): void + { + $command1 = new Command\Redis\Search\FTSEARCH(); + $command1->setArguments(['arg1', '*']); + + $command2 = new Command\Redis\Search\FTSEARCH(); + $command2->setArguments(['arg2', '*']); + + $command3 = new Command\Redis\Search\FTSEARCH(); + $command3->setArguments(['arg3', '*']); + + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $master = $this->getMockConnection('tcp://127.0.0.1:6379?role=master'); + $slave = $this->getMockConnection('tcp://127.0.0.1:6380?role=slave'); + $strategy = new Replication\ReplicationStrategy(); + $factory = new Connection\Factory(); + + $master + ->expects($this->exactly(3)) + ->method('isConnected') + ->willReturn(true); + + $slave + ->expects($this->never()) + ->method('write'); + + $master + ->expects($this->exactly(3)) + ->method('write') + ->withConsecutive( + [$command1->serializeCommand()], + [$command2->serializeCommand()], + [$command3->serializeCommand()] + ); + + $replication = new SentinelReplication('svc', [$sentinel], $factory, $strategy); + + $replication->add($master); + $replication->add($slave); + + $replication->write($command1->serializeCommand() . $command2->serializeCommand() . $command3->serializeCommand()); + } + + /** + * Regression guard for GHSA-w6f5-v2h6-g786 (CWE-93): a CRLF embedded in a + * bulk string's own value must not be mistaken for a command boundary + * (splitting one command into a smuggled extra command). + * + * @group disconnected + */ + public function testWriteHandlesCRLFEmbeddedInBulkStringValue(): void + { + $command = new Command\Redis\SET(); + $command->setArguments(['victim-key', "PAD\r\n*1\r\n\$7\r\nFLUSHDB"]); + + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $master = $this->getMockConnection('tcp://127.0.0.1:6379?role=master'); + $slave = $this->getMockConnection('tcp://127.0.0.1:6380?role=slave'); + $strategy = new Replication\ReplicationStrategy(); + $factory = new Connection\Factory(); + + $master + ->expects($this->once()) + ->method('isConnected') + ->willReturn(true); + + $slave + ->expects($this->never()) + ->method('write'); + + $master + ->expects($this->once()) + ->method('write') + ->with($command->serializeCommand()); + + $replication = new SentinelReplication('svc', [$sentinel], $factory, $strategy); + $replication->add($master); + $replication->add($slave); + + $replication->write($command->serializeCommand()); + } + + /** + * Regression guard for GHSA-w6f5-v2h6-g786 (CWE-93): a CRLF embedded in a + * bulk string KEY must not corrupt the argument list used to pick the + * target connection, which would silently route the command to the + * wrong node. + * + * @group disconnected + */ + public function testWriteHandlesCRLFEmbeddedInBulkStringKey(): void + { + $command = new Command\Redis\SET(); + $command->setArguments(["victim\r\n*1\r\n\$4\r\nEVIL", 'somevalue']); + + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $master = $this->getMockConnection('tcp://127.0.0.1:6379?role=master'); + $slave = $this->getMockConnection('tcp://127.0.0.1:6380?role=slave'); + $strategy = new Replication\ReplicationStrategy(); + $factory = new Connection\Factory(); + + $master + ->expects($this->once()) + ->method('isConnected') + ->willReturn(true); + + $slave + ->expects($this->never()) + ->method('write'); + + $master + ->expects($this->once()) + ->method('write') + ->with($command->serializeCommand()); + + $replication = new SentinelReplication('svc', [$sentinel], $factory, $strategy); + $replication->add($master); + $replication->add($slave); + + $replication->write($command->serializeCommand()); + } + + public function connectionsProvider(): array + { + return [ + 'master connection' => ['tcp://127.0.0.1:6379?role=master'], + 'slave connection' => ['tcp://127.0.0.1:6379'], + ]; + } + + /** + * @group disconnected + */ + public function testDiscardsSlaveWhenRespondsLOADINGAndExecutesReadOnlyCommandOnNextSlave(): void + { + $sentinel1 = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel1 + ->expects($this->any()) + ->method('executeCommand') + ->with($this->isRedisCommand( + 'SENTINEL', ['slaves', 'svc'] + )) + ->willReturn( + [ + [ + 'name', '127.0.0.1:6383', + 'ip', '127.0.0.1', + 'port', '6383', + 'runid', '1c0bf1291797fbc5608c07a17da394147dc62817', + 'flags', 'slave', + 'master-host', '127.0.0.1', + 'master-port', '6381', + ], + ] + ); + + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + + $slave1 = $this->getMockConnection('tcp://127.0.0.1:6382?role=slave'); + $slave1 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $slave1 + ->expects($this->once()) + ->method('executeCommand') + ->with( + $this->isRedisCommand('GET', ['key']) + ) + ->willReturn( + new Response\Error('LOADING') + ); + + $slave2 = $this->getMockConnection('tcp://127.0.0.1:6383?role=slave'); + $slave2 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $slave2 + ->expects($this->once()) + ->method('executeCommand') + ->withConsecutive( + [$this->isRedisCommand('GET', ['key'])] + ) + ->willReturnOnConsecutiveCalls( + 'value' + ); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6383', + 'role' => 'slave', + ]) + ->willReturn($slave2); + + $replication = $this->getReplicationConnection('svc', [$sentinel1], $factory); + + $replication->add($master); + $replication->add($slave1); + + $this->assertSame('value', $replication->executeCommand( + Command\RawCommand::create('get', 'key') + )); + $this->assertSame($slave2, $replication->getCurrent()); + } + + /** + * @group disconnected + */ + public function testDoesNotRetryOnNonCommunicationException(): void + { + $master = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $master + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('SET', ['key', 'value'])) + ->willThrowException(new RuntimeException('Non-communication error')); + + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + + $replication = $this->getReplicationConnection('svc', [$sentinel], $factory); + $replication->add($master); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Non-communication error'); + + $replication->executeCommand(Command\RawCommand::create('set', 'key', 'value')); + } + + /** + * @group disconnected + */ + public function testRetriesOnlyOnCommunicationException(): void + { + $master1 = $this->getMockConnection('tcp://127.0.0.1:6381?role=master'); + $master1 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $master1 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('SET', ['key', 'value'])) + ->willThrowException(new Connection\ConnectionException($master1, 'Connection failed')); + $master1 + ->expects($this->once()) + ->method('disconnect'); + + $master2 = $this->getMockConnection('tcp://127.0.0.1:6382?role=master'); + $master2 + ->expects($this->any()) + ->method('isConnected') + ->willReturn(true); + $master2 + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('SET', ['key', 'value'])) + ->willReturn('OK'); + + $sentinel = $this->getMockSentinelConnection('tcp://127.0.0.1:5381?role=sentinel'); + $sentinel + ->expects($this->once()) + ->method('executeCommand') + ->with($this->isRedisCommand('SENTINEL', ['get-master-addr-by-name', 'svc'])) + ->willReturn(['127.0.0.1', '6382']); + + /** @var Connection\FactoryInterface|MockObject */ + $factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); + $factory + ->expects($this->once()) + ->method('create') + ->with([ + 'host' => '127.0.0.1', + 'port' => '6382', + 'role' => 'master', + ]) + ->willReturn($master2); + + $replication = $this->getReplicationConnection('svc', [$sentinel], $factory); + $replication->add($master1); + + $this->assertSame('OK', $replication->executeCommand( + Command\RawCommand::create('set', 'key', 'value') + )); + } + + /** + * @medium + * @group disconnected + * @group slow + */ + public function testRetryCommandFailureOnCustomRetryConfiguration() + { + $mockStream = $this->getMockBuilder(StreamInterface::class)->getMock(); + $mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock(); + $parameters = new Parameters([ + 'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3), + 'role' => 'master', + ]); + + $mockStream + ->expects($this->exactly(3)) + ->method('close') + ->withAnyParameters(); + + $mockStream + ->expects($this->exactly(6)) + ->method('write') + ->withAnyParameters() + ->willReturnOnConsecutiveCalls( + 1000, + $this->throwException(new RuntimeException('', 2)), + $this->throwException(new RuntimeException('', 2)), + $this->throwException(new RuntimeException('', 2)), + 1000, + 1000 + ); + + $mockStream + ->expects($this->exactly(7)) + ->method('read') + ->withAnyParameters() + ->willReturnOnConsecutiveCalls("*1\r\n", "$5\r\n", "master\r\n", "*1\r\n", "$5\r\n", "master\r\n", "+OK\r\n"); + + $mockStreamFactory + ->expects($this->exactly(4)) + ->method('createStream') + ->withAnyParameters() + ->willReturn($mockStream); + + $connection = new StreamConnection($parameters, $mockStreamFactory); + + $mockFactory = $this->getMockBuilder(Connection\Factory::class)->getMock(); + $mockFactory + ->expects($this->any()) + ->method('create') + ->willReturn($connection); + + $sentinel = $this->getMockSentinelConnection(); + $sentinel + ->expects($this->any()) + ->method('executeCommand') + ->willReturn(['127.0.0.1', '6381']); + + $replication = new SentinelReplication('src', [$sentinel], $mockFactory); + $replication->add($connection); + + $this->assertEquals( + 'OK', + $replication->executeCommand(Command\RawCommand::create('SET', 1001)) + ); + } + + // ******************************************************************** // + // ---- HELPER METHODS ------------------------------------------------ // + // ******************************************************************** // + + /** + * Creates a new instance of replication connection. + * + * @param string $service Name of the service + * @param array $sentinels Array of sentinels + * @param ConnectionFactoryInterface|null $factory Optional connection factory instance + * + * @return SentinelReplication + */ + protected function getReplicationConnection(string $service, array $sentinels, ?Connection\FactoryInterface $factory = null): SentinelReplication + { + $factory = $factory ?: new Connection\Factory(); + + $replication = new SentinelReplication($service, $sentinels, $factory); + $replication->setRetryWait(0); + + return $replication; + } + + /** + * Returns a base mocked connection from Predis\Connection\NodeConnectionInterface. + * + * @param array|string $parameters Optional parameters + * + * @return mixed + */ + protected function getMockSentinelConnection($parameters = null) + { + $connection = $this->getMockConnection($parameters); + + return $connection; + } +}