mirror of
https://github.com/predis/predis.git
synced 2026-09-12 19:37:05 +00:00
Improved pipeline abstractions (#1438)
* Added pipelining for on-connection commands * Added server version condition for tests * Fixed static analysis and codestyle errors * Improved handshake session * Updated HELLO retry logic * Changed incorrect variable * Added return statement * Added more test coverage, codestyle fixes * Improved pipelining abstractions * Fixed unit tests * Fixed typo * Fixed another typo * Fixed another typo * Updated deserializeCommand signature * Codestyle fixes * Codestyle fixes * Changed pipelined sequence * Removed redundant method * Fixed classname
This commit is contained in:
Vladyslav Vildanov
committed by
GitHub
GitHub
parent
8b464f2a60
commit
40d72d716d
@@ -12,6 +12,9 @@
|
||||
|
||||
namespace Predis\Command;
|
||||
|
||||
use Predis\ClientConfiguration;
|
||||
use UnexpectedValueException;
|
||||
|
||||
/**
|
||||
* Base class for Redis commands.
|
||||
*/
|
||||
@@ -152,4 +155,46 @@ abstract class Command implements CommandInterface
|
||||
|
||||
return $buffer;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
public static function deserializeCommand(string $serializedCommand): CommandInterface
|
||||
{
|
||||
if ($serializedCommand[0] !== '*') {
|
||||
throw new UnexpectedValueException('Invalid serializing format');
|
||||
}
|
||||
|
||||
$commandArray = explode("\r\n", $serializedCommand);
|
||||
$commandId = $commandArray[2];
|
||||
$classPath = __NAMESPACE__ . '\Redis\\';
|
||||
|
||||
// Check if given command is a module command.
|
||||
if (count($commandIdArray = explode('.', $commandId)) > 1) {
|
||||
// Fetch module configuration to resolve namespace.
|
||||
$moduleConfiguration = array_filter(
|
||||
ClientConfiguration::getModules(),
|
||||
static function ($module) use ($commandIdArray) {
|
||||
return $module['commandPrefix'] === $commandIdArray[0];
|
||||
}
|
||||
);
|
||||
|
||||
$commandClass = strtoupper($commandIdArray[0] . $commandIdArray[1]);
|
||||
$classPath .= array_shift($moduleConfiguration)['name'] . '\\' . $commandClass;
|
||||
} else {
|
||||
$classPath .= $commandIdArray[0];
|
||||
}
|
||||
|
||||
$command = new $classPath();
|
||||
$arguments = [];
|
||||
|
||||
for ($i = 4, $iMax = count($commandArray); $i < $iMax; $i++) {
|
||||
$arguments[] = $commandArray[$i];
|
||||
++$i;
|
||||
}
|
||||
|
||||
$command->setArguments($arguments);
|
||||
|
||||
return $command;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -92,4 +92,12 @@ interface CommandInterface
|
||||
* @return string
|
||||
*/
|
||||
public function serializeCommand(): string;
|
||||
|
||||
/**
|
||||
* Creates command object from given serialized representation.
|
||||
*
|
||||
* @param string $serializedCommand
|
||||
* @return static
|
||||
*/
|
||||
public static function deserializeCommand(string $serializedCommand): CommandInterface;
|
||||
}
|
||||
|
||||
@@ -12,6 +12,9 @@
|
||||
|
||||
namespace Predis\Command;
|
||||
|
||||
use Predis\ClientConfiguration;
|
||||
use UnexpectedValueException;
|
||||
|
||||
/**
|
||||
* Class representing a generic Redis command.
|
||||
*
|
||||
@@ -149,4 +152,43 @@ final class RawCommand implements CommandInterface
|
||||
|
||||
return $buffer;
|
||||
}
|
||||
|
||||
public static function deserializeCommand(string $serializedCommand): CommandInterface
|
||||
{
|
||||
if ($serializedCommand[0] !== '*') {
|
||||
throw new UnexpectedValueException('Invalid serializing format');
|
||||
}
|
||||
|
||||
$commandArray = explode("\r\n", $serializedCommand);
|
||||
$commandId = $commandArray[2];
|
||||
$classPath = __NAMESPACE__ . '\Redis\\';
|
||||
|
||||
// Check if given command is a module command.
|
||||
if (count($commandIdArray = explode('.', $commandId)) > 1) {
|
||||
// Fetch module configuration to resolve namespace.
|
||||
$moduleConfiguration = array_filter(
|
||||
ClientConfiguration::getModules(),
|
||||
static function ($module) use ($commandIdArray) {
|
||||
return $module['commandPrefix'] === $commandIdArray[0];
|
||||
}
|
||||
);
|
||||
|
||||
$commandClass = strtoupper($commandIdArray[0] . $commandIdArray[1]);
|
||||
$classPath .= array_shift($moduleConfiguration)['name'] . '\\' . $commandClass;
|
||||
} else {
|
||||
$classPath .= $commandIdArray[0];
|
||||
}
|
||||
|
||||
$command = new $classPath();
|
||||
$arguments = [];
|
||||
|
||||
for ($i = 4, $iMax = count($commandArray); $i < $iMax; $i++) {
|
||||
$arguments[] = $commandArray[$i];
|
||||
++$i;
|
||||
}
|
||||
|
||||
$command->setArguments($arguments);
|
||||
|
||||
return $command;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the Predis package.
|
||||
*
|
||||
* (c) 2009-2020 Daniele Alessandri
|
||||
* (c) 2021-2023 Till Krüss
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Predis\Connection;
|
||||
|
||||
use Predis\Command\Command;
|
||||
use Predis\Command\CommandInterface;
|
||||
|
||||
abstract class AbstractAggregateConnection implements AggregateConnectionInterface
|
||||
{
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function add(NodeConnectionInterface $connection);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function remove(NodeConnectionInterface $connection);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function getConnectionByCommand(CommandInterface $command);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function getConnectionById($connectionID);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function connect();
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function disconnect();
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function isConnected();
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function writeRequest(CommandInterface $command);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function readResponse(CommandInterface $command);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function executeCommand(CommandInterface $command);
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
abstract public function getParameters();
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*/
|
||||
public function write(string $buffer): void
|
||||
{
|
||||
$rawCommands = [];
|
||||
$explodedBuffer = explode("\r\n", trim($buffer));
|
||||
|
||||
while (!empty($explodedBuffer)) {
|
||||
$argsLen = (int) explode('*', $explodedBuffer[0])[1];
|
||||
$cmdLen = ($argsLen * 2) + 1;
|
||||
$rawCommands[] = array_splice($explodedBuffer, 0, $cmdLen);
|
||||
}
|
||||
|
||||
foreach ($rawCommands as $command) {
|
||||
$command = implode("\r\n", $command) . "\r\n";
|
||||
$commandObj = Command::deserializeCommand($command);
|
||||
$this->getConnectionByCommand($commandObj)->write($command);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -17,7 +17,9 @@ use Countable;
|
||||
use IteratorAggregate;
|
||||
use Predis\Cluster\PredisStrategy;
|
||||
use Predis\Cluster\StrategyInterface;
|
||||
use Predis\Command\Command;
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\Connection\AbstractAggregateConnection;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
use Predis\Connection\ParametersInterface;
|
||||
use Predis\NotSupportedException;
|
||||
@@ -28,7 +30,7 @@ use Traversable;
|
||||
* Abstraction for a cluster of aggregate connections to various Redis servers
|
||||
* implementing client-side sharding based on pluggable distribution strategies.
|
||||
*/
|
||||
class PredisCluster implements ClusterInterface, IteratorAggregate, Countable
|
||||
class PredisCluster extends AbstractAggregateConnection implements ClusterInterface, IteratorAggregate, Countable
|
||||
{
|
||||
/**
|
||||
* @var NodeConnectionInterface[]
|
||||
|
||||
@@ -20,8 +20,10 @@ use Predis\ClientException;
|
||||
use Predis\Cluster\RedisStrategy as RedisClusterStrategy;
|
||||
use Predis\Cluster\SlotMap;
|
||||
use Predis\Cluster\StrategyInterface;
|
||||
use Predis\Command\Command;
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\Command\RawCommand;
|
||||
use Predis\Connection\AbstractAggregateConnection;
|
||||
use Predis\Connection\ConnectionException;
|
||||
use Predis\Connection\FactoryInterface;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
@@ -54,7 +56,7 @@ use Traversable;
|
||||
* Asking for the cluster configuration to Redis is actually done by issuing a
|
||||
* CLUSTER SLOTS command to a random node in the pool.
|
||||
*/
|
||||
class RedisCluster implements ClusterInterface, IteratorAggregate, Countable
|
||||
class RedisCluster extends AbstractAggregateConnection implements ClusterInterface, IteratorAggregate, Countable
|
||||
{
|
||||
private $useClusterSlots = true;
|
||||
|
||||
|
||||
@@ -53,6 +53,15 @@ interface ConnectionInterface
|
||||
*/
|
||||
public function readResponse(CommandInterface $command);
|
||||
|
||||
/**
|
||||
* Performs a write operation over the stream of the buffer containing a
|
||||
* command serialized with the Redis wire protocol.
|
||||
*
|
||||
* @param string $buffer
|
||||
* @return void
|
||||
*/
|
||||
public function write(string $buffer): void;
|
||||
|
||||
/**
|
||||
* Writes a request for the given command over the connection and reads back
|
||||
* the response returned by Redis.
|
||||
|
||||
@@ -14,8 +14,10 @@ namespace Predis\Connection\Replication;
|
||||
|
||||
use InvalidArgumentException;
|
||||
use Predis\ClientException;
|
||||
use Predis\Command\Command;
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\Command\RawCommand;
|
||||
use Predis\Connection\AbstractAggregateConnection;
|
||||
use Predis\Connection\ConnectionException;
|
||||
use Predis\Connection\FactoryInterface;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
@@ -28,7 +30,7 @@ use Predis\Response\ErrorInterface as ResponseErrorInterface;
|
||||
* Aggregate connection handling replication of Redis nodes configured in a
|
||||
* single master / multiple slaves setup.
|
||||
*/
|
||||
class MasterSlaveReplication implements ReplicationInterface
|
||||
class MasterSlaveReplication extends AbstractAggregateConnection implements ReplicationInterface
|
||||
{
|
||||
/**
|
||||
* @var ReplicationStrategy
|
||||
|
||||
@@ -13,9 +13,11 @@
|
||||
namespace Predis\Connection\Replication;
|
||||
|
||||
use InvalidArgumentException;
|
||||
use Predis\Command\Command;
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\Command\RawCommand;
|
||||
use Predis\CommunicationException;
|
||||
use Predis\Connection\AbstractAggregateConnection;
|
||||
use Predis\Connection\ConnectionException;
|
||||
use Predis\Connection\FactoryInterface as ConnectionFactoryInterface;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
@@ -31,7 +33,7 @@ use Predis\Response\ServerException;
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
* @author Ville Mattila <ville@eventio.fi>
|
||||
*/
|
||||
class SentinelReplication implements ReplicationInterface
|
||||
class SentinelReplication extends AbstractAggregateConnection implements ReplicationInterface
|
||||
{
|
||||
/**
|
||||
* @var NodeConnectionInterface
|
||||
|
||||
@@ -63,11 +63,14 @@ class Atomic extends Pipeline
|
||||
{
|
||||
$commandFactory = $this->getClient()->getCommandFactory();
|
||||
$connection->executeCommand($commandFactory->create('multi'));
|
||||
$buffer = '';
|
||||
|
||||
foreach ($commands as $command) {
|
||||
$connection->writeRequest($command);
|
||||
$buffer .= $command->serializeCommand();
|
||||
}
|
||||
|
||||
$connection->write($buffer);
|
||||
|
||||
foreach ($commands as $command) {
|
||||
$response = $connection->readResponse($command);
|
||||
|
||||
|
||||
@@ -56,13 +56,16 @@ class ConnectionErrorProof extends Pipeline
|
||||
{
|
||||
$responses = [];
|
||||
$sizeOfPipe = count($commands);
|
||||
$buffer = '';
|
||||
|
||||
foreach ($commands as $command) {
|
||||
try {
|
||||
$connection->writeRequest($command);
|
||||
} catch (CommunicationException $exception) {
|
||||
return array_fill(0, $sizeOfPipe, $exception);
|
||||
}
|
||||
$buffer .= $command->serializeCommand();
|
||||
}
|
||||
|
||||
try {
|
||||
$connection->write($buffer);
|
||||
} catch (CommunicationException $exception) {
|
||||
return array_fill(0, $sizeOfPipe, $exception);
|
||||
}
|
||||
|
||||
for ($i = 0; $i < $sizeOfPipe; ++$i) {
|
||||
@@ -89,21 +92,14 @@ class ConnectionErrorProof extends Pipeline
|
||||
$responses = [];
|
||||
$sizeOfPipe = count($commands);
|
||||
$exceptions = [];
|
||||
$buffer = '';
|
||||
|
||||
foreach ($commands as $command) {
|
||||
$cmdConnection = $connection->getConnectionByCommand($command);
|
||||
|
||||
if (isset($exceptions[spl_object_hash($cmdConnection)])) {
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
$cmdConnection->writeRequest($command);
|
||||
} catch (CommunicationException $exception) {
|
||||
$exceptions[spl_object_hash($cmdConnection)] = $exception;
|
||||
}
|
||||
$buffer .= $command->serializeCommand();
|
||||
}
|
||||
|
||||
$connection->write($buffer);
|
||||
|
||||
for ($i = 0; $i < $sizeOfPipe; ++$i) {
|
||||
$command = $commands->dequeue();
|
||||
|
||||
|
||||
@@ -25,10 +25,13 @@ class FireAndForget extends Pipeline
|
||||
*/
|
||||
protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
|
||||
{
|
||||
$buffer = '';
|
||||
|
||||
while (!$commands->isEmpty()) {
|
||||
$connection->writeRequest($commands->dequeue());
|
||||
$buffer .= $commands->dequeue()->serializeCommand();
|
||||
}
|
||||
|
||||
$connection->write($buffer);
|
||||
$connection->disconnect();
|
||||
|
||||
return [];
|
||||
|
||||
@@ -131,10 +131,14 @@ class Pipeline implements ClientContextInterface
|
||||
*/
|
||||
protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
|
||||
{
|
||||
$buffer = '';
|
||||
|
||||
foreach ($commands as $command) {
|
||||
$connection->writeRequest($command);
|
||||
$buffer .= $command->serializeCommand();
|
||||
}
|
||||
|
||||
$connection->write($buffer);
|
||||
|
||||
$responses = [];
|
||||
$exceptions = $this->throwServerExceptions();
|
||||
$protocolVersion = (int) $connection->getParameters()->protocol;
|
||||
|
||||
@@ -22,7 +22,7 @@ abstract class PredisCommandTestCase extends PredisTestCase
|
||||
/**
|
||||
* Returns the expected command for tests.
|
||||
*
|
||||
* @return Command\CommandInterface|string Instance or FQCN of the expected command
|
||||
* @return CommandInterface|string Instance or FQCN of the expected command
|
||||
*/
|
||||
abstract protected function getExpectedCommand(): string;
|
||||
|
||||
|
||||
@@ -12,8 +12,20 @@
|
||||
|
||||
namespace Predis\Command;
|
||||
|
||||
use Predis\Command\Redis\BloomFilter\BFADD;
|
||||
use Predis\Command\Redis\CountMinSketch\CMSINFO;
|
||||
use Predis\Command\Redis\CuckooFilter\CFADD;
|
||||
use Predis\Command\Redis\GET;
|
||||
use Predis\Command\Redis\Json\JSONSET;
|
||||
use Predis\Command\Redis\MGET;
|
||||
use Predis\Command\Redis\Search\FTSEARCH;
|
||||
use Predis\Command\Redis\TDigest\TDIGESTADD;
|
||||
use Predis\Command\Redis\TimeSeries\TSGET;
|
||||
use Predis\Command\Redis\TopK\TOPKQUERY;
|
||||
use Predis\Command\Redis\ZADD;
|
||||
use PredisTestCase;
|
||||
use stdClass;
|
||||
use UnexpectedValueException;
|
||||
|
||||
class CommandTest extends PredisTestCase
|
||||
{
|
||||
@@ -179,4 +191,81 @@ class CommandTest extends PredisTestCase
|
||||
$command->serializeCommand()
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* @dataProvider deserializeCommandProvider
|
||||
* @group disconnected
|
||||
*/
|
||||
public function testDeserializeCommand(string $class, array $arguments): void
|
||||
{
|
||||
$command = new $class();
|
||||
$command->setArguments($arguments);
|
||||
|
||||
$deserializedCommand = Command::deserializeCommand($command->serializeCommand());
|
||||
|
||||
$this->assertInstanceOf($class, $deserializedCommand);
|
||||
$this->assertSame($command->getArguments(), $deserializedCommand->getArguments());
|
||||
}
|
||||
|
||||
/**
|
||||
* @group disconnected
|
||||
* @return void
|
||||
*/
|
||||
public function testDeserializeCommandThrowsException(): void
|
||||
{
|
||||
$this->expectException(UnexpectedValueException::class);
|
||||
$this->expectExceptionMessage('Invalid serializing format');
|
||||
|
||||
Command::deserializeCommand('foobar');
|
||||
}
|
||||
|
||||
public function deserializeCommandProvider(): array
|
||||
{
|
||||
return [
|
||||
'GET' => [
|
||||
GET::class,
|
||||
['arg1'],
|
||||
],
|
||||
'MGET' => [
|
||||
MGET::class,
|
||||
['arg1', 'arg2', 'arg3'],
|
||||
],
|
||||
'ZADD' => [
|
||||
ZADD::class,
|
||||
['key', 'value', 'key1', 'value1'],
|
||||
],
|
||||
'JSONSET' => [
|
||||
JSONSET::class,
|
||||
['key', '$', '{"key":"value"}'],
|
||||
],
|
||||
'BFADD' => [
|
||||
BFADD::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'CFADD' => [
|
||||
CFADD::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'CMSINFO' => [
|
||||
CMSINFO::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'FTSEARCH' => [
|
||||
FTSEARCH::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'TDIGESTADD' => [
|
||||
TDIGESTADD::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'TSGET' => [
|
||||
TSGET::class,
|
||||
['key'],
|
||||
],
|
||||
'TOPKQUERY' => [
|
||||
TOPKQUERY::class,
|
||||
['key'],
|
||||
],
|
||||
];
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,7 +12,19 @@
|
||||
|
||||
namespace Predis\Command;
|
||||
|
||||
use Predis\Command\Redis\BloomFilter\BFADD;
|
||||
use Predis\Command\Redis\CountMinSketch\CMSINFO;
|
||||
use Predis\Command\Redis\CuckooFilter\CFADD;
|
||||
use Predis\Command\Redis\GET;
|
||||
use Predis\Command\Redis\Json\JSONSET;
|
||||
use Predis\Command\Redis\MGET;
|
||||
use Predis\Command\Redis\Search\FTSEARCH;
|
||||
use Predis\Command\Redis\TDigest\TDIGESTADD;
|
||||
use Predis\Command\Redis\TimeSeries\TSGET;
|
||||
use Predis\Command\Redis\TopK\TOPKQUERY;
|
||||
use Predis\Command\Redis\ZADD;
|
||||
use PredisTestCase;
|
||||
use UnexpectedValueException;
|
||||
|
||||
class RawCommandTest extends PredisTestCase
|
||||
{
|
||||
@@ -151,4 +163,81 @@ class RawCommandTest extends PredisTestCase
|
||||
$command->serializeCommand()
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* @dataProvider deserializeCommandProvider
|
||||
* @group disconnected
|
||||
*/
|
||||
public function testDeserializeCommand(string $class, array $arguments): void
|
||||
{
|
||||
$command = new $class();
|
||||
$command->setArguments($arguments);
|
||||
|
||||
$deserializedCommand = RawCommand::deserializeCommand($command->serializeCommand());
|
||||
|
||||
$this->assertInstanceOf($class, $deserializedCommand);
|
||||
$this->assertSame($command->getArguments(), $deserializedCommand->getArguments());
|
||||
}
|
||||
|
||||
/**
|
||||
* @group disconnected
|
||||
* @return void
|
||||
*/
|
||||
public function testDeserializeCommandThrowsException(): void
|
||||
{
|
||||
$this->expectException(UnexpectedValueException::class);
|
||||
$this->expectExceptionMessage('Invalid serializing format');
|
||||
|
||||
RawCommand::deserializeCommand('foobar');
|
||||
}
|
||||
|
||||
public function deserializeCommandProvider(): array
|
||||
{
|
||||
return [
|
||||
'GET' => [
|
||||
GET::class,
|
||||
['arg1'],
|
||||
],
|
||||
'MGET' => [
|
||||
MGET::class,
|
||||
['arg1', 'arg2', 'arg3'],
|
||||
],
|
||||
'ZADD' => [
|
||||
ZADD::class,
|
||||
['key', 'value', 'key1', 'value1'],
|
||||
],
|
||||
'JSONSET' => [
|
||||
JSONSET::class,
|
||||
['key', '$', '{"key":"value"}'],
|
||||
],
|
||||
'BFADD' => [
|
||||
BFADD::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'CFADD' => [
|
||||
CFADD::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'CMSINFO' => [
|
||||
CMSINFO::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'FTSEARCH' => [
|
||||
FTSEARCH::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'TDIGESTADD' => [
|
||||
TDIGESTADD::class,
|
||||
['key', 'value'],
|
||||
],
|
||||
'TSGET' => [
|
||||
TSGET::class,
|
||||
['key'],
|
||||
],
|
||||
'TOPKQUERY' => [
|
||||
TOPKQUERY::class,
|
||||
['key'],
|
||||
],
|
||||
];
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
namespace Predis\Connection\Cluster;
|
||||
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\Command\Redis\GET;
|
||||
use Predis\Connection\Parameters;
|
||||
use PredisTestCase;
|
||||
|
||||
@@ -471,4 +472,48 @@ class PredisClusterTest extends PredisTestCase
|
||||
|
||||
$this->assertEquals(['response1', 'response2', 'response3'], $cluster->executeCommandOnEachNode($mockCommand));
|
||||
}
|
||||
|
||||
/**
|
||||
* @group disconnected
|
||||
*/
|
||||
public function testWrite(): void
|
||||
{
|
||||
$command1 = new GET();
|
||||
$command1->setArguments(['arg1']);
|
||||
|
||||
$command2 = new GET();
|
||||
$command2->setArguments(['arg2']);
|
||||
|
||||
$command3 = new GET();
|
||||
$command3->setArguments(['arg3']);
|
||||
|
||||
$connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
|
||||
$connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
|
||||
$connection3 = $this->getMockConnection('tcp://127.0.0.1:7003');
|
||||
|
||||
$connection1
|
||||
->expects($this->exactly(3))
|
||||
->method('write')
|
||||
->withConsecutive(
|
||||
[$command1->serializeCommand()],
|
||||
[$command2->serializeCommand()],
|
||||
[$command3->serializeCommand()]
|
||||
);
|
||||
|
||||
$connection2
|
||||
->expects($this->never())
|
||||
->method('write');
|
||||
|
||||
$connection3
|
||||
->expects($this->never())
|
||||
->method('write');
|
||||
|
||||
$cluster = new PredisCluster(new Parameters());
|
||||
|
||||
$cluster->add($connection1);
|
||||
$cluster->add($connection2);
|
||||
$cluster->add($connection3);
|
||||
|
||||
$cluster->write($command1->serializeCommand() . $command2->serializeCommand() . $command3->serializeCommand());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -331,7 +331,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
);
|
||||
|
||||
// TODO: I'm not sure about mocking a protected method, but it'll do for now
|
||||
/** @var Connection\Cluster\RedisCluster|MockObject */
|
||||
/** @var RedisCluster|MockObject */
|
||||
$cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
|
||||
->onlyMethods(['getRandomConnection'])
|
||||
->setConstructorArgs([$factory, new Parameters()])
|
||||
@@ -633,7 +633,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
'value:5001'
|
||||
);
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->once())
|
||||
@@ -711,7 +711,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->willReturn($connection4);
|
||||
|
||||
// TODO: I'm not sure about mocking a protected method, but it'll do for now
|
||||
/** @var Connection\Cluster\RedisCluster|MockObject */
|
||||
/** @var RedisCluster|MockObject */
|
||||
$cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
|
||||
->onlyMethods(['getRandomConnection'])
|
||||
->setConstructorArgs([$factory, new Parameters()])
|
||||
@@ -740,7 +740,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
$this->expectException('Predis\ClientException');
|
||||
$this->expectExceptionMessage('No connections available in the pool');
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->never())
|
||||
@@ -758,7 +758,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
*/
|
||||
public function testAskSlotMapReturnEmptyArrayOnEmptyConnectionsPool(): void
|
||||
{
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->never())
|
||||
@@ -818,7 +818,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->method('create');
|
||||
|
||||
// TODO: I'm not sure about mocking a protected method, but it'll do for now
|
||||
/** @var Connection\Cluster\RedisCluster|MockObject */
|
||||
/** @var RedisCluster|MockObject */
|
||||
$cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
|
||||
->onlyMethods(['getRandomConnection'])
|
||||
->setConstructorArgs([$factory, new Parameters()])
|
||||
@@ -884,7 +884,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->method('create');
|
||||
|
||||
// TODO: I'm not sure about mocking a protected method, but it'll do for now
|
||||
/** @var Connection\Cluster\RedisCluster|MockObject */
|
||||
/** @var RedisCluster|MockObject */
|
||||
$cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
|
||||
->onlyMethods(['getRandomConnection'])
|
||||
->setConstructorArgs([$factory, new Parameters()])
|
||||
@@ -958,7 +958,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
'foobar'
|
||||
);
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->never())
|
||||
@@ -1009,7 +1009,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
'foobar'
|
||||
);
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->once())
|
||||
@@ -1054,7 +1054,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->with($command)
|
||||
->willReturnOnConsecutiveCalls('foobar', 'foobar');
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory->expects($this->never())->method('create');
|
||||
|
||||
@@ -1097,7 +1097,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->with($command)
|
||||
->willReturnOnConsecutiveCalls('foobar', 'foobar');
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->once())
|
||||
@@ -1142,7 +1142,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->with($command)
|
||||
->willReturn('foobar');
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->once())
|
||||
@@ -1238,7 +1238,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
'foobar'
|
||||
);
|
||||
|
||||
/** @var Connection\FactoryInterface|MockObject */
|
||||
/** @var FactoryInterface|MockObject */
|
||||
$factory = $this->getMockBuilder('Predis\Connection\FactoryInterface')->getMock();
|
||||
$factory
|
||||
->expects($this->once())
|
||||
@@ -1407,7 +1407,7 @@ class RedisClusterTest extends PredisTestCase
|
||||
->method('create');
|
||||
|
||||
// TODO: I'm not sure about mocking a protected method, but it'll do for now
|
||||
/** @var Connection\Cluster\RedisCluster|MockObject */
|
||||
/** @var RedisCluster|MockObject */
|
||||
$cluster = $this->getMockBuilder('Predis\Connection\Cluster\RedisCluster')
|
||||
->onlyMethods(['getRandomConnection'])
|
||||
->setConstructorArgs([$factory, new Parameters()])
|
||||
@@ -1518,4 +1518,48 @@ class RedisClusterTest extends PredisTestCase
|
||||
|
||||
$this->assertInstanceOf(Connection\RelayConnection::class, $cluster->getConnectionBySlot(9999));
|
||||
}
|
||||
|
||||
/**
|
||||
* @group disconnected
|
||||
*/
|
||||
public function testWrite(): void
|
||||
{
|
||||
$command1 = new Command\Redis\GET();
|
||||
$command1->setArguments(['arg1']);
|
||||
|
||||
$command2 = new Command\Redis\GET();
|
||||
$command2->setArguments(['arg2']);
|
||||
|
||||
$command3 = new Command\Redis\GET();
|
||||
$command3->setArguments(['arg3']);
|
||||
|
||||
$factory = $this->getMockBuilder(FactoryInterface::class)->getMock();
|
||||
|
||||
$connection1 = $this->getMockConnection('tcp://127.0.0.1:7001');
|
||||
$connection2 = $this->getMockConnection('tcp://127.0.0.1:7002');
|
||||
$connection3 = $this->getMockConnection('tcp://127.0.0.1:7003');
|
||||
|
||||
$connection1
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($command3->serializeCommand());
|
||||
|
||||
$connection2
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($command2->serializeCommand());
|
||||
|
||||
$connection3
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($command1->serializeCommand());
|
||||
|
||||
$cluster = new RedisCluster($factory, new Parameters());
|
||||
|
||||
$cluster->add($connection1);
|
||||
$cluster->add($connection2);
|
||||
$cluster->add($connection3);
|
||||
|
||||
$cluster->write($command1->serializeCommand() . $command2->serializeCommand() . $command3->serializeCommand());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1451,6 +1451,44 @@ repl_backlog_histlen:12978
|
||||
$this->assertSame($connection->getParameters(), $replication->getParameters());
|
||||
}
|
||||
|
||||
/**
|
||||
* @group disconnected
|
||||
*/
|
||||
public function testWrite(): void
|
||||
{
|
||||
$command1 = new Command\Redis\Json\JSONGET();
|
||||
$command1->setArguments(['arg1']);
|
||||
|
||||
$command2 = new Command\Redis\Json\JSONGET();
|
||||
$command2->setArguments(['arg2']);
|
||||
|
||||
$command3 = new Command\Redis\Json\JSONGET();
|
||||
$command3->setArguments(['arg3']);
|
||||
|
||||
$master = $this->getMockConnection('tcp://127.0.0.1:6379?role=master');
|
||||
$slave1 = $this->getMockConnection('tcp://127.0.0.1:6380?role=slave');
|
||||
|
||||
$slave1
|
||||
->expects($this->never())
|
||||
->method('write');
|
||||
|
||||
$master
|
||||
->expects($this->exactly(3))
|
||||
->method('write')
|
||||
->withConsecutive(
|
||||
[$command1->serializeCommand()],
|
||||
[$command2->serializeCommand()],
|
||||
[$command3->serializeCommand()]
|
||||
);
|
||||
|
||||
$replication = new MasterSlaveReplication();
|
||||
|
||||
$replication->add($master);
|
||||
$replication->add($slave1);
|
||||
|
||||
$replication->write($command1->serializeCommand() . $command2->serializeCommand() . $command3->serializeCommand());
|
||||
}
|
||||
|
||||
public function connectionsProvider(): array
|
||||
{
|
||||
return [
|
||||
|
||||
@@ -1545,6 +1545,52 @@ class SentinelReplicationTest extends PredisTestCase
|
||||
$this->assertSame($sentinel->getParameters(), $replication->getParameters());
|
||||
}
|
||||
|
||||
/**
|
||||
* @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());
|
||||
}
|
||||
|
||||
public function connectionsProvider(): array
|
||||
{
|
||||
return [
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use Predis\Client;
|
||||
use Predis\Command\Redis\PING;
|
||||
use Predis\Connection\Parameters;
|
||||
use Predis\Response;
|
||||
use PredisTestCase;
|
||||
@@ -26,6 +27,7 @@ class AtomicTest extends PredisTestCase
|
||||
{
|
||||
$pong = new Response\Status('PONG');
|
||||
$queued = new Response\Status('QUEUED');
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
|
||||
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
||||
$connection
|
||||
@@ -40,13 +42,9 @@ class AtomicTest extends PredisTestCase
|
||||
[$pong, $pong, $pong]
|
||||
);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest')
|
||||
->withConsecutive(
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')]
|
||||
);
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('readResponse')
|
||||
@@ -75,6 +73,7 @@ class AtomicTest extends PredisTestCase
|
||||
*/
|
||||
public function testThrowsExceptionOnAbortedTransaction(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$this->expectException('Predis\ClientException');
|
||||
$this->expectExceptionMessage('The underlying transaction has been aborted by the server');
|
||||
|
||||
@@ -93,13 +92,9 @@ class AtomicTest extends PredisTestCase
|
||||
null
|
||||
);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest')
|
||||
->withConsecutive(
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')]
|
||||
);
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('readResponse')
|
||||
@@ -123,6 +118,7 @@ class AtomicTest extends PredisTestCase
|
||||
*/
|
||||
public function testPipelineWithErrorInTransaction(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$this->expectException('Predis\Response\ServerException');
|
||||
$this->expectExceptionMessage('ERR Test error');
|
||||
|
||||
@@ -142,13 +138,9 @@ class AtomicTest extends PredisTestCase
|
||||
new Response\Status('OK')
|
||||
);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest')
|
||||
->withConsecutive(
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')]
|
||||
);
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('readResponse')
|
||||
@@ -172,6 +164,7 @@ class AtomicTest extends PredisTestCase
|
||||
*/
|
||||
public function testThrowsServerExceptionOnResponseErrorByDefault(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$this->expectException('Predis\Response\ServerException');
|
||||
$this->expectExceptionMessage('ERR Test error');
|
||||
|
||||
@@ -188,12 +181,9 @@ class AtomicTest extends PredisTestCase
|
||||
new Response\Status('OK')
|
||||
);
|
||||
$connection
|
||||
->expects($this->exactly(2))
|
||||
->method('writeRequest')
|
||||
->withConsecutive(
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')]
|
||||
);
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->once())
|
||||
->method('readResponse')
|
||||
@@ -214,6 +204,7 @@ class AtomicTest extends PredisTestCase
|
||||
*/
|
||||
public function testReturnsResponseErrorWithClientExceptionsSetToFalse(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$pong = new Response\Status('PONG');
|
||||
$queued = new Response\Status('QUEUED');
|
||||
$error = new Response\Error('ERR Test error');
|
||||
@@ -231,13 +222,9 @@ class AtomicTest extends PredisTestCase
|
||||
[$pong, $pong, $error]
|
||||
);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest')
|
||||
->withConsecutive(
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')],
|
||||
[$this->isRedisCommand('PING')]
|
||||
);
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('readResponse')
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use Predis\Client;
|
||||
use Predis\Command\Redis\PING;
|
||||
use PredisTestCase;
|
||||
|
||||
class FireAndForgetTest extends PredisTestCase
|
||||
@@ -22,10 +23,12 @@ class FireAndForgetTest extends PredisTestCase
|
||||
*/
|
||||
public function testPipelineWithSingleConnection(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$connection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest');
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->never())
|
||||
->method('readResponse');
|
||||
@@ -44,14 +47,16 @@ class FireAndForgetTest extends PredisTestCase
|
||||
*/
|
||||
public function testSwitchesToMasterWithReplicationConnection(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$connection = $this->getMockBuilder('Predis\Connection\Replication\ReplicationInterface')
|
||||
->getMock();
|
||||
$connection
|
||||
->expects($this->once())
|
||||
->method('switchToMaster');
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest');
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->never())
|
||||
->method('readResponse');
|
||||
@@ -64,4 +69,23 @@ class FireAndForgetTest extends PredisTestCase
|
||||
|
||||
$this->assertEmpty($pipeline->execute());
|
||||
}
|
||||
|
||||
/**
|
||||
* @group connected
|
||||
* @group cluster
|
||||
* @requiresRedisVersion >= 6.2.0
|
||||
*/
|
||||
public function testClusterExecutePipeline(): void
|
||||
{
|
||||
$pipeline = new FireAndForget($this->createClient());
|
||||
|
||||
$pipeline->set('foo', 'bar');
|
||||
$pipeline->get('foo');
|
||||
$pipeline->set('bar', 'foo');
|
||||
$pipeline->get('bar');
|
||||
$pipeline->set('baz', 'baz');
|
||||
$pipeline->get('baz');
|
||||
|
||||
$this->assertEmpty($pipeline->execute());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,8 @@ 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\Parameters;
|
||||
use Predis\Response;
|
||||
use PredisTestCase;
|
||||
@@ -215,10 +217,21 @@ class PipelineTest extends PredisTestCase
|
||||
*/
|
||||
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->exactly(3))
|
||||
->method('writeRequest');
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('readResponse')
|
||||
@@ -261,10 +274,27 @@ class PipelineTest extends PredisTestCase
|
||||
*/
|
||||
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(4))
|
||||
->method('writeRequest');
|
||||
->expects($this->exactly(2))
|
||||
->method('write')
|
||||
->withConsecutive([$buffer1], [$buffer2]);
|
||||
$connection
|
||||
->expects($this->exactly(4))
|
||||
->method('readResponse')
|
||||
@@ -291,6 +321,7 @@ class PipelineTest extends PredisTestCase
|
||||
*/
|
||||
public function testSwitchesToMasterWithReplicationConnection(): void
|
||||
{
|
||||
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
|
||||
$pong = new Response\Status('PONG');
|
||||
|
||||
$connection = $this->getMockBuilder('Predis\Connection\Replication\ReplicationInterface')->getMock();
|
||||
@@ -298,8 +329,9 @@ class PipelineTest extends PredisTestCase
|
||||
->expects($this->once())
|
||||
->method('switchToMaster');
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('writeRequest');
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(3))
|
||||
->method('readResponse')
|
||||
@@ -366,10 +398,28 @@ class PipelineTest extends PredisTestCase
|
||||
*/
|
||||
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->exactly(4))
|
||||
->method('writeRequest');
|
||||
->expects($this->once())
|
||||
->method('write')
|
||||
->with($buffer);
|
||||
$connection
|
||||
->expects($this->exactly(4))
|
||||
->method('readResponse')
|
||||
@@ -546,6 +596,36 @@ class PipelineTest extends PredisTestCase
|
||||
$this->assertSame('bar', $results[2]);
|
||||
}
|
||||
|
||||
/**
|
||||
* @group connected
|
||||
* @group cluster
|
||||
* @requiresRedisVersion >= 6.2.0
|
||||
*/
|
||||
public function testClusterExecutePipeline(): void
|
||||
{
|
||||
$client = $this->getClient();
|
||||
|
||||
$results = $client->pipeline(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);
|
||||
}
|
||||
|
||||
// ******************************************************************** //
|
||||
// ---- HELPER METHODS ------------------------------------------------ //
|
||||
// ******************************************************************** //
|
||||
|
||||
Reference in New Issue
Block a user