Implement command queues as the new basis for pipelining.

The scope of these classes in the new Predis\Pipeline\Queue namespace is
to take instances of Predis\Command\CommandInterface and enqueue them so
that they can be flushed to the desired target connection.

As a general rule, unless otherwise stated, flushing the queue returns a
traversable that MUST always be consumed by the caller to make sure all
pending responses on the target connection are cleared to avoid protocol
desynchronization issues.

Queue strategies implemented in the Predis\Pipeline\Queue namespace are:

- Basic: default command queue that implements a standard queue strategy
  where commands are enqueued until a flush is explicitly requested over
  the target connection.

- FireAndForget: queued commands are sent over the target connection but
  responses are not read back from the server. This is accomplished by
  closing the underlying connection in order to quickly drop any pending
  response, meaning that the returned traversable can be safely ignored.

- Atomic: wraps an existing command queue instance in a MULTI/EXEC block
  so that the whole pipeline is flushed in a transaction hiding all the
  required logic. Due to the very nature of MULTI/EXEC, the traversable
  instance from the inner queue is automatically consumed so only actual
  responses in EXEC response payload are returned to the caller.

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