From d56e32715bcb34c5698f31480536035d1f59629b Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Thu, 28 Dec 2023 20:17:04 +0200 Subject: [PATCH] Improved Relay integration (#1423) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Fixed Relay objects structure (#1419) * Added OSS cluster tests (#1421) --------- Co-authored-by: Till Krüss --- .github/workflows/tests.yml | 3 - phpstan.dist.neon | 3 - src/Configuration/Option/Connections.php | 32 +- src/Configuration/OptionsInterface.php | 2 + src/Connection/Factory.php | 14 +- src/Connection/RelayConnection.php | 227 ++++++---- src/Connection/RelayFactory.php | 149 +++++++ tests/Predis/ClientTest.php | 15 + tests/Predis/Command/Redis/CLIENT_Test.php | 1 + .../Configuration/Option/ConnectionsTest.php | 68 ++- .../Connection/Cluster/RedisClusterTest.php | 24 + .../Predis/Connection/RelayConnectionTest.php | 419 +++++++++++------- tests/Predis/Connection/RelayFactoryTest.php | 106 +++++ 13 files changed, 734 insertions(+), 329 deletions(-) create mode 100644 src/Connection/RelayFactory.php create mode 100644 tests/Predis/Connection/RelayFactoryTest.php diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index ddfad811..c1f2c3aa 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -84,7 +84,6 @@ jobs: uses: coverallsapp/github-action@v2 if: ${{ matrix.php == '8.1' && matrix.redis == '7' }} with: - flag-name: Predis-Relay github-token: ${{ secrets.GITHUB_TOKEN }} parallel: true @@ -215,7 +214,6 @@ jobs: uses: coverallsapp/github-action@v2 if: ${{ matrix.php == '8.1' }} with: - flag-name: Stack github-token: ${{ secrets.GITHUB_TOKEN }} parallel: true @@ -229,4 +227,3 @@ jobs: uses: coverallsapp/github-action@v2 with: parallel-finished: true - carryforward: "Predis-Relay,Stack" diff --git a/phpstan.dist.neon b/phpstan.dist.neon index b9df88b3..46821cf8 100644 --- a/phpstan.dist.neon +++ b/phpstan.dist.neon @@ -26,9 +26,6 @@ parameters: - message: "#^Call to an undefined method Predis\\\\Command\\\\FactoryInterface\\:\\:undefine\\(\\)\\.$#" count: 1 path: src/Configuration/Option/Commands.php - - message: "#^Access to an undefined property Predis\\\\Configuration\\\\OptionsInterface\\:\\:\\$parameters\\.$#" - count: 1 - path: src/Configuration/Option/Connections.php - message: "#^Access to an undefined property Predis\\\\Connection\\\\ParametersInterface\\:\\:\\$weight\\.$#" count: 1 path: src/Connection/Cluster/PredisCluster.php diff --git a/src/Configuration/Option/Connections.php b/src/Configuration/Option/Connections.php index a3e14668..80d576da 100644 --- a/src/Configuration/Option/Connections.php +++ b/src/Configuration/Option/Connections.php @@ -18,6 +18,7 @@ use Predis\Configuration\OptionsInterface; use Predis\Connection\Factory; use Predis\Connection\FactoryInterface; use Predis\Connection\RelayConnection; +use Predis\Connection\RelayFactory; /** * Configures a new connection factory instance. @@ -94,28 +95,18 @@ class Connections implements OptionInterface */ protected function createFactoryByString(OptionsInterface $options, string $value) { - /** - * @var FactoryInterface - */ - $factory = $this->getDefault($options); - switch (strtolower($value)) { case 'relay': - $factory->define('tcp', RelayConnection::class); - $factory->define('redis', RelayConnection::class); - $factory->define('unix', RelayConnection::class); - break; + return $this->getRelayFactory($options); case 'default': - return $factory; + return $this->getDefault($options); default: throw new InvalidArgumentException(sprintf( '%s does not recognize `%s` as a supported configuration string', static::class, $value )); } - - return $factory; } /** @@ -131,4 +122,21 @@ class Connections implements OptionInterface return $factory; } + + /** + * Creates RelayFactory instance. + * + * @param OptionsInterface $options + * @return FactoryInterface + */ + private function getRelayFactory(OptionsInterface $options): FactoryInterface + { + $factory = new RelayFactory(); + + if ($options->defined('parameters')) { + $factory->setDefaultParameters($options->parameters); + } + + return $factory; + } } diff --git a/src/Configuration/OptionsInterface.php b/src/Configuration/OptionsInterface.php index f627e35e..38d13ddb 100644 --- a/src/Configuration/OptionsInterface.php +++ b/src/Configuration/OptionsInterface.php @@ -13,6 +13,7 @@ namespace Predis\Configuration; use Predis\Command\Processor\ProcessorInterface; +use Predis\Connection\ParametersInterface; /** * @property callable $aggregate Custom aggregate connection initializer @@ -21,6 +22,7 @@ use Predis\Command\Processor\ProcessorInterface; * @property bool $exceptions Toggles exceptions in client for -ERR responses * @property ProcessorInterface $prefix Key prefixing strategy using the supplied string as prefix * @property \Predis\Command\FactoryInterface $commands Command factory for creating Redis commands + * @property array|ParametersInterface $parameters Parameters associated with connection. * @property callable $replication Aggregate connection initializer for replication * @property int $readTimeout Timeout in milliseconds between read operations on reading from multiple connections. */ diff --git a/src/Connection/Factory.php b/src/Connection/Factory.php index 6b93c7b6..941673bf 100644 --- a/src/Connection/Factory.php +++ b/src/Connection/Factory.php @@ -174,15 +174,13 @@ class Factory implements FactoryInterface ); } - if (!$connection instanceof RelayConnection) { - $connection->addConnectCommand( - new RawCommand('CLIENT', ['SETINFO', 'LIB-NAME', 'predis']) - ); + $connection->addConnectCommand( + new RawCommand('CLIENT', ['SETINFO', 'LIB-NAME', 'predis']) + ); - $connection->addConnectCommand( - new RawCommand('CLIENT', ['SETINFO', 'LIB-VER', Client::VERSION]) - ); - } + $connection->addConnectCommand( + new RawCommand('CLIENT', ['SETINFO', 'LIB-VER', Client::VERSION]) + ); if (isset($parameters->protocol) && (int) $parameters->protocol > 2) { $connection->addConnectCommand( diff --git a/src/Connection/RelayConnection.php b/src/Connection/RelayConnection.php index 649dc86b..4a948ce9 100644 --- a/src/Connection/RelayConnection.php +++ b/src/Connection/RelayConnection.php @@ -16,6 +16,7 @@ use InvalidArgumentException; use Predis\ClientException; use Predis\Command\CommandInterface; use Predis\NotSupportedException; +use Predis\Response\ErrorInterface as ErrorResponseInterface; use Predis\Response\ServerException; use Relay\Exception as RelayException; use Relay\Relay; @@ -49,7 +50,7 @@ use Relay\Relay; * * @see https://github.com/cachewerk/relay */ -class RelayConnection extends StreamConnection +class RelayConnection extends AbstractConnection { use RelayMethods; @@ -89,12 +90,10 @@ class RelayConnection extends StreamConnection /** * {@inheritdoc} */ - public function __construct(ParametersInterface $parameters) + public function __construct(ParametersInterface $parameters, Relay $client) { - $this->assertExtensions(); - $this->parameters = $this->assertParameters($parameters); - $this->client = $this->createClient(); + $this->client = $client; } /** @@ -115,76 +114,6 @@ class RelayConnection extends StreamConnection } } - /** - * Checks if the Relay extension is loaded in PHP. - */ - private function assertExtensions() - { - if (!extension_loaded('relay')) { - throw new NotSupportedException( - 'The "relay" extension is required by this connection backend.' - ); - } - } - - /** - * {@inheritdoc} - */ - protected function assertParameters(ParametersInterface $parameters) - { - if (!in_array($parameters->scheme, ['tcp', 'tls', 'unix', 'redis', 'rediss'])) { - throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'."); - } - - if (!in_array($parameters->serializer, [null, 'php', 'igbinary', 'msgpack', 'json'])) { - throw new InvalidArgumentException("Invalid serializer: '{$parameters->serializer}'."); - } - - if (!in_array($parameters->compression, [null, 'lzf', 'lz4', 'zstd'])) { - throw new InvalidArgumentException("Invalid compression algorithm: '{$parameters->compression}'."); - } - - return $parameters; - } - - /** - * Creates a new instance of the client. - * - * @return Relay - */ - private function createClient() - { - $client = new Relay(); - - // throw when errors occur and return `null` for non-existent keys - $client->setOption(Relay::OPT_PHPREDIS_COMPATIBILITY, false); - - // use reply literals - $client->setOption(Relay::OPT_REPLY_LITERAL, true); - - // disable Relay's command/connection retry - $client->setOption(Relay::OPT_MAX_RETRIES, 0); - - // whether to use in-memory caching - $client->setOption(Relay::OPT_USE_CACHE, $this->parameters->cache ?? true); - - // set data serializer - $client->setOption(Relay::OPT_SERIALIZER, constant(sprintf( - '%s::SERIALIZER_%s', - Relay::class, - strtoupper($this->parameters->serializer ?? 'none') - ))); - - // set data compression algorithm - $client->setOption(Relay::OPT_COMPRESSION, constant(sprintf( - '%s::COMPRESSION_%s', - Relay::class, - strtoupper($this->parameters->compression ?? 'none') - ))); - - return $client; - } - /** * Returns the underlying client. * @@ -196,17 +125,12 @@ class RelayConnection extends StreamConnection } /** - * {@inheritdoc} + * @param ParametersInterface $parameters + * @param $address + * @param $flags + * @return Relay */ - protected function getIdentifier() - { - return $this->client->endpointId(); - } - - /** - * {@inheritdoc} - */ - protected function createStreamSocket(ParametersInterface $parameters, $address, $flags) + protected function connectWithConfiguration(ParametersInterface $parameters, $address, $flags) { $timeout = isset($parameters->timeout) ? (float) $parameters->timeout : 5.0; @@ -310,14 +234,6 @@ class RelayConnection extends StreamConnection throw new NotSupportedException('The "relay" extension does not support writing requests.'); } - /** - * {@inheritdoc} - */ - public function readResponse(CommandInterface $command) - { - throw new NotSupportedException('The "relay" extension does not support reading responses.'); - } - /** * {@inheritdoc} */ @@ -329,9 +245,128 @@ class RelayConnection extends StreamConnection /** * {@inheritdoc} */ - public function __wakeup() + protected function createResource() { - $this->assertExtensions(); - $this->client = $this->createClient(); + switch ($this->parameters->scheme) { + case 'tcp': + case 'redis': + return $this->initializeTcpConnection($this->parameters); + + case 'unix': + return $this->initializeUnixConnection($this->parameters); + + default: + throw new InvalidArgumentException("Invalid scheme: '{$this->parameters->scheme}'."); + } + } + + /** + * Initializes a TCP connection via client. + * + * @param ParametersInterface $parameters Initialization parameters for the connection. + * + * @return Relay + */ + protected function initializeTcpConnection(ParametersInterface $parameters) + { + if (!filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) { + $address = "tcp://$parameters->host:$parameters->port"; + } else { + $address = "tcp://[$parameters->host]:$parameters->port"; + } + + $flags = STREAM_CLIENT_CONNECT; + + if (isset($parameters->async_connect) && $parameters->async_connect) { + $flags |= STREAM_CLIENT_ASYNC_CONNECT; + } + + if (isset($parameters->persistent)) { + if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) { + $flags |= STREAM_CLIENT_PERSISTENT; + + if ($persistent === null) { + $address = "{$address}/{$parameters->persistent}"; + } + } + } + + return $this->connectWithConfiguration($parameters, $address, $flags); + } + + /** + * Initializes a UNIX connection via client. + * + * @param ParametersInterface $parameters Initialization parameters for the connection. + * + * @return Relay + */ + protected function initializeUnixConnection(ParametersInterface $parameters) + { + if (!isset($parameters->path)) { + throw new InvalidArgumentException('Missing UNIX domain socket path.'); + } + + $flags = STREAM_CLIENT_CONNECT; + + if (isset($parameters->persistent)) { + if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) { + $flags |= STREAM_CLIENT_PERSISTENT; + + if ($persistent === null) { + throw new InvalidArgumentException( + 'Persistent connection IDs are not supported when using UNIX domain sockets.' + ); + } + } + } + + return $this->connectWithConfiguration($parameters, "unix://{$parameters->path}", $flags); + } + + /** + * {@inheritdoc} + */ + public function connect() + { + if (parent::connect() && $this->initCommands) { + foreach ($this->initCommands as $command) { + $response = $this->executeCommand($command); + + if ($response instanceof ErrorResponseInterface && ($command->getId() === 'CLIENT')) { + // Do nothing on CLIENT SETINFO command failure + } elseif ($response instanceof ErrorResponseInterface) { + $this->onConnectionError("`{$command->getId()}` failed: {$response->getMessage()}", 0); + } + } + } + } + + /** + * {@inheritdoc} + */ + public function read() + { + throw new NotSupportedException('The "relay" extension does not support reading responses.'); + } + + /** + * {@inheritdoc} + */ + protected function assertParameters(ParametersInterface $parameters) + { + if (!in_array($parameters->scheme, ['tcp', 'tls', 'unix', 'redis', 'rediss'])) { + throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'."); + } + + if (!in_array($parameters->serializer, [null, 'php', 'igbinary', 'msgpack', 'json'])) { + throw new InvalidArgumentException("Invalid serializer: '{$parameters->serializer}'."); + } + + if (!in_array($parameters->compression, [null, 'lzf', 'lz4', 'zstd'])) { + throw new InvalidArgumentException("Invalid compression algorithm: '{$parameters->compression}'."); + } + + return $parameters; } } diff --git a/src/Connection/RelayFactory.php b/src/Connection/RelayFactory.php new file mode 100644 index 00000000..26b647d0 --- /dev/null +++ b/src/Connection/RelayFactory.php @@ -0,0 +1,149 @@ + RelayConnection::class, + 'tls' => RelayConnection::class, + 'unix' => RelayConnection::class, + 'redis' => RelayConnection::class, + 'rediss' => RelayConnection::class, + ]; + + /** + * {@inheritDoc} + */ + public function define($scheme, $initializer) + { + throw new NotSupportedException('Does not allow to override existing initializer.'); + } + + /** + * {@inheritDoc} + */ + public function undefine($scheme) + { + throw new NotSupportedException('Does not allow to override existing initializer.'); + } + + /** + * {@inheritDoc} + */ + public function create($parameters): NodeConnectionInterface + { + $this->assertExtensions(); + + if (!$parameters instanceof ParametersInterface) { + $parameters = $this->createParameters($parameters); + } + + $scheme = $parameters->scheme; + + if (!isset($this->schemes[$scheme])) { + throw new InvalidArgumentException("Unknown connection scheme: '$scheme'."); + } + + $initializer = $this->schemes[$scheme]; + $client = $this->createClient(); + + $connection = new $initializer($parameters, $client); + + $this->prepareConnection($connection); + + return $connection; + } + + /** + * Checks if the Relay extension is loaded in PHP. + */ + private function assertExtensions() + { + if (!extension_loaded('relay')) { + throw new NotSupportedException( + 'The "relay" extension is required by this connection backend.' + ); + } + } + + /** + * Creates a new instance of the client. + * + * @return Relay + */ + private function createClient() + { + $client = new Relay(); + + // throw when errors occur and return `null` for non-existent keys + $client->setOption(Relay::OPT_PHPREDIS_COMPATIBILITY, false); + + // use reply literals + $client->setOption(Relay::OPT_REPLY_LITERAL, true); + + // disable Relay's command/connection retry + $client->setOption(Relay::OPT_MAX_RETRIES, 0); + + // whether to use in-memory caching + $client->setOption(Relay::OPT_USE_CACHE, $this->parameters->cache ?? true); + + // set data serializer + $client->setOption(Relay::OPT_SERIALIZER, constant(sprintf( + '%s::SERIALIZER_%s', + Relay::class, + strtoupper($this->parameters->serializer ?? 'none') + ))); + + // set data compression algorithm + $client->setOption(Relay::OPT_COMPRESSION, constant(sprintf( + '%s::COMPRESSION_%s', + Relay::class, + strtoupper($this->parameters->compression ?? 'none') + ))); + + return $client; + } + + /** + * {@inheritdoc} + */ + protected function prepareConnection(NodeConnectionInterface $connection) + { + $parameters = $connection->getParameters(); + + if (isset($parameters->password) && strlen($parameters->password)) { + $cmdAuthArgs = isset($parameters->username) && strlen($parameters->username) + ? [$parameters->username, $parameters->password] + : [$parameters->password]; + + $connection->addConnectCommand( + new RawCommand('AUTH', $cmdAuthArgs) + ); + } + + if (isset($parameters->database) && strlen($parameters->database)) { + $connection->addConnectCommand( + new RawCommand('SELECT', [$parameters->database]) + ); + } + } +} diff --git a/tests/Predis/ClientTest.php b/tests/Predis/ClientTest.php index ed5633ea..3eb34088 100644 --- a/tests/Predis/ClientTest.php +++ b/tests/Predis/ClientTest.php @@ -1323,6 +1323,21 @@ class ClientTest extends PredisTestCase $this->assertSame(Client::VERSION, $libVer); } + /** + * @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')); + } + // ******************************************************************** // // ---- HELPER METHODS ------------------------------------------------ // // ******************************************************************** // diff --git a/tests/Predis/Command/Redis/CLIENT_Test.php b/tests/Predis/Command/Redis/CLIENT_Test.php index 78bcc23f..44be4fd7 100644 --- a/tests/Predis/Command/Redis/CLIENT_Test.php +++ b/tests/Predis/Command/Redis/CLIENT_Test.php @@ -203,6 +203,7 @@ BUFFER; /** * @group connected + * @group relay-incompatible * @requiresRedisVersion >= 6.0.0 */ public function testGetsNameOfConnectionResp3(): void diff --git a/tests/Predis/Configuration/Option/ConnectionsTest.php b/tests/Predis/Configuration/Option/ConnectionsTest.php index 5d44ea56..985b88b7 100644 --- a/tests/Predis/Configuration/Option/ConnectionsTest.php +++ b/tests/Predis/Configuration/Option/ConnectionsTest.php @@ -13,6 +13,7 @@ namespace Predis\Configuration\Option; use Predis\Configuration\OptionsInterface; +use Predis\Connection\RelayFactory; use PredisTestCase; use stdClass; @@ -64,31 +65,44 @@ class ConnectionsTest extends PredisTestCase /** * @group disconnected - * @dataProvider provideSupportedStringValuesForOption */ - public function testAcceptsStringToConfigureRelayBackend($value, $classFQCN) + public function testAcceptsStringToConfigureRelayBackendWithoutParameters() { $options = $this->getMockBuilder('Predis\Configuration\OptionsInterface')->getMock(); - $default = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock(); - $default - ->expects($this->exactly(3)) - ->method('define') - ->with($this->matchesRegularExpression('/^tcp|unix|redis$/'), $classFQCN); - - $option = $this->getMockBuilder('Predis\Configuration\Option\Connections') - ->setMethods(['getDefault']) - ->getMock(); - $option + $options ->expects($this->once()) - ->method('getDefault') - ->with($options) - ->will($this->returnValue($default)); + ->method('defined') + ->with('parameters') + ->willReturn(false); - $factory = $option->filter($options, $value); + $option = new Connections(); + $factory = $option->filter($options, 'relay'); - $this->assertInstanceOf('Predis\Connection\FactoryInterface', $factory); - $this->assertSame($default, $factory); + $this->assertInstanceOf(RelayFactory::class, $factory); + $this->assertEmpty($factory->getDefaultParameters()); + } + + /** + * @group disconnected + */ + public function testAcceptsStringToConfigureRelayBackendWithParameters() + { + $options = $this->getMockBuilder('Predis\Configuration\OptionsInterface')->getMock(); + + $options + ->expects($this->once()) + ->method('defined') + ->with('parameters') + ->willReturn(true); + + $options->parameters = ['foo' => 'bar']; + + $option = new Connections(); + $factory = $option->filter($options, 'relay'); + + $this->assertInstanceOf(RelayFactory::class, $factory); + $this->assertSame(['foo' => 'bar'], $factory->getDefaultParameters()); } /** @@ -139,7 +153,7 @@ class ConnectionsTest extends PredisTestCase { $parameters = ['database' => 5, 'password' => 'mypassword']; - /** @var \Predis\Configuration\OptionsInterface|\PHPUnit\Framework\MockObject\MockObject\MockObject */ + /** @var OptionsInterface|\PHPUnit\Framework\MockObject\MockObject\MockObject */ $options = $this->getMockBuilder('Predis\Configuration\OptionsInterface')->getMock(); $options ->expects($this->once()) @@ -216,20 +230,4 @@ class ConnectionsTest extends PredisTestCase $option->filter($options, new stdClass()); } - - // ******************************************************************** // - // ---- HELPER METHODS ------------------------------------------------ // - // ******************************************************************** // - - /** - * Test provider for string values supported by this client option. - * - * @return array - */ - public function provideSupportedStringValuesForOption() - { - return [ - ['relay', \Predis\Connection\RelayConnection::class], - ]; - } } diff --git a/tests/Predis/Connection/Cluster/RedisClusterTest.php b/tests/Predis/Connection/Cluster/RedisClusterTest.php index 0327bf35..f32fc833 100644 --- a/tests/Predis/Connection/Cluster/RedisClusterTest.php +++ b/tests/Predis/Connection/Cluster/RedisClusterTest.php @@ -1494,4 +1494,28 @@ class RedisClusterTest extends PredisTestCase $this->assertEquals(['response1', 'response2', 'response3'], $cluster->executeCommandOnEachNode($mockCommand)); } + + /** + * @group disconnected + */ + public function testCreatesRelayConnectionsCluster(): void + { + $factory = $this->getMockBuilder(Connection\RelayFactory::class)->getMock(); + $expectedConnection = $this + ->getMockBuilder(Connection\RelayConnection::class) + ->disableOriginalConstructor() + ->getMock(); + $parameters = new Parameters([ + 'scheme' => 'tcp', + 'host' => '127.0.0.1', + 'port' => 7001, + 'protocol' => '3', + ]); + + $cluster = new RedisCluster($factory, $parameters); + + $cluster->add($expectedConnection); + + $this->assertInstanceOf(Connection\RelayConnection::class, $cluster->getConnectionBySlot(9999)); + } } diff --git a/tests/Predis/Connection/RelayConnectionTest.php b/tests/Predis/Connection/RelayConnectionTest.php index 11bc72e3..c60411df 100644 --- a/tests/Predis/Connection/RelayConnectionTest.php +++ b/tests/Predis/Connection/RelayConnectionTest.php @@ -12,16 +12,244 @@ namespace Predis\Connection; -use PHPUnit\Framework\MockObject\MockObject; use Predis\Command\RawCommand; use Predis\Response\Error as ErrorResponse; +use PredisTestCase; +use Relay\Relay; /** * @group ext-relay * @requires extension relay */ -class RelayConnectionTest extends PredisConnectionTestCase +class RelayConnectionTest extends PredisTestCase { + /** + * @var Relay + */ + private $mockClient; + + /** + * @var ParametersInterface + */ + private $parameters; + + /** + * @var RelayConnection + */ + private $connection; + + protected function setUp(): void + { + $this->mockClient = $this->getMockBuilder(Relay::class)->getMock(); + $this->parameters = new Parameters(); + + $this->connection = new RelayConnection($this->parameters, $this->mockClient); + } + + /** + * @group disconnected + * @return void + */ + public function testIsConnected(): void + { + $this->mockClient + ->expects($this->once()) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(true); + + $this->assertTrue($this->connection->isConnected()); + } + + /** + * @group disconnected + * @return void + */ + public function testDisconnect(): void + { + $this->mockClient + ->expects($this->once()) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(true); + + $this->mockClient + ->expects($this->once()) + ->method('close') + ->withAnyParameters(); + + $this->connection->disconnect(); + } + + /** + * @group disconnected + * @return void + */ + public function testGetClient(): void + { + $this->assertSame($this->mockClient, $this->connection->getClient()); + } + + /** + * @group disconnected + * @return void + */ + public function testExecuteCommandOnAlreadyConnectedClient(): void + { + $this->mockClient + ->expects($this->once()) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(true); + + $this->mockClient + ->expects($this->once()) + ->method('rawCommand') + ->with('GET', 'foo') + ->willReturn('bar'); + + $response = $this->connection->executeCommand(new RawCommand('GET', ['foo'])); + + $this->assertSame('bar', $response); + } + + /** + * @group disconnected + * @return void + */ + public function testExecuteAtypicalCommandOnAlreadyConnectedClient(): void + { + $this->mockClient + ->expects($this->once()) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(true); + + $this->mockClient + ->expects($this->once()) + ->method('AUTH') + ->with('foo', 'bar') + ->willReturn(true); + + $response = $this->connection->executeCommand(new RawCommand('AUTH', ['foo', 'bar'])); + + $this->assertTrue($response); + } + + /** + * @group disconnected + * @return void + */ + public function testExecuteCommandOnNonConnectedClient(): void + { + $this->mockClient + ->expects($this->exactly(2)) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(false); + + $this->mockClient + ->expects($this->once()) + ->method('connect') + ->with($this->parameters->host, $this->parameters->port); + + $this->mockClient + ->expects($this->once()) + ->method('rawCommand') + ->with('GET', 'foo') + ->willReturn('bar'); + + $response = $this->connection->executeCommand(new RawCommand('GET', ['foo'])); + + $this->assertSame('bar', $response); + } + + /** + * @group disconnected + * @return void + */ + public function testConnectExecutesOnConnectionCommand(): void + { + $this->mockClient + ->expects($this->exactly(3)) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(false); + + $this->mockClient + ->expects($this->once()) + ->method('connect') + ->with($this->parameters->host, $this->parameters->port); + + $this->mockClient + ->expects($this->exactly(2)) + ->method('rawCommand') + ->withConsecutive(['GET', 'foo'], ['GET', 'bar']) + ->willReturnOnConsecutiveCalls('baz', 'bad'); + + $this->connection->addConnectCommand(new RawCommand('GET', ['foo'])); + $this->connection->addConnectCommand(new RawCommand('GET', ['bar'])); + + $this->connection->connect(); + } + + /** + * @group disconnected + * @return void + */ + public function testConnectThrowsExceptionOnErrorResponse(): void + { + $this->mockClient + ->expects($this->exactly(4)) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(false); + + $this->mockClient + ->expects($this->once()) + ->method('connect') + ->with($this->parameters->host, $this->parameters->port); + + $this->mockClient + ->expects($this->exactly(2)) + ->method('rawCommand') + ->withConsecutive(['GET', 'foo'], ['GET', 'bar']) + ->willReturnOnConsecutiveCalls('baz', new ErrorResponse('FooBar')); + + $this->connection->addConnectCommand(new RawCommand('GET', ['foo'])); + $this->connection->addConnectCommand(new RawCommand('GET', ['bar'])); + + $this->expectException(ConnectionException::class); + $this->expectExceptionMessage('`GET` failed: FooBar [tcp://127.0.0.1:6379]'); + + $this->connection->connect(); + } + + /** + * @group disconnected + * @return void + */ + public function testConnectDoNotExecuteCommands(): void + { + $this->mockClient + ->expects($this->once()) + ->method('isConnected') + ->withAnyParameters() + ->willReturn(false); + + $this->mockClient + ->expects($this->once()) + ->method('connect') + ->with($this->parameters->host, $this->parameters->port); + + $this->mockClient + ->expects($this->never()) + ->method('rawCommand') + ->withAnyParameters(); + + $this->connection->connect(); + } + /** * {@inheritDoc} */ @@ -30,35 +258,6 @@ class RelayConnectionTest extends PredisConnectionTestCase return RelayConnection::class; } - /** - * @group disconnected - */ - public function testThrowsExceptionOnInitializationCommandFailure(): void - { - $this->expectException('Predis\Connection\ConnectionException'); - $this->expectExceptionMessage('`SELECT` failed: ERR invalid DB index [tcp://127.0.0.1:6379]'); - - $cmdSelect = RawCommand::create('SELECT', '1000'); - - /** @var NodeConnectionInterface|MockObject */ - $connection = $this - ->getMockBuilder($this->getConnectionClass()) - ->onlyMethods(['executeCommand', 'createResource']) - ->setConstructorArgs([new Parameters()]) - ->getMock(); - $connection - ->method('executeCommand') - ->with($cmdSelect) - ->willReturn( - new ErrorResponse('ERR invalid DB index') - ); - - $connection->method('createResource'); - - $connection->addConnectCommand($cmdSelect); - $connection->connect(); - } - // ******************************************************************** // // ---- INTEGRATION TESTS --------------------------------------------- // // ******************************************************************** // @@ -68,7 +267,7 @@ class RelayConnectionTest extends PredisConnectionTestCase */ public function testGetResourceForcesConnection(): void { - $connection = $this->createConnection(); + $connection = new RelayConnection(new Parameters(), new \Relay\Relay()); $this->assertFalse($connection->isConnected()); $connection->getResource(); @@ -77,158 +276,34 @@ class RelayConnectionTest extends PredisConnectionTestCase /** * @group connected - * @group slow - * @requires PHP 5.4 */ - public function testThrowsExceptionOnReadWriteTimeout(): void + public function testExecutesCommand(): void { - $this->expectException('Predis\Connection\ConnectionException'); + $connection = new RelayConnection(new Parameters(), new \Relay\Relay()); - $connection = $this->createConnectionWithParams([ - 'read_write_timeout' => 0.5, - ], true); + $this->assertEquals( + 'OK', + $connection->executeCommand(new RawCommand('SET', ['key', 'value'])) + ); - $connection->executeCommand( - $this->getCommandFactory()->create('brpop', ['foo', 3]) + $this->assertEquals( + 'value', + $connection->executeCommand(new RawCommand('GET', ['key'])) ); } /** - * @medium * @group connected - * @group relay-incompatible + * @requiresRedisVersion >= 7.2.0 */ - public function testThrowsExceptionOnProtocolDesynchronizationErrors(): void + public function testConnectWithOnConnectionCommands(): void { - $this->expectException('Predis\Protocol\ProtocolException'); + $connection = new RelayConnection(new Parameters(), new \Relay\Relay()); + $connection->addConnectCommand(new RawCommand('CLIENT', ['SETNAME', 'predis'])); - $connection = $this->createConnection(); - $stream = $connection->getResource(); + $connection->connect(); - $connection->writeRequest($this->getCommandFactory()->create('ping')); - stream_socket_recvfrom($stream, 1); - - $connection->read(); - } - - /** - * @group connected - * @group relay-incompatible - * @requires PHP 5.4 - */ - public function testPersistentParameterWithFalseLikeValues(): void - { - $connection1 = $this->createConnectionWithParams(['persistent' => 0]); - $this->assertNonPersistentConnection($connection1); - - $connection2 = $this->createConnectionWithParams(['persistent' => false]); - $this->assertNonPersistentConnection($connection2); - - $connection3 = $this->createConnectionWithParams(['persistent' => '0']); - $this->assertNonPersistentConnection($connection3); - - $connection4 = $this->createConnectionWithParams(['persistent' => 'false']); - $this->assertNonPersistentConnection($connection4); - } - - /** - * @group connected - * @group relay-incompatible - * @requires PHP 5.4 - */ - public function testPersistentParameterWithTrueLikeValues(): void - { - $connection1 = $this->createConnectionWithParams(['persistent' => 1]); - $this->assertPersistentConnection($connection1); - - $connection2 = $this->createConnectionWithParams(['persistent' => true]); - $this->assertPersistentConnection($connection2); - - $connection3 = $this->createConnectionWithParams(['persistent' => '1']); - $this->assertPersistentConnection($connection3); - - $connection4 = $this->createConnectionWithParams(['persistent' => 'true']); - $this->assertPersistentConnection($connection4); - - $connection1->disconnect(); - } - - /** - * @group connected - * @group relay-incompatible - * @requires PHP 5.4 - */ - public function testPersistentConnectionsToSameNodeShareResource(): void - { - $connection1 = $this->createConnectionWithParams(['persistent' => true]); - $connection2 = $this->createConnectionWithParams(['persistent' => true]); - - $this->assertPersistentConnection($connection1); - $this->assertPersistentConnection($connection2); - - $this->assertSame($connection1->getResource(), $connection2->getResource()); - - $connection1->disconnect(); - } - - /** - * @group connected - * @group relay-incompatible - * @requires PHP 5.4 - */ - public function testPersistentConnectionsToSameNodeDoNotShareResourceUsingDifferentPersistentID(): void - { - $connection1 = $this->createConnectionWithParams(['persistent' => 'conn1']); - $connection2 = $this->createConnectionWithParams(['persistent' => 'conn2']); - - $this->assertPersistentConnection($connection1); - $this->assertPersistentConnection($connection2); - - $this->assertNotSame($connection1->getResource(), $connection2->getResource()); - } - - /** - * @group connected - * @group relay-incompatible - */ - public function testTcpNodelayParameterSetsContextFlagWhenTrue() - { - $connection = $this->createConnectionWithParams(['tcp_nodelay' => true]); - $options = stream_context_get_options($connection->getResource()); - - $this->assertIsArray($options); - $this->assertArrayHasKey('socket', $options); - $this->assertArrayHasKey('tcp_nodelay', $options['socket']); - $this->assertTrue($options['socket']['tcp_nodelay']); - } - - /** - * @group connected - * @group relay-incompatible - */ - public function testTcpNodelayParameterDoesNotSetContextFlagWhenFalse() - { - $connection = $this->createConnectionWithParams(['tcp_nodelay' => false]); - $options = stream_context_get_options($connection->getResource()); - - $this->assertIsArray($options); - $this->assertArrayHasKey('socket', $options); - $this->assertArrayHasKey('tcp_nodelay', $options['socket']); - $this->assertFalse($options['socket']['tcp_nodelay']); - } - - /** - * @group connected - * @group relay-incompatible - */ - public function testTcpDelayContextFlagIsNotSetByDefault() - { - $connection = $this->createConnectionWithParams([]); - $options = stream_context_get_options($connection->getResource()); - - $this->assertIsArray($options); - $this->assertArrayHasKey('socket', $options); - $this->assertArrayHasKey('tcp_nodelay', $options['socket']); - $this->assertFalse($options['socket']['tcp_nodelay']); + $response = $connection->executeCommand(new RawCommand('CLIENT', ['GETNAME'])); + $this->assertSame('predis', $response); } } diff --git a/tests/Predis/Connection/RelayFactoryTest.php b/tests/Predis/Connection/RelayFactoryTest.php new file mode 100644 index 00000000..e9b03a26 --- /dev/null +++ b/tests/Predis/Connection/RelayFactoryTest.php @@ -0,0 +1,106 @@ +factory = new RelayFactory(); + } + + /** + * @group disconnected + * @return void + */ + public function testDefine(): void + { + $this->expectException(NotSupportedException::class); + $this->expectExceptionMessage('Does not allow to override existing initializer.'); + + $this->factory->define('foo', 'bar'); + } + + /** + * @group disconnected + * @return void + */ + public function testUndefine(): void + { + $this->expectException(NotSupportedException::class); + $this->expectExceptionMessage('Does not allow to override existing initializer.'); + + $this->factory->undefine('foo'); + } + + /** + * @group disconnected + * @return void + */ + public function testCreatesConnectionWithParametersAsObject(): void + { + $connection = $this->factory->create(new Parameters()); + + $this->assertInstanceOf(RelayConnection::class, $connection); + } + + /** + * @group disconnected + * @return void + */ + public function testCreatesConnectionWithParametersAsArray(): void + { + $connection = $this->factory->create([]); + + $this->assertInstanceOf(RelayConnection::class, $connection); + } + + /** + * @group disconnected + * @return void + */ + public function testCreatesConnectionWithInitCommands(): void + { + $connection = $this->factory->create(['username' => 'foo', 'password' => 'bar', 'database' => 15]); + + $this->assertInstanceOf(RelayConnection::class, $connection); + $this->assertEquals(new RawCommand('AUTH', ['foo', 'bar']), $connection->getInitCommands()[0]); + $this->assertEquals(new RawCommand('SELECT', [15]), $connection->getInitCommands()[1]); + } + + /** + * @group disconnected + * @return void + */ + public function testCreateThrowsExceptionOnUnexpectedScheme(): void + { + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage("Unknown connection scheme: 'foobar'."); + + $this->factory->create(['scheme' => 'foobar']); + } +}