mirror of
https://github.com/predis/predis.git
synced 2026-08-17 16:53:41 +00:00
Rework pipeline abstraction on top of command queues.
There are almost no changes from a consumer point of view when pipelines are initialized using Predis\Client::pipeline(): their default behaviour is the same of Predis v1.1 and the same pipeline options are supported. This is just on the surface though because now we have just one pipeline class that internally relies on different command queues to encapsulate different behaviours (basic pipelining, atomic, fire-and-forget). Thanks to the fact that command queues always return a Traversable for responses, now a pipeline can be configured to "stream" replies to the caller instead of just returning an array. Please note that actually not all command queues support streaming in a real sense, it mostly depends on their underlying implementation. Anyway, traversable responses should be handled with care in order to avoid any desynchronization issue. Also note that partial flushes are unsupported when a pipeline is configured to return a traversable response. Pipelines can be configured to return traversable responses from Predis\Client::pipeline() simply by setting the new "traversable" option to TRUE in the pipeline options dictionary (default is FALSE). Atomic pipelines rely on the atomic command queue, a wrapper for other command queues, meaning that fire-and-forget pipelines can now be sent in a MULTI / EXEC context simply by setting both "fire-and-forget" and "atomic" to true in the pipeline options array. Furthermore, now pipelines can even support iterable multibulk responses when using Predis\Connection\CompositeStreamConnection. This is kind of an obscure feature but when combined with pipelines configured to return traversable responses it allows to effectively stream *every* element of the response at the protocol level. Atomic pipelines are supported, too! And when the pipeline is configured to return plain array responses, any iterable multibulk response is consumed automatically by the pipeline and returned as a plain array element. This feature should be used with even more care than normal traversable responses returned by pipelines. NOTE: we had to apply minor changes to method signatures in the shared Predis\ClientContextInterface interface so with this commit we adapted Predis\Transaction\MultiExec to these changes.
This commit is contained in:
Daniele Alessandri
+25
-20
@@ -21,6 +21,7 @@ use Predis\Connection\ParametersInterface;
|
||||
use Predis\Connection\Replication\SentinelReplication;
|
||||
use Predis\Monitor\Consumer as MonitorConsumer;
|
||||
use Predis\Pipeline\Pipeline;
|
||||
use Predis\Pipeline\Queue;
|
||||
use Predis\PubSub\Consumer as PubSubConsumer;
|
||||
use Predis\Response\ErrorInterface as ErrorResponseInterface;
|
||||
use Predis\Response\ResponseInterface;
|
||||
@@ -423,12 +424,16 @@ class Client implements ClientInterface, \IteratorAggregate
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new pipeline context and returns it, or returns the results of
|
||||
* a pipeline executed inside the optionally provided callable object.
|
||||
* Creates a new pipeline context.
|
||||
*
|
||||
* @param mixed ... Array of options, a callable for execution, or both.
|
||||
* This method returns the pipeline object unless a callable is provided by
|
||||
* the user. In this case the user-supplied callable receives the pipeline
|
||||
* instance as the only argument and the pipeline will be executed by the
|
||||
* client at the end, returning the responses read back from the server.
|
||||
*
|
||||
* @return Pipeline|array
|
||||
* @param mixed ... Array of options, a callable for execution, or both
|
||||
*
|
||||
* @return Pipeline|Traversable|array
|
||||
*/
|
||||
public function pipeline(/* arguments */)
|
||||
{
|
||||
@@ -436,27 +441,27 @@ class Client implements ClientInterface, \IteratorAggregate
|
||||
}
|
||||
|
||||
/**
|
||||
* Actual pipeline context initializer method.
|
||||
* Pipeline context initializer method.
|
||||
*
|
||||
* @param array $options Options for the context.
|
||||
* @param mixed $callable Optional callable used to execute the context.
|
||||
* @param array $options Optional configuration for the context
|
||||
* @param callable $callable Optional callable used to execute the context
|
||||
*
|
||||
* @return Pipeline|array
|
||||
* @return Pipeline|Traversable|array
|
||||
*/
|
||||
protected function createPipeline(array $options = null, $callable = null)
|
||||
protected function createPipeline(array $options = null, callable $callable = null)
|
||||
{
|
||||
if (isset($options['atomic']) && $options['atomic']) {
|
||||
$class = 'Predis\Pipeline\Atomic';
|
||||
} elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) {
|
||||
$class = 'Predis\Pipeline\FireAndForget';
|
||||
} else {
|
||||
$class = 'Predis\Pipeline\Pipeline';
|
||||
$traversable = (bool) ($options['traversable'] ?? false);
|
||||
$isFireAndForget = (bool) ($options['fire-and-forget'] ?? false);
|
||||
$isAtomic = (bool) ($options['atomic'] ?? false);
|
||||
$throwExceptions = (bool) ($options['exceptions'] ?? $this->getOptions()->exceptions);
|
||||
|
||||
$queue = $isFireAndForget ? new Queue\FireAndForget() : new Queue\Basic();
|
||||
if ($isAtomic) {
|
||||
$queue = new Queue\Atomic($queue);
|
||||
}
|
||||
|
||||
/*
|
||||
* @var ClientContextInterface
|
||||
*/
|
||||
$pipeline = new $class($this);
|
||||
$pipeline = new Pipeline($this, $queue, $traversable);
|
||||
$pipeline->setThrowOnErrorResponse($throwExceptions);
|
||||
|
||||
if (isset($callable)) {
|
||||
return $pipeline->execute($callable);
|
||||
@@ -471,7 +476,7 @@ class Client implements ClientInterface, \IteratorAggregate
|
||||
*
|
||||
* @param mixed ... Array of options, a callable for execution, or both.
|
||||
*
|
||||
* @return MultiExecTransaction|array
|
||||
* @return MultiExecTransaction|iterable
|
||||
*/
|
||||
public function transaction(/* arguments */)
|
||||
{
|
||||
|
||||
@@ -172,7 +172,7 @@ interface ClientContextInterface
|
||||
/**
|
||||
* Sends the specified command instance to Redis.
|
||||
*
|
||||
* @param CommandInterface $command Command instance.
|
||||
* @param CommandInterface $command Command instance
|
||||
*
|
||||
* @return mixed
|
||||
*/
|
||||
@@ -186,14 +186,14 @@ interface ClientContextInterface
|
||||
*
|
||||
* @return mixed
|
||||
*/
|
||||
public function __call($method, $arguments);
|
||||
public function __call(string $method, array $arguments);
|
||||
|
||||
/**
|
||||
* Starts the execution of the context.
|
||||
*
|
||||
* @param mixed $callable Optional callback for execution.
|
||||
* @param callable $callable Optional callable for execution
|
||||
*
|
||||
* @return array
|
||||
* @return ?iterable
|
||||
*/
|
||||
public function execute($callable = null);
|
||||
public function execute(callable $callable = null): ?iterable;
|
||||
}
|
||||
|
||||
@@ -1,119 +0,0 @@
|
||||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the Predis package.
|
||||
*
|
||||
* (c) Daniele Alessandri <suppakilla@gmail.com>
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use Predis\ClientException;
|
||||
use Predis\ClientInterface;
|
||||
use Predis\Connection\ConnectionInterface;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
use Predis\Response\ErrorInterface as ErrorResponseInterface;
|
||||
use Predis\Response\ResponseInterface;
|
||||
use Predis\Response\ServerException;
|
||||
|
||||
/**
|
||||
* Command pipeline wrapped into a MULTI / EXEC transaction.
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class Atomic extends Pipeline
|
||||
{
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function __construct(ClientInterface $client)
|
||||
{
|
||||
if (!$client->getCommandFactory()->supports('multi', 'exec', 'discard')) {
|
||||
throw new ClientException(
|
||||
"'MULTI', 'EXEC' and 'DISCARD' are not supported by the current command factory."
|
||||
);
|
||||
}
|
||||
|
||||
parent::__construct($client);
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function getConnection()
|
||||
{
|
||||
$connection = $this->getClient()->getConnection();
|
||||
|
||||
if (!$connection instanceof NodeConnectionInterface) {
|
||||
$class = __CLASS__;
|
||||
|
||||
throw new ClientException("The class '$class' does not support aggregate connections.");
|
||||
}
|
||||
|
||||
return $connection;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
|
||||
{
|
||||
$commandFactory = $this->getClient()->getCommandFactory();
|
||||
$connection->executeCommand($commandFactory->create('multi'));
|
||||
|
||||
foreach ($commands as $command) {
|
||||
$connection->writeRequest($command);
|
||||
}
|
||||
|
||||
foreach ($commands as $command) {
|
||||
$response = $connection->readResponse($command);
|
||||
|
||||
if ($response instanceof ErrorResponseInterface) {
|
||||
$connection->executeCommand($commandFactory->create('discard'));
|
||||
throw new ServerException($response->getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
$executed = $connection->executeCommand($commandFactory->create('exec'));
|
||||
|
||||
if (!isset($executed)) {
|
||||
// TODO: should be throwing a more appropriate exception.
|
||||
throw new ClientException(
|
||||
'The underlying transaction has been aborted by the server.'
|
||||
);
|
||||
}
|
||||
|
||||
if (count($executed) !== count($commands)) {
|
||||
$expected = count($commands);
|
||||
$received = count($executed);
|
||||
|
||||
throw new ClientException(
|
||||
"Invalid number of responses [expected $expected, received $received]."
|
||||
);
|
||||
}
|
||||
|
||||
$responses = array();
|
||||
$sizeOfPipe = count($commands);
|
||||
$exceptions = $this->throwServerExceptions();
|
||||
|
||||
for ($i = 0; $i < $sizeOfPipe; ++$i) {
|
||||
$command = $commands->dequeue();
|
||||
$response = $executed[$i];
|
||||
|
||||
if (!$response instanceof ResponseInterface) {
|
||||
$responses[] = $command->parseResponse($response);
|
||||
} elseif ($response instanceof ErrorResponseInterface && $exceptions) {
|
||||
$this->exception($connection, $response);
|
||||
} else {
|
||||
$responses[] = $response;
|
||||
}
|
||||
|
||||
unset($executed[$i]);
|
||||
}
|
||||
|
||||
return $responses;
|
||||
}
|
||||
}
|
||||
@@ -1,130 +0,0 @@
|
||||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the Predis package.
|
||||
*
|
||||
* (c) Daniele Alessandri <suppakilla@gmail.com>
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use Predis\CommunicationException;
|
||||
use Predis\Connection\Cluster\ClusterInterface;
|
||||
use Predis\Connection\ConnectionInterface;
|
||||
use Predis\Connection\NodeConnectionInterface;
|
||||
use Predis\NotSupportedException;
|
||||
|
||||
/**
|
||||
* Command pipeline that does not throw exceptions on connection errors, but
|
||||
* returns the exception instances as the rest of the response elements.
|
||||
*
|
||||
* @todo Awful naming!
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class ConnectionErrorProof extends Pipeline
|
||||
{
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function getConnection()
|
||||
{
|
||||
return $this->getClient()->getConnection();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
|
||||
{
|
||||
if ($connection instanceof NodeConnectionInterface) {
|
||||
return $this->executeSingleNode($connection, $commands);
|
||||
} elseif ($connection instanceof ClusterInterface) {
|
||||
return $this->executeCluster($connection, $commands);
|
||||
} else {
|
||||
$class = get_class($connection);
|
||||
|
||||
throw new NotSupportedException("The connection class '$class' is not supported.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function executeSingleNode(NodeConnectionInterface $connection, \SplQueue $commands)
|
||||
{
|
||||
$responses = array();
|
||||
$sizeOfPipe = count($commands);
|
||||
|
||||
foreach ($commands as $command) {
|
||||
try {
|
||||
$connection->writeRequest($command);
|
||||
} catch (CommunicationException $exception) {
|
||||
return array_fill(0, $sizeOfPipe, $exception);
|
||||
}
|
||||
}
|
||||
|
||||
for ($i = 0; $i < $sizeOfPipe; ++$i) {
|
||||
$command = $commands->dequeue();
|
||||
|
||||
try {
|
||||
$responses[$i] = $connection->readResponse($command);
|
||||
} catch (CommunicationException $exception) {
|
||||
$add = count($commands) - count($responses);
|
||||
$responses = array_merge($responses, array_fill(0, $add, $exception));
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return $responses;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function executeCluster(ClusterInterface $connection, \SplQueue $commands)
|
||||
{
|
||||
$responses = array();
|
||||
$sizeOfPipe = count($commands);
|
||||
$exceptions = array();
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
for ($i = 0; $i < $sizeOfPipe; ++$i) {
|
||||
$command = $commands->dequeue();
|
||||
|
||||
$cmdConnection = $connection->getConnectionByCommand($command);
|
||||
$connectionHash = spl_object_hash($cmdConnection);
|
||||
|
||||
if (isset($exceptions[$connectionHash])) {
|
||||
$responses[$i] = $exceptions[$connectionHash];
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
$responses[$i] = $cmdConnection->readResponse($command);
|
||||
} catch (CommunicationException $exception) {
|
||||
$responses[$i] = $exception;
|
||||
$exceptions[$connectionHash] = $exception;
|
||||
}
|
||||
}
|
||||
|
||||
return $responses;
|
||||
}
|
||||
}
|
||||
@@ -1,36 +0,0 @@
|
||||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the Predis package.
|
||||
*
|
||||
* (c) Daniele Alessandri <suppakilla@gmail.com>
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use Predis\Connection\ConnectionInterface;
|
||||
|
||||
/**
|
||||
* Command pipeline that writes commands to the servers but discards responses.
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class FireAndForget extends Pipeline
|
||||
{
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
|
||||
{
|
||||
while (!$commands->isEmpty()) {
|
||||
$connection->writeRequest($commands->dequeue());
|
||||
}
|
||||
|
||||
$connection->disconnect();
|
||||
|
||||
return array();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the Predis package.
|
||||
*
|
||||
* (c) Daniele Alessandri <suppakilla@gmail.com>
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Predis\Pipeline\Iterator;
|
||||
|
||||
use NoRewindIterator;
|
||||
|
||||
/**
|
||||
* Non-rewindable iterator for pipeline results.
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class Responses extends NoRewindIterator
|
||||
{
|
||||
/**
|
||||
* Returns pipeline results as an array.
|
||||
*
|
||||
* Since pipeline results are non-rewindable, invoking this method after the
|
||||
* start of an iteration will just return the reminder of the results set.
|
||||
*
|
||||
* @return array
|
||||
*/
|
||||
public function all(): array
|
||||
{
|
||||
return iterator_to_array($this, false);
|
||||
}
|
||||
}
|
||||
+267
-109
@@ -11,95 +11,123 @@
|
||||
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use AppendIterator;
|
||||
use ArrayIterator;
|
||||
use Countable;
|
||||
use Exception;
|
||||
use IteratorAggregate;
|
||||
use Predis\ClientContextInterface;
|
||||
use Predis\ClientException;
|
||||
use Predis\ClientInterface;
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\Connection\ConnectionInterface;
|
||||
use Predis\Connection\Replication\ReplicationInterface;
|
||||
use Predis\Pipeline\Queue\CommandQueueException;
|
||||
use Predis\Pipeline\Queue\CommandQueueInterface;
|
||||
use Predis\Response\ErrorInterface as ErrorResponseInterface;
|
||||
use Predis\Response\Iterator\MultiBulkIterator;
|
||||
use Predis\Response\ResponseInterface;
|
||||
use Predis\Response\ServerException;
|
||||
use Throwable;
|
||||
use Traversable;
|
||||
|
||||
/**
|
||||
* Implementation of a command pipeline in which write and read operations of
|
||||
* Redis commands are pipelined to alleviate the effects of network round-trips.
|
||||
* Abstraction for pipelining commands to Redis.
|
||||
*
|
||||
* Pipelines can use different underlying command queue implementations in order
|
||||
* to change the behaviour of how commands are flushed over the connection, for
|
||||
* example by using a fire-and-forget approach or by wrapping the whole pipeline
|
||||
* in a MULTI / EXEC transaction block.
|
||||
*
|
||||
* {@inheritdoc}
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class Pipeline implements ClientContextInterface
|
||||
class Pipeline implements ClientContextInterface, Countable, IteratorAggregate
|
||||
{
|
||||
private $executing = false;
|
||||
private $pending = false;
|
||||
private $traversable = false;
|
||||
private $throwOnErrorResponse = false;
|
||||
/** @var ClientInterface */
|
||||
private $client;
|
||||
private $pipeline;
|
||||
|
||||
private $responses = array();
|
||||
private $running = false;
|
||||
/** @var CommandQueueInterface */
|
||||
private $queue;
|
||||
/** @var NoRewindIterator|AppendIterator */
|
||||
private $responses;
|
||||
|
||||
/**
|
||||
* @param ClientInterface $client Client instance used by the context.
|
||||
* @param ClientInterface $client Client instance used by the pipeline
|
||||
* @param CommandQueueInterface $queue Optional command queue implementation for pipeline execution
|
||||
* @param bool $travesable Flag to return responses as a traversable iterator or an array
|
||||
*/
|
||||
public function __construct(ClientInterface $client)
|
||||
public function __construct(ClientInterface $client, CommandQueueInterface $queue = null, bool $traversable = true)
|
||||
{
|
||||
$this->client = $client;
|
||||
$this->pipeline = new \SplQueue();
|
||||
$this->queue = $queue ?? new Queue\Basic();
|
||||
$this->traversable = $traversable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Queues a command into the pipeline buffer.
|
||||
*
|
||||
* @param string $method Command ID.
|
||||
* @param array $arguments Arguments for the command.
|
||||
*
|
||||
* @return $this
|
||||
* @inheritdoc
|
||||
*/
|
||||
public function __call($method, $arguments)
|
||||
public function __destruct()
|
||||
{
|
||||
$command = $this->client->createCommand($method, $arguments);
|
||||
$this->recordCommand($command);
|
||||
// NOTE: it is safer to force-close the underlying connection on pending
|
||||
// traversable responses to avoid protocol desynchronization issues when
|
||||
// the pipeline goes out of scope and the GC kicks in, especially after
|
||||
// a script terminates and connections are configured to be persistent.
|
||||
if ($this->pending && $this->traversable) {
|
||||
$this->client->disconnect();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Queues the command instance into the pipeline queue.
|
||||
*
|
||||
* @param CommandInterface $command Command to be queued into the pipeline
|
||||
*
|
||||
* @return object
|
||||
*/
|
||||
protected function recordCommand(CommandInterface $command): object
|
||||
{
|
||||
if ($this->executing) {
|
||||
$message = 'The pipeline context is still executing';
|
||||
if ($this->traversable) {
|
||||
$message .= ' (iteration over responses may not be concluded yet)';
|
||||
}
|
||||
|
||||
throw new PipelineException($this, $message);
|
||||
}
|
||||
|
||||
$this->queue->enqueue($command);
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Queues a command instance into the pipeline buffer.
|
||||
* Stores a command into the pipeline for transmission.
|
||||
*
|
||||
* @param CommandInterface $command Command to be queued in the buffer.
|
||||
* @param string $method Command ID
|
||||
* @param array $arguments Arguments for the command
|
||||
*
|
||||
* @return $this|mixed
|
||||
*/
|
||||
protected function recordCommand(CommandInterface $command)
|
||||
public function __call(string $method, array $arguments)
|
||||
{
|
||||
$this->pipeline->enqueue($command);
|
||||
$command = $this->client->createCommand($method, $arguments);
|
||||
return $this->recordCommand($command);
|
||||
}
|
||||
|
||||
/**
|
||||
* Queues a command instance into the pipeline buffer.
|
||||
* Stores a command instance into the pipeline for transmission.
|
||||
*
|
||||
* @param CommandInterface $command Command instance to be queued in the buffer.
|
||||
* @param CommandInterface $command Command to be queued into the pipeline
|
||||
*
|
||||
* @return $this
|
||||
* @return $this|mixed
|
||||
*/
|
||||
public function executeCommand(CommandInterface $command)
|
||||
{
|
||||
$this->recordCommand($command);
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Throws an exception on -ERR responses returned by Redis.
|
||||
*
|
||||
* @param ConnectionInterface $connection Redis connection that returned the error.
|
||||
* @param ErrorResponseInterface $response Instance of the error response.
|
||||
*
|
||||
* @throws ServerException
|
||||
*/
|
||||
protected function exception(ConnectionInterface $connection, ErrorResponseInterface $response)
|
||||
{
|
||||
$connection->disconnect();
|
||||
$message = $response->getMessage();
|
||||
|
||||
throw new ServerException($message);
|
||||
return $this->recordCommand($command);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -107,7 +135,7 @@ class Pipeline implements ClientContextInterface
|
||||
*
|
||||
* @return ConnectionInterface
|
||||
*/
|
||||
protected function getConnection()
|
||||
protected function getConnection(): ConnectionInterface
|
||||
{
|
||||
$connection = $this->getClient()->getConnection();
|
||||
|
||||
@@ -119,124 +147,254 @@ class Pipeline implements ClientContextInterface
|
||||
}
|
||||
|
||||
/**
|
||||
* Implements the logic to flush the queued commands and read the responses
|
||||
* from the current connection.
|
||||
* Flushes the pipeline over the target connection and returns responses.
|
||||
*
|
||||
* @param ConnectionInterface $connection Current connection instance.
|
||||
* @param \SplQueue $commands Queued commands.
|
||||
* @param ConnectionInterface $connection Target connection
|
||||
*
|
||||
* @return array
|
||||
* @return Traversable
|
||||
*/
|
||||
protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
|
||||
protected function executePipeline(ConnectionInterface $connection): Traversable
|
||||
{
|
||||
foreach ($commands as $command) {
|
||||
$connection->writeRequest($command);
|
||||
}
|
||||
$this->pending = true;
|
||||
|
||||
$responses = array();
|
||||
$exceptions = $this->throwServerExceptions();
|
||||
/** @var Traversable */
|
||||
$responses = null;
|
||||
/** @var CommandInterface */
|
||||
$command = null;
|
||||
|
||||
while (!$commands->isEmpty()) {
|
||||
$command = $commands->dequeue();
|
||||
$response = $connection->readResponse($command);
|
||||
try {
|
||||
$responses = $this->queue->flush($connection);
|
||||
|
||||
if (!$response instanceof ResponseInterface) {
|
||||
$responses[] = $command->parseResponse($response);
|
||||
} elseif ($response instanceof ErrorResponseInterface && $exceptions) {
|
||||
$this->exception($connection, $response);
|
||||
} else {
|
||||
$responses[] = $response;
|
||||
foreach ($responses as $command => $response) {
|
||||
if ($response instanceof ResponseInterface) {
|
||||
if ($response instanceof ErrorResponseInterface) {
|
||||
$response = $this->onResponseError($connection, $command, $response);
|
||||
} elseif ($response instanceof MultiBulkIterator) {
|
||||
$response = $this->onResponseTraversable($connection, $command, $response);
|
||||
}
|
||||
} else {
|
||||
$response = $command->parseResponse($response);
|
||||
}
|
||||
|
||||
yield $command => $response;
|
||||
}
|
||||
} catch (CommandQueueException $exception) {
|
||||
$this->onExceptionDuringExecution($exception, $connection, $responses, $command);
|
||||
|
||||
throw new PipelineException($this, $exception->getMessage(), $exception->getCode(), $exception);
|
||||
} catch (Exception $exception) {
|
||||
$this->onExceptionDuringExecution($exception, $connection, $responses, $command);
|
||||
|
||||
throw $exception;
|
||||
} finally {
|
||||
$this->executing = false;
|
||||
}
|
||||
|
||||
return $responses;
|
||||
// NOTE: this flag gets set only when the generator properly reaches the
|
||||
// end of the iteration, otherwise this is skipped. This could happen if
|
||||
// the script terminates before the end of an iteration or an unhandled
|
||||
// exception occurs. The flags is used in __destruct() to verify when we
|
||||
// should drop the underlying connection in order to avoid any protocol
|
||||
// desynchronization issue.
|
||||
$this->pending = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flushes the buffer holding all of the commands queued so far.
|
||||
* Performs clean-ups when an exception occurs during pipeline execution.
|
||||
*
|
||||
* @param bool $send Specifies if the commands in the buffer should be sent to Redis.
|
||||
* @param Throwable $exception Exception thrown during pipeline
|
||||
* @param ConnectionInterface $connection Redis connection that returned the error
|
||||
* @param ?Traversable $responses Current reponse set returned by Redis
|
||||
* @param ?CommandInterface $command Command affected by the error
|
||||
*/
|
||||
protected function onExceptionDuringExecution(
|
||||
Throwable $exception,
|
||||
ConnectionInterface $connection,
|
||||
?Traversable $responses = null,
|
||||
?CommandInterface $command = null): void
|
||||
{
|
||||
$connection->disconnect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles RESP error (prefix `-`) responses returned by Redis.
|
||||
*
|
||||
* @param ConnectionInterface $connection Redis connection that returned the error
|
||||
* @param CommandInterface $command Command affected by the error
|
||||
* @param ErrorResponseInterface $response Error response instance
|
||||
*
|
||||
* @return mixed
|
||||
*
|
||||
* @throws ServerException
|
||||
*/
|
||||
protected function onResponseError(ConnectionInterface $connection, CommandInterface $command, ErrorResponseInterface $response)
|
||||
{
|
||||
if ($this->throwOnErrorResponse) {
|
||||
throw new ServerException($response->getMessage());
|
||||
}
|
||||
|
||||
return $response;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles traversable RESP array (prefix `*`) responses returned by Redis.
|
||||
*
|
||||
* @param ConnectionInterface $connection Redis connection that returned the error
|
||||
* @param CommandInterface $command Command affected by the error
|
||||
* @param MultiBulkIterator $response Traversable response instance
|
||||
*
|
||||
* @return iterable
|
||||
*/
|
||||
protected function onResponseTraversable(ConnectionInterface $connection, CommandInterface $command, MultiBulkIterator $response): iterable
|
||||
{
|
||||
if (!$this->traversable) {
|
||||
$response = iterator_to_array($response, false);
|
||||
}
|
||||
|
||||
return $response;
|
||||
}
|
||||
|
||||
/**
|
||||
* Flushes the current pipeline queue.
|
||||
*
|
||||
* @return $this
|
||||
*/
|
||||
public function flushPipeline($send = true)
|
||||
public function flushQueued(): self
|
||||
{
|
||||
if ($send && !$this->pipeline->isEmpty()) {
|
||||
$responses = $this->executePipeline($this->getConnection(), $this->pipeline);
|
||||
$this->responses = array_merge($this->responses, $responses);
|
||||
} else {
|
||||
$this->pipeline = new \SplQueue();
|
||||
if ($this->traversable) {
|
||||
throw new PipelineException($this, sprintf(
|
||||
'%s does not support intermediate flushes when configured to return Traversable responses',
|
||||
static::class
|
||||
));
|
||||
}
|
||||
|
||||
$this->flushQueuedInternal();
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Marks the running status of the pipeline.
|
||||
* Flushes the current pipeline queue (internal method).
|
||||
*
|
||||
* @param bool $bool Sets the running status of the pipeline.
|
||||
*
|
||||
* @throws ClientException
|
||||
* @return void
|
||||
*/
|
||||
private function setRunning($bool)
|
||||
protected function flushQueuedInternal(): void
|
||||
{
|
||||
if ($bool && $this->running) {
|
||||
throw new ClientException('The current pipeline context is already being executed.');
|
||||
$this->executing = true;
|
||||
|
||||
$responses = $this->executePipeline($this->getConnection());
|
||||
|
||||
if ($this->traversable) {
|
||||
$this->responses = new Iterator\Responses($responses);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$this->running = $bool;
|
||||
if ($this->responses === null) {
|
||||
$this->responses = new AppendIterator();
|
||||
}
|
||||
|
||||
$this->responses->append(new ArrayIterator(
|
||||
iterator_to_array($responses, false)
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles the actual execution of the whole pipeline.
|
||||
* Drops the current pipeline queue.
|
||||
*
|
||||
* @param mixed $callable Optional callback for execution.
|
||||
*
|
||||
* @throws \Exception
|
||||
* @throws \InvalidArgumentException
|
||||
*
|
||||
* @return array
|
||||
* @return $this
|
||||
*/
|
||||
public function execute($callable = null)
|
||||
public function dropQueued(): self
|
||||
{
|
||||
if ($callable && !is_callable($callable)) {
|
||||
throw new \InvalidArgumentException('The argument must be a callable object.');
|
||||
}
|
||||
$this->queue->reset();
|
||||
|
||||
$exception = null;
|
||||
$this->setRunning(true);
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles the execution of a pipeline.
|
||||
*
|
||||
* Execution can be wrapped inside a callable provided by the user and that
|
||||
* receives an instance of the pipeline (self) as the only argument. Queued
|
||||
* commands will be automatically flushed when the callable returns.
|
||||
*
|
||||
* @param callable $callable Optional callable for execution
|
||||
*
|
||||
* @throws Exception
|
||||
*
|
||||
* @return ?iterable
|
||||
*/
|
||||
public function execute(callable $callable = null): ?iterable
|
||||
{
|
||||
$responses = null;
|
||||
|
||||
try {
|
||||
if ($callable) {
|
||||
call_user_func($callable, $this);
|
||||
}
|
||||
|
||||
$this->flushPipeline();
|
||||
} catch (\Exception $exception) {
|
||||
// NOOP
|
||||
$this->flushQueuedInternal();
|
||||
} finally {
|
||||
[$responses, $this->responses] = [$this->responses, null];
|
||||
}
|
||||
|
||||
$this->setRunning(false);
|
||||
|
||||
if ($exception) {
|
||||
throw $exception;
|
||||
if (!$this->traversable) {
|
||||
$responses = iterator_to_array($responses, false);
|
||||
}
|
||||
|
||||
return $this->responses;
|
||||
return $responses;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns if the pipeline should throw exceptions on server errors.
|
||||
* @inheritdoc
|
||||
*/
|
||||
public function count(): int
|
||||
{
|
||||
return count($this->queue);
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritdoc
|
||||
*/
|
||||
public function getIterator(): Traversable
|
||||
{
|
||||
return is_array($iterable = $this->execute())
|
||||
? new ArrayIterator($iterable)
|
||||
: $iterable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets if the pipeline is set to return responses as Travesable instances.
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
protected function throwServerExceptions()
|
||||
public function isTraversable(): bool
|
||||
{
|
||||
return (bool) $this->client->getOptions()->exceptions;
|
||||
return $this->traversable;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the underlying client instance used by the pipeline object.
|
||||
* Configures the pipeline to throw an exception on -ERR response.
|
||||
*
|
||||
* @param bool $value
|
||||
*/
|
||||
public function setThrowOnErrorResponse(bool $value): void
|
||||
{
|
||||
$this->throwOnErrorResponse = $value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the current configuration for exceptions on -ERR responses.
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
public function getThrowOnErrorResponse(): bool
|
||||
{
|
||||
return $this->throwOnErrorResponse;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the underlying client instance used by the pipeline.
|
||||
*
|
||||
* @return ClientInterface
|
||||
*/
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
<?php
|
||||
|
||||
/*
|
||||
* This file is part of the Predis package.
|
||||
*
|
||||
* (c) Daniele Alessandri <suppakilla@gmail.com>
|
||||
*
|
||||
* For the full copyright and license information, please view the LICENSE
|
||||
* file that was distributed with this source code.
|
||||
*/
|
||||
|
||||
namespace Predis\Pipeline;
|
||||
|
||||
use Predis\PredisException;
|
||||
use Throwable;
|
||||
|
||||
/**
|
||||
* Exception class for pipeline errors.
|
||||
*
|
||||
* @author Daniele Alessandri <suppakilla@gmail.com>
|
||||
*/
|
||||
class PipelineException extends PredisException
|
||||
{
|
||||
protected $pipeline;
|
||||
|
||||
/**
|
||||
* @param Pipeline $connection Pipeline associated to the exception
|
||||
* @param string $message Exception message
|
||||
* @param integer $code Exception code
|
||||
* @param Throwable $previous Previous exception for exception chaining
|
||||
*/
|
||||
public function __construct(Pipeline $pipeline, string $message = '', int $code = 0, Throwable $previous = null)
|
||||
{
|
||||
parent::__construct($message, $code, $previous);
|
||||
|
||||
$this->pipeline = $pipeline;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the pipeline associated to the exception.
|
||||
*
|
||||
* @return Pipeline
|
||||
*/
|
||||
public function getPipeline(): Pipeline
|
||||
{
|
||||
return $this->pipeline;
|
||||
}
|
||||
}
|
||||
@@ -155,7 +155,7 @@ class MultiExec implements ClientContextInterface
|
||||
*
|
||||
* @return mixed
|
||||
*/
|
||||
public function __call($method, $arguments)
|
||||
public function __call(string $method, array $arguments)
|
||||
{
|
||||
return $this->executeCommand(
|
||||
$this->client->createCommand($method, $arguments)
|
||||
@@ -348,15 +348,15 @@ class MultiExec implements ClientContextInterface
|
||||
/**
|
||||
* Handles the actual execution of the whole transaction.
|
||||
*
|
||||
* @param mixed $callable Optional callback for execution.
|
||||
* @param callable $callable Optional callable for execution
|
||||
*
|
||||
* @throws CommunicationException
|
||||
* @throws AbortedMultiExecException
|
||||
* @throws ServerException
|
||||
*
|
||||
* @return array
|
||||
* @return ?iterable
|
||||
*/
|
||||
public function execute($callable = null)
|
||||
public function execute(callable $callable = null): ?iterable
|
||||
{
|
||||
$this->checkBeforeExecution($callable);
|
||||
|
||||
@@ -373,7 +373,7 @@ class MultiExec implements ClientContextInterface
|
||||
$this->discard();
|
||||
}
|
||||
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
|
||||
$execResponse = $this->call('EXEC');
|
||||
|
||||
Reference in New Issue
Block a user