geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-09-12 11:27:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor pipeline data writing depends on connection type (#1586)

Browse Source 
* Refactor pipeline data writing depends on connection type

* Updated CHANGELOG.md
This commit is contained in:
Vladyslav Vildanov
2025-08-13 09:59:39 +03:00
committed by GitHub
parent 0f4bc7653c
commit 053cb4b6ac
8 changed files with 184 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+2
View File
@@ -4,6 +4,8 @@
### Added
- Added cluster support for `XADD`, `XDEL` and `XRANGE` (#1587)
### Changed
- Refactor pipeline data writing depends on connection type (#1586)
### Maintenance
## v3.2.0 (2025-08-05)
src/Pipeline/Atomic.php
+5 -5
View File
@@ -14,6 +14,7 @@ namespace Predis\Pipeline;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Connection\AggregateConnectionInterface;
use Predis\Connection\ConnectionInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
@@ -63,14 +64,13 @@ class Atomic extends Pipeline
{
$commandFactory = $this->getClient()->getCommandFactory();
$connection->executeCommand($commandFactory->create('multi'));
$buffer = '';
foreach ($commands as $command) {
$buffer .= $command->serializeCommand();
if ($connection instanceof AggregateConnectionInterface) {
$this->writeToMultiNode($connection, $commands);
} else {
$this->writeToSingleNode($connection, $commands);
}
$connection->write($buffer);
foreach ($commands as $command) {
$response = $connection->readResponse($command);
src/Pipeline/ConnectionErrorProof.php
+2 -4
View File
@@ -92,14 +92,12 @@ class ConnectionErrorProof extends Pipeline
$responses = [];
$sizeOfPipe = count($commands);
$exceptions = [];
$buffer = '';
foreach ($commands as $command) {
$buffer .= $command->serializeCommand();
$nodeConnection = $connection->getConnectionByCommand($command);
$nodeConnection->write($command->serializeCommand());
}
$connection->write($buffer);
for ($i = 0; $i < $sizeOfPipe; ++$i) {
$command = $commands->dequeue();
src/Pipeline/FireAndForget.php
+5 -5
View File
@@ -12,6 +12,7 @@
namespace Predis\Pipeline;
use Predis\Connection\AggregateConnectionInterface;
use Predis\Connection\ConnectionInterface;
use SplQueue;
@@ -25,13 +26,12 @@ class FireAndForget extends Pipeline
*/
protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
{
$buffer = '';
while (!$commands->isEmpty()) {
$buffer .= $commands->dequeue()->serializeCommand();
if ($connection instanceof AggregateConnectionInterface) {
$this->writeToMultiNode($connection, $commands);
} else {
$this->writeToSingleNode($connection, $commands);
}
$connection->write($buffer);
$connection->disconnect();
return [];
src/Pipeline/Pipeline.php
+38 -6
View File
@@ -18,6 +18,7 @@ use Predis\ClientContextInterface;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Command\CommandInterface;
use Predis\Connection\AggregateConnectionInterface;
use Predis\Connection\ConnectionInterface;
use Predis\Connection\Replication\ReplicationInterface;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
@@ -131,14 +132,12 @@ class Pipeline implements ClientContextInterface
*/
protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
{
$buffer = '';
foreach ($commands as $command) {
$buffer .= $command->serializeCommand();
if ($connection instanceof AggregateConnectionInterface) {
$this->writeToMultiNode($connection, $commands);
} else {
$this->writeToSingleNode($connection, $commands);
}
$connection->write($buffer);
$responses = [];
$exceptions = $this->throwServerExceptions();
$protocolVersion = (int) $connection->getParameters()->protocol;
@@ -163,6 +162,39 @@ class Pipeline implements ClientContextInterface
return $responses;
}
/**
* Writes pipelined commands to single node connection.
*
* @param ConnectionInterface $connection
* @param SplQueue $commands
* @return void
*/
protected function writeToSingleNode(ConnectionInterface $connection, SplQueue $commands)
{
$buffer = '';
foreach ($commands as $command) {
$buffer .= $command->serializeCommand();
}
$connection->write($buffer);
}
/**
* Writes pipelined commands to multi node connection.
*
* @param AggregateConnectionInterface $connection
* @param SplQueue $commands
* @return void
*/
protected function writeToMultiNode(AggregateConnectionInterface $connection, SplQueue $commands)
{
foreach ($commands as $command) {
$nodeConnection = $connection->getConnectionByCommand($command);
$nodeConnection->write($command->serializeCommand());
}
}
/**
* Flushes the buffer holding all of the commands queued so far.
*
tests/Predis/Pipeline/AtomicTest.php
+44
View File
@@ -13,6 +13,7 @@
namespace Predis\Pipeline;
use Predis\Client;
use Predis\ClientInterface;
use Predis\Command\Redis\PING;
use Predis\Connection\Parameters;
use Predis\Response;
@@ -263,4 +264,47 @@ class AtomicTest extends PredisTestCase
$pipeline->execute();
}
/**
* @group connected
* @group relay-incompatible
*/
public function testReplicationExecutesPipelineWithCRLFValues(): void
{
$parameters = $this->getDefaultParametersArray();
$client = $this->getClient(
["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
['replication' => 'predis']
);
$results = $client->pipeline(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);
}
}
tests/Predis/Pipeline/FireAndForgetTest.php
+54 -4
View File
@@ -13,7 +13,9 @@
namespace Predis\Pipeline;
use Predis\Client;
use Predis\ClientInterface;
use Predis\Command\Redis\PING;
use Predis\Response;
use PredisTestCase;
class FireAndForgetTest extends PredisTestCase
@@ -47,16 +49,21 @@ class FireAndForgetTest extends PredisTestCase
*/
public function testSwitchesToMasterWithReplicationConnection(): void
{
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
$nodeConnection = $this->getMockBuilder('Predis\Connection\NodeConnectionInterface')->getMock();
$nodeConnection
->expects($this->exactly(3))
->method('write')
->with((new PING())->serializeCommand());
$connection = $this->getMockBuilder('Predis\Connection\Replication\ReplicationInterface')
->getMock();
$connection
->expects($this->once())
->method('switchToMaster');
$connection
->expects($this->once())
->method('write')
->with($buffer);
->expects($this->exactly(3))
->method('getConnectionByCommand')
->willReturn($nodeConnection);
$connection
->expects($this->never())
->method('readResponse');
@@ -89,4 +96,47 @@ class FireAndForgetTest extends PredisTestCase
$this->assertEmpty($pipeline->execute());
}
/**
* @group connected
* @group relay-incompatible
*/
public function testReplicationExecutesPipelineWithCRLFValues(): void
{
$parameters = $this->getDefaultParametersArray();
$client = $this->getClient(
["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
['replication' => 'predis']
);
$results = $client->pipeline(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);
}
}
tests/Predis/Pipeline/PipelineTest.php
+34 -4
View File
@@ -321,7 +321,11 @@ class PipelineTest extends PredisTestCase
*/
public function testSwitchesToMasterWithReplicationConnection(): void
{
$buffer = (new PING())->serializeCommand() . (new PING())->serializeCommand() . (new PING())->serializeCommand();
$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();
@@ -329,9 +333,9 @@ class PipelineTest extends PredisTestCase
->expects($this->once())
->method('switchToMaster');
$connection
->expects($this->once())
->method('write')
->with($buffer);
->expects($this->exactly(3))
->method('getConnectionByCommand')
->willReturn($nodeConnection);
$connection
->expects($this->exactly(3))
->method('readResponse')
@@ -644,6 +648,32 @@ class PipelineTest extends PredisTestCase
$this->assertSameValues($expectedResults, $results);
}
/**
* @group connected
* @group relay-incompatible
*/
public function testReplicationExecutesPipelineWithCRLFValues(): void
{
$parameters = $this->getDefaultParametersArray();
$client = $this->getClient(
["tcp://{$parameters['host']}:{$parameters['port']}?role=master&database={$parameters['database']}&password={$parameters['password']}"],
['replication' => 'predis']
);
$results = $client->pipeline(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 ------------------------------------------------ //
// ******************************************************************** //