mirror of
https://github.com/predis/predis.git
synced 2026-09-01 05:03:28 +00:00
Till Krüss
1092 lines
33 KiB
PHP
1092 lines
33 KiB
PHP
<?php
|
|
|
|
/*
|
|
* This file is part of the Predis package.
|
|
*
|
|
* (c) 2009-2020 Daniele Alessandri
|
|
* (c) 2021-2026 Till Krüss
|
|
*
|
|
* For the full copyright and license information, please view the LICENSE
|
|
* file that was distributed with this source code.
|
|
*/
|
|
|
|
namespace Predis\Pipeline;
|
|
|
|
use Exception;
|
|
use InvalidArgumentException;
|
|
use Predis\Client;
|
|
use Predis\ClientException;
|
|
use Predis\ClientInterface;
|
|
use Predis\Command\CommandInterface;
|
|
use Predis\Command\Redis\ECHO_;
|
|
use Predis\Command\Redis\PING;
|
|
use Predis\Connection\Cluster\RedisCluster;
|
|
use Predis\Connection\Factory;
|
|
use Predis\Connection\Parameters;
|
|
use Predis\Connection\Replication\MasterSlaveReplication;
|
|
use Predis\Connection\Resource\StreamFactoryInterface;
|
|
use Predis\Connection\StreamConnection;
|
|
use Predis\Replication\ReplicationStrategy;
|
|
use Predis\Response;
|
|
use Predis\Retry\Retry;
|
|
use Predis\Retry\Strategy\ExponentialBackoff;
|
|
use Predis\TimeoutException;
|
|
use PredisTestCase;
|
|
use Psr\Http\Message\StreamInterface;
|
|
use RuntimeException;
|
|
use stdClass;
|
|
|
|
class PipelineTest extends PredisTestCase
|
|
{
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testConstructor(): void
|
|
{
|
|
$client = new Client();
|
|
$pipeline = new Pipeline($client);
|
|
|
|
$this->assertSame($client, $pipeline->getClient());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testCallDoesNotSendCommandsWithoutExecute(): void
|
|
{
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->never())
|
|
->method('writeRequest');
|
|
$connection
|
|
->expects($this->never())
|
|
->method('readResponse');
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->echo('one');
|
|
$pipeline->echo('two');
|
|
$pipeline->echo('three');
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testCallReturnsPipelineForFluentInterface(): void
|
|
{
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->never())
|
|
->method('writeRequest');
|
|
$connection
|
|
->expects($this->never())
|
|
->method('readResponse');
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$this->assertSame($pipeline, $pipeline->echo('one'));
|
|
$this->assertSame($pipeline, $pipeline->echo('one')->echo('two')->echo('three'));
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testDoesNotParseComplexResponseObjects(): void
|
|
{
|
|
$object = $this->getMockBuilder('Predis\Response\ResponseInterface')->getMock();
|
|
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->once())
|
|
->method('readResponse')
|
|
->willReturn($object);
|
|
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->ping();
|
|
|
|
$this->assertSame([$object], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testThrowsServerExceptionOnResponseErrorByDefault(): void
|
|
{
|
|
$this->expectException('Predis\Response\ServerException');
|
|
$this->expectExceptionMessage('ERR Test error');
|
|
|
|
$error = new Response\Error('ERR Test error');
|
|
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->once())
|
|
->method('readResponse')
|
|
->willReturn($error);
|
|
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->ping();
|
|
$pipeline->ping();
|
|
|
|
$pipeline->execute();
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testReturnsResponseErrorWithClientExceptionsSetToFalse(): void
|
|
{
|
|
$error = $this->getMockBuilder('Predis\Response\ErrorInterface')->getMock();
|
|
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('readResponse')
|
|
->willReturn($error);
|
|
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$client = new Client($connection, ['exceptions' => false]);
|
|
|
|
$pipeline = new Pipeline($client);
|
|
|
|
$pipeline->ping();
|
|
$pipeline->ping();
|
|
|
|
$this->assertSame([$error, $error], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteReturnsPipelineForFluentInterface(): void
|
|
{
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$command = $this->getCommandFactory()->create('echo', ['one']);
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$this->assertSame($pipeline, $pipeline->executeCommand($command));
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteCommandDoesNotSendCommandsWithoutExecute(): void
|
|
{
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->never())
|
|
->method('writeRequest');
|
|
$connection
|
|
->expects($this->never())
|
|
->method('readResponse');
|
|
|
|
$commands = $this->getCommandFactory();
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->executeCommand($commands->create('echo', ['one']));
|
|
$pipeline->executeCommand($commands->create('echo', ['two']));
|
|
$pipeline->executeCommand($commands->create('echo', ['three']));
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteWithEmptyBuffer(): void
|
|
{
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->never())
|
|
->method('writeRequest');
|
|
$connection
|
|
->expects($this->never())
|
|
->method('readResponse');
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$this->assertSame([], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteWithFilledBuffer(): void
|
|
{
|
|
$command1 = new ECHO_();
|
|
$command1->setArguments(['one']);
|
|
|
|
$command2 = new ECHO_();
|
|
$command2->setArguments(['two']);
|
|
|
|
$command3 = new ECHO_();
|
|
$command3->setArguments(['three']);
|
|
|
|
$buffer = $command1->serializeCommand() . $command2->serializeCommand() . $command3->serializeCommand();
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->once())
|
|
->method('write')
|
|
->with($buffer);
|
|
$connection
|
|
->expects($this->exactly(3))
|
|
->method('readResponse')
|
|
->willReturnCallback($this->getReadCallback());
|
|
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->echo('one');
|
|
$pipeline->echo('two');
|
|
$pipeline->echo('three');
|
|
|
|
$pipeline->flushPipeline();
|
|
|
|
$this->assertSame(['one', 'two', 'three'], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testFlushWithFalseArgumentDiscardsBuffer(): void
|
|
{
|
|
$pipeline = new Pipeline(new Client());
|
|
|
|
$pipeline->echo('one');
|
|
$pipeline->echo('two');
|
|
$pipeline->echo('three');
|
|
|
|
$pipeline->flushPipeline(false);
|
|
|
|
$this->assertSame([], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testFlushHandlesPartialBuffers(): void
|
|
{
|
|
$command1 = new ECHO_();
|
|
$command1->setArguments(['one']);
|
|
|
|
$command2 = new ECHO_();
|
|
$command2->setArguments(['two']);
|
|
|
|
$buffer1 = $command1->serializeCommand() . $command2->serializeCommand();
|
|
|
|
$command3 = new ECHO_();
|
|
$command3->setArguments(['three']);
|
|
|
|
$command4 = new ECHO_();
|
|
$command4->setArguments(['four']);
|
|
|
|
$buffer2 = $command3->serializeCommand() . $command4->serializeCommand();
|
|
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('write')
|
|
->withConsecutive([$buffer1], [$buffer2]);
|
|
$connection
|
|
->expects($this->exactly(4))
|
|
->method('readResponse')
|
|
->willReturnCallback($this->getReadCallback());
|
|
|
|
$connection
|
|
->expects($this->exactly(4))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->echo('one');
|
|
$pipeline->echo('two');
|
|
$pipeline->flushPipeline();
|
|
$pipeline->echo('three');
|
|
$pipeline->echo('four');
|
|
|
|
$this->assertSame(['one', 'two', 'three', 'four'], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testSwitchesToMasterWithReplicationConnection(): void
|
|
{
|
|
$nodeConnection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$nodeConnection
|
|
->expects($this->exactly(3))
|
|
->method('write')
|
|
->with((new PING())->serializeCommand());
|
|
$pong = new Response\Status('PONG');
|
|
|
|
$connection = $this->getMockBuilder('Predis\Connection\Replication\ReplicationInterface')->getMock();
|
|
$connection
|
|
->expects($this->once())
|
|
->method('switchToMaster');
|
|
$connection
|
|
->expects($this->exactly(6))
|
|
->method('getConnectionByCommand')
|
|
->willReturn($nodeConnection);
|
|
$nodeConnection
|
|
->expects($this->exactly(3))
|
|
->method('readResponse')
|
|
->willReturn($pong);
|
|
$connection
|
|
->expects($this->exactly(3))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$pipeline->ping();
|
|
$pipeline->ping();
|
|
$pipeline->ping();
|
|
|
|
$this->assertSame([$pong, $pong, $pong], $pipeline->execute());
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteAcceptsCallableArgument(): void
|
|
{
|
|
$test = $this;
|
|
$pipeline = new Pipeline(new Client());
|
|
|
|
$callable = static function (Pipeline $pipe) use ($test, $pipeline) {
|
|
$test->assertSame($pipeline, $pipe);
|
|
$pipe->flushPipeline(false);
|
|
};
|
|
|
|
$pipeline->execute($callable);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteDoesNotAcceptNonCallableArgument(): void
|
|
{
|
|
$this->expectException('InvalidArgumentException');
|
|
|
|
$noncallable = new stdClass();
|
|
|
|
$pipeline = new Pipeline(new Client());
|
|
$pipeline->execute($noncallable);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteInsideCallableArgumentThrowsException(): void
|
|
{
|
|
$this->expectException('Predis\ClientException');
|
|
|
|
$pipeline = new Pipeline(new Client());
|
|
|
|
$pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->execute();
|
|
});
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteWithCallableArgumentRunsPipelineInCallable(): void
|
|
{
|
|
$command1 = new ECHO_();
|
|
$command1->setArguments(['one']);
|
|
|
|
$command2 = new ECHO_();
|
|
$command2->setArguments(['two']);
|
|
|
|
$command3 = new ECHO_();
|
|
$command3->setArguments(['three']);
|
|
|
|
$command4 = new ECHO_();
|
|
$command4->setArguments(['four']);
|
|
|
|
$buffer = $command1->serializeCommand()
|
|
. $command2->serializeCommand()
|
|
. $command3->serializeCommand()
|
|
. $command4->serializeCommand();
|
|
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->once())
|
|
->method('write')
|
|
->with($buffer);
|
|
$connection
|
|
->expects($this->exactly(4))
|
|
->method('readResponse')
|
|
->willReturnCallback($this->getReadCallback());
|
|
$connection
|
|
->expects($this->exactly(2))
|
|
->method('getParameters')
|
|
->willReturn(new Parameters(['protocol' => 2]));
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->echo('one');
|
|
$pipe->echo('two');
|
|
$pipe->echo('three');
|
|
$pipe->echo('four');
|
|
});
|
|
|
|
$this->assertSame(['one', 'two', 'three', 'four'], $responses);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
*/
|
|
public function testExecuteWithCallableArgumentHandlesExceptions(): void
|
|
{
|
|
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
|
$connection
|
|
->expects($this->never())
|
|
->method('writeRequest');
|
|
$connection
|
|
->expects($this->never())
|
|
->method('readResponse');
|
|
|
|
$exception = null;
|
|
$responses = null;
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
try {
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->echo('one');
|
|
$pipe->echo('two');
|
|
throw new ClientException('TEST');
|
|
});
|
|
} catch (Exception $exception) {
|
|
// NOOP
|
|
}
|
|
|
|
$this->assertInstanceOf('Predis\ClientException', $exception);
|
|
$this->assertSame('TEST', $exception->getMessage());
|
|
$this->assertNull($responses);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
* @throws Exception
|
|
*/
|
|
public function testRetryStandalonePipelineOnRetryableErrors(): void
|
|
{
|
|
$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->exactly(3))
|
|
->method('read')
|
|
->withAnyParameters()
|
|
->willReturn("+PONG\r\n");
|
|
|
|
$mockStreamFactory
|
|
->expects($this->exactly(4))
|
|
->method('createStream')
|
|
->withAnyParameters()
|
|
->willReturn($mockStream);
|
|
|
|
$connection = new StreamConnection($parameters, $mockStreamFactory);
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->ping();
|
|
$pipe->ping();
|
|
$pipe->ping();
|
|
});
|
|
|
|
$this->assertEquals(['PONG', 'PONG', 'PONG'], $responses);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
* @throws Exception
|
|
*/
|
|
public function testRetryClusterPipelineOnRetryableErrors(): void
|
|
{
|
|
$mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
|
|
$mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
|
|
$mockConnectionFactory = $this->getMockBuilder(Factory::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(6))
|
|
->method('write')
|
|
->withAnyParameters()
|
|
->willReturnOnConsecutiveCalls(
|
|
$this->throwException(new RuntimeException('', 2)),
|
|
$this->throwException(new RuntimeException('', 2)),
|
|
$this->throwException(new RuntimeException('', 2)),
|
|
1000,
|
|
1000,
|
|
1000
|
|
);
|
|
|
|
$mockStream
|
|
->expects($this->exactly(3))
|
|
->method('read')
|
|
->withAnyParameters()
|
|
->willReturn("+OK\r\n");
|
|
|
|
$mockStreamFactory
|
|
->expects($this->exactly(4))
|
|
->method('createStream')
|
|
->withAnyParameters()
|
|
->willReturn($mockStream);
|
|
|
|
$streamConnection = new StreamConnection($parameters, $mockStreamFactory);
|
|
$connection = new RedisCluster($mockConnectionFactory, $parameters);
|
|
$connection->add($streamConnection);
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
});
|
|
|
|
$this->assertEquals(['OK', 'OK', 'OK'], $responses);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
* @throws Exception
|
|
*/
|
|
public function testExecutePipelineInvokesOnAggregateConnectionFailCallbackOnConnectionException(): void
|
|
{
|
|
$mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
|
|
$mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
|
|
$mockConnectionFactory = $this->getMockBuilder(Factory::class)->getMock();
|
|
$parameters = new Parameters([
|
|
'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
|
|
]);
|
|
|
|
$streamConnection = new StreamConnection($parameters, $mockStreamFactory);
|
|
$connection = $this->getMockBuilder(RedisCluster::class)
|
|
->setConstructorArgs([$mockConnectionFactory, $parameters])
|
|
->onlyMethods(['getConnectionByCommand', 'remove', 'askSlotMap'])
|
|
->getMock();
|
|
|
|
// Disable useClusterSlots to avoid askSlotMap() calls
|
|
$connection->useClusterSlots = false;
|
|
|
|
$mockStream
|
|
->expects($this->atLeast(3))
|
|
->method('close')
|
|
->withAnyParameters();
|
|
|
|
$mockStream
|
|
->expects($this->exactly(6))
|
|
->method('write')
|
|
->withAnyParameters()
|
|
->willReturnOnConsecutiveCalls(
|
|
$this->throwException(new \Predis\Connection\ConnectionException($streamConnection, 'Connection failed')),
|
|
$this->throwException(new \Predis\Connection\ConnectionException($streamConnection, 'Connection failed')),
|
|
$this->throwException(new \Predis\Connection\ConnectionException($streamConnection, 'Connection failed')),
|
|
1000,
|
|
1000,
|
|
1000
|
|
);
|
|
|
|
$mockStream
|
|
->expects($this->exactly(3))
|
|
->method('read')
|
|
->withAnyParameters()
|
|
->willReturn("+OK\r\n");
|
|
|
|
$mockStreamFactory
|
|
->expects($this->exactly(4))
|
|
->method('createStream')
|
|
->withAnyParameters()
|
|
->willReturn($mockStream);
|
|
|
|
$connection
|
|
->expects($this->exactly(9))
|
|
->method('getConnectionByCommand')
|
|
->willReturn($streamConnection);
|
|
|
|
// Verify that remove() is called on the aggregate connection during retry
|
|
$connection
|
|
->expects($this->exactly(3))
|
|
->method('remove')
|
|
->with($streamConnection);
|
|
|
|
// Verify that askSlotMap() is NOT called since useClusterSlots is false
|
|
$connection
|
|
->expects($this->never())
|
|
->method('askSlotMap');
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
});
|
|
|
|
$this->assertEquals(['OK', 'OK', 'OK'], $responses);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
* @throws Exception
|
|
*/
|
|
public function testExecutePipelineInvokesOnAggregateConnectionFailCallbackOnTimeoutException(): void
|
|
{
|
|
$mockStream = $this->getMockBuilder(StreamInterface::class)->getMock();
|
|
$mockStreamFactory = $this->getMockBuilder(StreamFactoryInterface::class)->getMock();
|
|
$mockConnectionFactory = $this->getMockBuilder(Factory::class)->getMock();
|
|
$parameters = new Parameters([
|
|
'retry' => new Retry(new ExponentialBackoff(1000, 10000), 3),
|
|
]);
|
|
|
|
$streamConnection = new StreamConnection($parameters, $mockStreamFactory);
|
|
$connection = $this->getMockBuilder(RedisCluster::class)
|
|
->setConstructorArgs([$mockConnectionFactory, $parameters])
|
|
->onlyMethods(['getConnectionByCommand', 'remove'])
|
|
->getMock();
|
|
|
|
// Disable useClusterSlots to avoid askSlotMap() calls
|
|
$connection->useClusterSlots = false;
|
|
|
|
$mockStream
|
|
->expects($this->atLeast(3))
|
|
->method('close')
|
|
->withAnyParameters();
|
|
|
|
$mockStream
|
|
->expects($this->exactly(6))
|
|
->method('write')
|
|
->withAnyParameters()
|
|
->willReturnOnConsecutiveCalls(
|
|
$this->throwException(new TimeoutException($streamConnection, 0)),
|
|
$this->throwException(new TimeoutException($streamConnection, 0)),
|
|
$this->throwException(new TimeoutException($streamConnection, 0)),
|
|
1000,
|
|
1000,
|
|
1000
|
|
);
|
|
|
|
$mockStream
|
|
->expects($this->exactly(3))
|
|
->method('read')
|
|
->withAnyParameters()
|
|
->willReturn("+OK\r\n");
|
|
|
|
$mockStreamFactory
|
|
->expects($this->exactly(4))
|
|
->method('createStream')
|
|
->withAnyParameters()
|
|
->willReturn($mockStream);
|
|
|
|
$connection
|
|
->expects($this->exactly(9))
|
|
->method('getConnectionByCommand')
|
|
->willReturn($streamConnection);
|
|
|
|
// Verify that remove() is NOT called for TimeoutException
|
|
$connection
|
|
->expects($this->never())
|
|
->method('remove');
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
});
|
|
|
|
$this->assertEquals(['OK', 'OK', 'OK'], $responses);
|
|
}
|
|
|
|
/**
|
|
* @group disconnected
|
|
* @throws Exception
|
|
*/
|
|
public function testRetryReplicationPipelineOnRetryableErrors(): void
|
|
{
|
|
$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->atLeast(3))
|
|
->method('close')
|
|
->withAnyParameters();
|
|
|
|
$mockStream
|
|
->expects($this->exactly(6))
|
|
->method('write')
|
|
->withAnyParameters()
|
|
->willReturnOnConsecutiveCalls(
|
|
$this->throwException(new RuntimeException('', 2)),
|
|
$this->throwException(new RuntimeException('', 2)),
|
|
$this->throwException(new RuntimeException('', 2)),
|
|
1000,
|
|
1000,
|
|
1000
|
|
);
|
|
|
|
$mockStream
|
|
->expects($this->exactly(3))
|
|
->method('read')
|
|
->withAnyParameters()
|
|
->willReturn("+OK\r\n");
|
|
|
|
$mockStreamFactory
|
|
->expects($this->exactly(4))
|
|
->method('createStream')
|
|
->withAnyParameters()
|
|
->willReturn($mockStream);
|
|
|
|
$streamConnection = new StreamConnection($parameters, $mockStreamFactory);
|
|
|
|
$connection = new MasterSlaveReplication(new ReplicationStrategy());
|
|
$connection->add($streamConnection);
|
|
|
|
$pipeline = new Pipeline(new Client($connection));
|
|
|
|
$responses = $pipeline->execute(static function (Pipeline $pipe) {
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
$pipe->set('key', 'value');
|
|
});
|
|
|
|
$this->assertEquals(['OK', 'OK', 'OK'], $responses);
|
|
}
|
|
|
|
// ******************************************************************** //
|
|
// ---- INTEGRATION TESTS --------------------------------------------- //
|
|
// ******************************************************************** //
|
|
|
|
/**
|
|
* @group connected
|
|
*/
|
|
public function testIntegrationWithFluentInterface(): void
|
|
{
|
|
$pipeline = $this->getClient()->pipeline();
|
|
|
|
$results = $pipeline
|
|
->echo('one')
|
|
->echo('two')
|
|
->echo('three')
|
|
->execute();
|
|
|
|
$this->assertSame(['one', 'two', 'three'], $results);
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
* @requiresRedisVersion >= 6.0.0
|
|
*/
|
|
public function testIntegrationWithFluentInterfaceResp3(): void
|
|
{
|
|
$pipeline = $this->getClient(['protocol' => 3])->pipeline();
|
|
|
|
$results = $pipeline
|
|
->echo('one')
|
|
->echo('two')
|
|
->echo('three')
|
|
->execute();
|
|
|
|
$this->assertSame(['one', 'two', 'three'], $results);
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
*/
|
|
public function testIntegrationWithCallableBlock(): void
|
|
{
|
|
$client = $this->getClient();
|
|
|
|
$results = $client->pipeline(static function (Pipeline $pipe) {
|
|
$pipe->set('foo', 'bar');
|
|
$pipe->get('foo');
|
|
});
|
|
|
|
$this->assertEquals(['OK', 'bar'], $results);
|
|
$this->assertSame(1, $client->exists('foo'));
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
*/
|
|
public function testOutOfBandMessagesInsidePipeline(): void
|
|
{
|
|
$oob = null;
|
|
$client = $this->getClient();
|
|
|
|
$results = $client->pipeline(static function (Pipeline $pipe) use (&$oob) {
|
|
$pipe->set('foo', 'bar');
|
|
$oob = $pipe->getClient()->echo('oob message');
|
|
$pipe->get('foo');
|
|
});
|
|
|
|
$this->assertEquals(['OK', 'bar'], $results);
|
|
$this->assertSame('oob message', $oob);
|
|
$this->assertSame(1, $client->exists('foo'));
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
*/
|
|
public function testIntegrationWithClientExceptionInCallableBlock(): void
|
|
{
|
|
$exception = null;
|
|
|
|
$client = $this->getClient();
|
|
|
|
try {
|
|
$client->pipeline(static function (Pipeline $pipe) {
|
|
$pipe->set('foo', 'bar');
|
|
throw new ClientException('TEST');
|
|
});
|
|
} catch (Exception $exception) {
|
|
// NOOP
|
|
}
|
|
|
|
$this->assertInstanceOf('Predis\ClientException', $exception);
|
|
$this->assertSame('TEST', $exception->getMessage());
|
|
$this->assertSame(0, $client->exists('foo'));
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
*/
|
|
public function testIntegrationWithServerExceptionInCallableBlock(): void
|
|
{
|
|
$exception = null;
|
|
|
|
$client = $this->getClient();
|
|
|
|
try {
|
|
$client->pipeline(static function (Pipeline $pipe) {
|
|
$pipe->set('foo', 'bar');
|
|
// LPUSH on a string key fails, but won't stop
|
|
// the pipeline to send the commands.
|
|
$pipe->lpush('foo', 'bar');
|
|
$pipe->set('hoge', 'piyo');
|
|
});
|
|
} catch (Exception $exception) {
|
|
// NOOP
|
|
}
|
|
|
|
$this->assertInstanceOf('Predis\Response\ServerException', $exception);
|
|
$this->assertSame(1, $client->exists('foo'));
|
|
$this->assertSame(1, $client->exists('hoge'));
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
*/
|
|
public function testIntegrationWithServerErrorInCallableBlock(): void
|
|
{
|
|
$client = $this->getClient([], ['exceptions' => false]);
|
|
|
|
$results = $client->pipeline(static function (Pipeline $pipe) {
|
|
$pipe->set('foo', 'bar');
|
|
$pipe->lpush('foo', 'bar'); // LPUSH on a string key fails.
|
|
$pipe->get('foo');
|
|
});
|
|
|
|
$this->assertEquals('OK', $results[0]);
|
|
$this->assertInstanceOf('Predis\Response\Error', $results[1]);
|
|
$this->assertSame('bar', $results[2]);
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
* @group cluster
|
|
* @group relay-incompatible
|
|
* @requiresRedisVersion >= 6.2.0
|
|
*/
|
|
public function testClusterExecutePipeline(): void
|
|
{
|
|
$client = $this->getClient();
|
|
|
|
$results = $client->pipeline(static function (Pipeline $pipe) {
|
|
$pipe->set('foo', 'bar');
|
|
$pipe->set('bar', 'foo');
|
|
$pipe->set('baz', 'baz');
|
|
$pipe->get('foo');
|
|
$pipe->get('bar');
|
|
$pipe->get('baz');
|
|
});
|
|
|
|
$expectedResults = [
|
|
new Response\Status('OK'),
|
|
new Response\Status('OK'),
|
|
new Response\Status('OK'),
|
|
'bar',
|
|
'foo',
|
|
'baz',
|
|
];
|
|
|
|
$this->assertSameValues($expectedResults, $results);
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
* @group relay-incompatible
|
|
* @requiresRedisVersion >= 6.2.0
|
|
* @return void
|
|
*/
|
|
public function testStandaloneRetryPipelineOnTimeoutException(): void
|
|
{
|
|
$client = $this->getClient([
|
|
'retry' => new Retry(new ExponentialBackoff(100, 1000), 3),
|
|
'read_write_timeout' => 0.1,
|
|
]);
|
|
|
|
$this->expectException(TimeoutException::class);
|
|
|
|
$client->pipeline(static function (Pipeline $pipe) use (&$retries) {
|
|
$pipe->incr('test_key');
|
|
$pipe->blpop('foo', 3);
|
|
});
|
|
$this->assertEquals(3, $client->get('test_key'));
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
* @group cluster
|
|
* @group relay-incompatible
|
|
* @requiresRedisVersion >= 6.2.0
|
|
* @return void
|
|
*/
|
|
public function testClusterRetryPipelineOnTimeoutException(): void
|
|
{
|
|
$retries = 0;
|
|
$client = $this->getClient([], [
|
|
'parameters' => [
|
|
'retry' => new Retry(new ExponentialBackoff(100, 1000), 3),
|
|
'read_write_timeout' => 0.1,
|
|
],
|
|
]);
|
|
|
|
$this->expectException(TimeoutException::class);
|
|
|
|
$client->pipeline(static function (Pipeline $pipe) use (&$retries) {
|
|
++$retries;
|
|
$pipe->blpop('foo', 3);
|
|
});
|
|
$this->assertEquals(3, $retries);
|
|
}
|
|
|
|
/**
|
|
* @group connected
|
|
* @group relay-incompatible
|
|
*/
|
|
public function testReplicationExecutesPipelineWithCRLFValues(): void
|
|
{
|
|
$parameters = $this->getDefaultParametersArray();
|
|
|
|
$client = new Client(
|
|
["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
|
|
['replication' => 'predis']
|
|
);
|
|
|
|
$results = $client->pipeline(static function (Pipeline $pipe) {
|
|
$pipe->set('foo', "bar\r\nbaz");
|
|
$pipe->get('foo');
|
|
});
|
|
|
|
$expectedResults = [
|
|
new Response\Status('OK'),
|
|
"bar\r\nbaz",
|
|
];
|
|
|
|
$this->assertSameValues($expectedResults, $results);
|
|
}
|
|
|
|
// ******************************************************************** //
|
|
// ---- HELPER METHODS ------------------------------------------------ //
|
|
// ******************************************************************** //
|
|
|
|
/**
|
|
* Returns a client instance connected to the specified Redis server.
|
|
*
|
|
* @param array $parameters Additional connection parameters
|
|
* @param array $options Additional client options
|
|
*
|
|
* @return ClientInterface
|
|
*/
|
|
protected function getClient(array $parameters = [], array $options = []): ClientInterface
|
|
{
|
|
return $this->createClient($parameters, $options);
|
|
}
|
|
|
|
/**
|
|
* Helper method returning a callback used to responses to ECHO command.
|
|
*
|
|
* @return callable
|
|
*/
|
|
protected function getReadCallback(): callable
|
|
{
|
|
return static function (CommandInterface $command) {
|
|
if (($id = $command->getId()) !== 'ECHO') {
|
|
throw new InvalidArgumentException("Expected ECHO, got {$id}");
|
|
}
|
|
|
|
[$echoed] = $command->getArguments();
|
|
|
|
return $echoed;
|
|
};
|
|
}
|
|
}
|