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.

src/Client.php

@@ -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 */)
     {

src/ClientContextInterface.php

@@ -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;
 }

src/Pipeline/Atomic.php

@@ -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;
-    }
-}

src/Pipeline/ConnectionErrorProof.php

@@ -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;
-    }
-}

src/Pipeline/FireAndForget.php

@@ -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();
-    }
-}

src/Pipeline/Iterator/Responses.php

@@ -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);
+    }
+}

src/Pipeline/Pipeline.php

@@ -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
      */

src/Pipeline/PipelineException.php

@@ -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;
+    }
+}

src/Pipeline/Queue/Atomic.php

@@ -0,0 +1,215 @@
+<?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\Queue;
+
+use SplQueue;
+use Throwable;
+use Traversable;
+use NoRewindIterator;
+use InvalidArgumentException;
+use Predis\Command\CommandInterface;
+use Predis\Command\RawCommand;
+use Predis\Connection\ConnectionInterface;
+use Predis\Response\ErrorInterface;
+
+/**
+ * Atomic command queue that wraps flushes into MULTI / EXEC transactions.
+ */
+class Atomic implements CommandQueueInterface
+{
+    private $discarded = false;
+    /** @var CommandQueueInterface */
+    private $innerQueue;
+    /** @var SplQueue */
+    private $txCommands;
+
+    /**
+     * @inheritdoc
+     */
+    public function __construct(CommandQueueInterface $queue)
+    {
+        $this->checkInnerQueue($queue);
+
+        $this->innerQueue = $queue;
+        $this->txCommands = new SplQueue;
+    }
+
+    /**
+     * Verifies if the inner command queue is compatibile with atomic queues.
+     *
+     * @throws InvalidArgumentException
+     */
+    protected function checkInnerQueue(CommandQueueInterface $queue): void
+    {
+        if ($queue instanceof static) {
+            throw new InvalidArgumentException('Atomic command queues cannot be nested');
+        }
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function count(): int
+    {
+        return count($this->txCommands);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function reset(): void
+    {
+        $this->txCommands = new SplQueue();
+        $this->innerQueue->reset();
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function enqueue(CommandInterface $command): void
+    {
+        if ($this->txCommands->isEmpty()) {
+            $this->innerQueue->enqueue(RawCommand::create('MULTI'));
+        }
+
+        if ($this->discarded) {
+            return;
+        }
+
+        if (0 === strcasecmp($command->getId(), 'DISCARD')) {
+            $this->discarded = true;
+            $this->reset();
+
+            return;
+        }
+
+        $this->txCommands->enqueue($command);
+        $this->innerQueue->enqueue($command);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function flush(ConnectionInterface $connection): Traversable
+    {
+        if ($this->txCommands->isEmpty()) {
+            yield from [];
+
+            return 0;
+        }
+
+        $this->innerQueue->enqueue(RawCommand::create('EXEC'));
+
+        $pending = count($this->txCommands);
+        $responses = new NoRewindIterator($this->innerQueue->flush($connection));
+
+        if (!$responses->valid()) {
+            yield from [];
+
+            return 0;
+        }
+
+        // NOTE: first queued command is `MULTI` so we check its response as it
+        // could return "-ERR" if Redis detects a nested `MULTI` when the atomic
+        // queue is flushed over a connection in a running MULTI / EXEC context.
+        if ($responses->current() instanceof ErrorInterface) {
+            throw $this->onLogicError($connection, $responses->current()->getMessage());
+        }
+
+        for ($i = 0; $i <= $pending; $i++) {
+            $responses->next();
+        }
+
+        /** @var iterable|ErrorInterface */
+        $execResponse = $responses->current();
+
+        $this->ensureValidExecResponse($connection, $execResponse);
+
+        foreach ($execResponse as $response) {
+            yield $this->txCommands->dequeue() => $response;
+        }
+
+        return count($execResponse);
+    }
+
+    /**
+     * Makes sure the response payload returned by `EXEC` is valid.
+     *
+     * @param ConnectionInterface     $connection   Target connection
+     * @param iterable|ErrorInterface $execResponse Response returned by `EXEC`
+     *
+     * @throws AtomicFlushException when the response returned by `EXEC` is not valid
+     */
+    protected function ensureValidExecResponse(ConnectionInterface $connection, $execResponse): void
+    {
+        if (null === $execResponse) {
+            $this->reset();
+
+            throw $this->onAbortedError($connection, 'Transaction discarded because of previous errors (NULL response)');
+        } elseif ($execResponse instanceof ErrorInterface) {
+            $this->reset();
+
+            if ('EXECABORT' === $execResponse->getErrorType()) {
+                throw $this->onAbortedError($connection, 'Transaction discarded because of previous errors (-EXECABORT response)');
+            } else {
+                throw $this->onLogicError($connection, $execResponse->getMessage());
+            }
+        } elseif (!is_iterable($execResponse)) {
+            $this->reset();
+            $connection->disconnect();
+
+            throw $this->onStateError($connection, sprintf(
+                'Protocol desynchronization detected on `EXEC` response (array expected, `%s` received)',
+                is_object($execResponse) ? get_class($execResponse) : gettype($execResponse)
+            ));
+        }
+    }
+
+    /**
+     * Returns exception for atomic state errors on queue flush.
+     *
+     * @param ConnectionInterface $connection Connection associated to the exception
+     * @param string              $message    Exception message
+     *
+     * @return Throwable
+     */
+    protected function onStateError(ConnectionInterface $connection, string $message): Throwable
+    {
+        return new AtomicFlushException($connection, $message, AtomicFlushException::TX_STATE);
+    }
+
+    /**
+     * Returns exception for atomic logic errors on queue flush.
+     *
+     * @param ConnectionInterface $connection Connection associated to the exception
+     * @param string              $message    Exception message
+     *
+     * @return Throwable
+     */
+    protected function onLogicError(ConnectionInterface $connection, string $message): Throwable
+    {
+        return new AtomicFlushException($connection, $message, AtomicFlushException::TX_LOGIC);
+    }
+
+    /**
+     * Returns exception for aborted atomic on queue flush.
+     *
+     * @param ConnectionInterface $connection Connection associated to the exception
+     * @param string              $message    Exception message
+     *
+     * @return Throwable
+     */
+    protected function onAbortedError(ConnectionInterface $connection, string $message): Throwable
+    {
+        return new AtomicFlushException($connection, $message, AtomicFlushException::TX_ABORT);
+    }
+}

src/Pipeline/Queue/AtomicFlushException.php

@@ -0,0 +1,57 @@
+<?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\Queue;
+
+use Predis\Connection\ConnectionInterface;
+use Throwable;
+
+/**
+ * Exception class for command queue errors during flush operations.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class AtomicFlushException extends FlushException
+{
+    const TX_STATE = 0b001;
+    const TX_LOGIC = 0b010;
+    const TX_ABORT = 0b100;
+
+    /**
+     * Returns whether the exception code matches a transaction logic error.
+     *
+     * @return bool
+     */
+    public function isStateError(): bool
+    {
+        return $this->getCode() === self::TX_STATE;
+    }
+
+    /**
+     * Returns whether the exception code matches a transaction logic error.
+     *
+     * @return bool
+     */
+    public function isLogicError(): bool
+    {
+        return $this->getCode() === self::TX_LOGIC;
+    }
+
+    /**
+     * Returns whether the exception code matches an aborted transaction.
+     *
+     * @return bool
+     */
+    public function isAbortedError(): bool
+    {
+        return $this->getCode() === self::TX_ABORT;
+    }
+}

src/Pipeline/Queue/Basic.php

@@ -0,0 +1,95 @@
+<?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\Queue;
+
+use Traversable;
+use Predis\Command\CommandInterface;
+use Predis\Connection\ConnectionInterface;
+
+/**
+ * Standard command queue used by default for pipelining.
+ */
+class Basic extends CommandQueue
+{
+    /**
+     * Writes a command to the target connection.
+     *
+     * @param ConnectionInterface $connection Target connection
+     * @param CommandInterface    $command    Command to be written on the target connection
+     *
+     * @return void
+     */
+    public function writeQueuedCommand(ConnectionInterface $connection, CommandInterface $command): void
+    {
+        $connection->writeRequest($command);
+    }
+
+    /**
+     * Reads a response for a command from the target connection.
+     *
+     * @param ConnectionInterface $connection Target connection
+     * @param CommandInterface    $command    Command associated to the pending response in the target connection
+     *
+     * @return mixed
+     */
+    public function readQueuedResponse(ConnectionInterface $connection, CommandInterface $command)
+    {
+        return $connection->readResponse($command);
+    }
+
+    /**
+     * Writes queued commands to the target connection.
+     *
+     * @param ConnectionInterface $connection Target connection
+     *
+     * @return void
+     */
+    protected function writeQueuedCommands(ConnectionInterface $connection): void
+    {
+        foreach ($this->getQueue() as $command) {
+            $this->writeQueuedCommand($connection, $command);
+        }
+    }
+
+    /**
+     * Reads pending responses from the target connection.
+     *
+     * @param ConnectionInterface $connection Target connections
+     *
+     * @return Traversable
+     */
+    protected function readQueuedResponses(ConnectionInterface $connection): Traversable
+    {
+        $commands = $this->getQueue();
+        $dequeued = count($commands);
+
+        while (!$commands->isEmpty()) {
+            $command = $commands->dequeue();
+            $response = $this->readQueuedResponse($connection, $command);
+
+            yield $command => $response;
+        }
+
+        return $dequeued;
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function flush(ConnectionInterface $connection): Traversable
+    {
+        $this->writeQueuedCommands($connection);
+        $dequeued = $this->readQueuedResponses($connection);
+
+        return $dequeued;
+    }
+}

src/Pipeline/Queue/CommandQueue.php

@@ -0,0 +1,75 @@
+<?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\Queue;
+
+use SplQueue;
+use Traversable;
+use Predis\Command\CommandInterface;
+use Predis\Connection\ConnectionInterface;
+
+/**
+ * Base class for implementing a command queue.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+abstract class CommandQueue implements CommandQueueInterface
+{
+    /** @var SplQueue */
+    private $queue;
+
+    /**
+     * Initializes a new command queue.
+     */
+    public function __construct()
+    {
+        $this->queue = new SplQueue();
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function count(): int
+    {
+        return count($this->queue);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function reset(): void
+    {
+        $this->queue = new SplQueue();
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function enqueue(CommandInterface $command): void
+    {
+        $this->queue->enqueue($command);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public abstract function flush(ConnectionInterface $connection): Traversable;
+
+    /**
+     * Returns the underlying queue storage.
+     *
+     * @return SplQueue
+     */
+    protected function getQueue(): SplQueue
+    {
+        return $this->queue;
+    }
+}

src/Pipeline/Queue/CommandQueueException.php

@@ -0,0 +1,23 @@
+<?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\Queue;
+
+use Predis\PredisException;
+
+/**
+ * Generic exception class for command queues errors.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+abstract class CommandQueueException extends PredisException
+{
+}

src/Pipeline/Queue/CommandQueueInterface.php

@@ -0,0 +1,65 @@
+<?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\Queue;
+
+use Countable;
+use Traversable;
+use Predis\Command\CommandInterface;
+use Predis\Connection\ConnectionInterface;
+
+/**
+ * Defines the minimum API interface for a command queue.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+interface CommandQueueInterface extends Countable
+{
+    /**
+     * Returns the number of commands in the queue.
+     *
+     * @return integer
+     */
+    public function count(): int;
+
+    /**
+     * Discards queued commands and resets the state of the queue.
+     *
+     * @return void
+     */
+    public function reset(): void;
+
+    /**
+     * Puts a command in the queue.
+     *
+     * @param CommandInterface $command Command to be queued
+     *
+     * @throws EnqueueException when an error occurs while trying to enqueue a command
+     *
+     * @return void
+     */
+    public function enqueue(CommandInterface $command): void;
+
+    /**
+     * Flushes queued commands to the connection and returns their responses.
+     *
+     * This method returns a Traversable that MUST be consumed by the caller to
+     * make sure that pending responses on the connection are properly dequeued
+     * and prevent any protocol desynchronization issue.
+     *
+     * @param ConnectionInterface $connection Target connection
+     *
+     * @throws FlushException when an error occurs while trying to flush the queue
+     *
+     * @return Traversable
+     */
+    public function flush(ConnectionInterface $connection): Traversable;
+}

src/Pipeline/Queue/EnqueueException.php

@@ -0,0 +1,21 @@
+<?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\Queue;
+
+/**
+ * Exception class for command queue errors during enqueue operations.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class EnqueueException extends CommandQueueException
+{
+}

src/Pipeline/Queue/FireAndForget.php

@@ -0,0 +1,43 @@
+<?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\Queue;
+
+use Traversable;
+use Predis\Connection\ConnectionInterface;
+
+/**
+ * Fire-and-forget command queue.
+ *
+ * This command queue simply writes enqueued commands to the target connection
+ * and then ignores any response by dropping the underlying connection.
+ */
+class FireAndForget extends CommandQueue
+{
+    /**
+     * @inheritdoc
+     */
+    public function flush(ConnectionInterface $connection): Traversable
+    {
+        $commands = $this->getQueue();
+        $dequeued = count($commands);
+
+        while (!$commands->isEmpty()) {
+            $connection->writeRequest($commands->dequeue());
+        }
+
+        $connection->disconnect();
+
+        yield from [];
+
+        return $dequeued;
+    }
+}

src/Pipeline/Queue/FlushException.php

@@ -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\Queue;
+
+use Predis\Connection\ConnectionInterface;
+use Throwable;
+
+/**
+ * Exception class for command queue errors during flush operations.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class FlushException extends CommandQueueException
+{
+    protected $connection;
+
+    /**
+     * @param ConnectionInterface $connection Connection 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(ConnectionInterface $connection, string $message = '', int $code = 0, Throwable $previous = null)
+    {
+        parent::__construct($message, $code, $previous);
+
+        $this->connection = $connection;
+    }
+
+    /**
+     * Returns the connection associated to the exception.
+     *
+     * @return ConnectionInterface
+     */
+    public function getConnection(): ConnectionInterface
+    {
+        return $this->connection;
+    }
+}

src/Transaction/MultiExec.php

@@ -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');