code:
<?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\Connection;

use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\Status as StatusResponse;

/**
 * This class provides the implementation of a Predis connection that uses the
 * PHP socket extension for network communication and wraps the phpiredis C
 * extension (PHP bindings for hiredis) to parse the Redis protocol.
 *
 * This class is intended to provide an optional low-overhead alternative for
 * processing responses from Redis compared to the standard pure-PHP classes.
 * Differences in speed when dealing with short inline responses are practically
 * nonexistent, the actual speed boost is for big multibulk responses when this
 * protocol processor can parse and return responses very fast.
 *
 * For instructions on how to build and install the phpiredis extension, please
 * consult the repository of the project.
 *
 * The connection parameters supported by this class are:
 *
 *  - scheme: it can be either 'redis', 'tcp' or 'unix'.
 *  - host: hostname or IP address of the server.
 *  - port: TCP port of the server.
 *  - path: path of a UNIX domain socket when scheme is 'unix'.
 *  - timeout: timeout to perform the connection (default is 5 seconds).
 *  - read_write_timeout: timeout of read / write operations.
 *
 * @link http://github.com/nrk/phpiredis
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class PhpiredisSocketConnection extends AbstractConnection
{
    private $reader;

    /**
     * {@inheritdoc}
     */
    public function __construct(ParametersInterface $parameters)
    {
        $this->assertExtensions();

        parent::__construct($parameters);

        $this->reader = $this->createReader();
    }

    /**
     * Disconnects from the server and destroys the underlying resource and the
     * protocol reader resource when PHP's garbage collector kicks in.
     */
    public function __destruct()
    {
        phpiredis_reader_destroy($this->reader);

        parent::__destruct();
    }

    /**
     * Checks if the socket and phpiredis extensions are loaded in PHP.
     */
    protected function assertExtensions()
    {
        if (!extension_loaded('sockets')) {
            throw new NotSupportedException(
                'The "sockets" extension is required by this connection backend.'
            );
        }

        if (!extension_loaded('phpiredis')) {
            throw new NotSupportedException(
                'The "phpiredis" extension is required by this connection backend.'
            );
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function assertParameters(ParametersInterface $parameters)
    {
        switch ($parameters->scheme) {
            case 'tcp':
            case 'redis':
            case 'unix':
                break;

            default:
                throw new \InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
        }

        if (isset($parameters->persistent)) {
            throw new NotSupportedException(
                'Persistent connections are not supported by this connection backend.'
            );
        }

        return $parameters;
    }

    /**
     * Creates a new instance of the protocol reader resource.
     *
     * @return resource
     */
    private function createReader()
    {
        $reader = phpiredis_reader_create();

        phpiredis_reader_set_status_handler($reader, $this->getStatusHandler());
        phpiredis_reader_set_error_handler($reader, $this->getErrorHandler());

        return $reader;
    }

    /**
     * Returns the underlying protocol reader resource.
     *
     * @return resource
     */
    protected function getReader()
    {
        return $this->reader;
    }

    /**
     * Returns the handler used by the protocol reader for inline responses.
     *
     * @return \Closure
     */
    protected function getStatusHandler()
    {
        static $statusHandler;

        if (!$statusHandler) {
            $statusHandler = function ($payload) {
                return StatusResponse::get($payload);
            };
        }

        return $statusHandler;
    }

    /**
     * Returns the handler used by the protocol reader for error responses.
     *
     * @return \Closure
     */
    protected function getErrorHandler()
    {
        static $errorHandler;

        if (!$errorHandler) {
            $errorHandler = function ($errorMessage) {
                return new ErrorResponse($errorMessage);
            };
        }

        return $errorHandler;
    }

    /**
     * Helper method used to throw exceptions on socket errors.
     */
    private function emitSocketError()
    {
        $errno = socket_last_error();
        $errstr = socket_strerror($errno);

        $this->disconnect();

        $this->onConnectionError(trim($errstr), $errno);
    }

    /**
     * Gets the address of an host from connection parameters.
     *
     * @param ParametersInterface $parameters Parameters used to initialize the connection.
     *
     * @return string
     */
    protected static function getAddress(ParametersInterface $parameters)
    {
        if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP)) {
            return $host;
        }

        if ($host === $address = gethostbyname($host)) {
            return false;
        }

        return $address;
    }

    /**
     * {@inheritdoc}
     */
    protected function createResource()
    {
        $parameters = $this->parameters;

        if ($parameters->scheme === 'unix') {
            $address = $parameters->path;
            $domain = AF_UNIX;
            $protocol = 0;
        } else {
            if (false === $address = self::getAddress($parameters)) {
                $this->onConnectionError("Cannot resolve the address of '$parameters->host'.");
            }

            $domain = filter_var($address, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6) ? AF_INET6 : AF_INET;
            $protocol = SOL_TCP;
        }

        $socket = @socket_create($domain, SOCK_STREAM, $protocol);

        if (!is_resource($socket)) {
            $this->emitSocketError();
        }

        $this->setSocketOptions($socket, $parameters);
        $this->connectWithTimeout($socket, $address, $parameters);

        return $socket;
    }

    /**
     * Sets options on the socket resource from the connection parameters.
     *
     * @param resource            $socket     Socket resource.
     * @param ParametersInterface $parameters Parameters used to initialize the connection.
     */
    private function setSocketOptions($socket, ParametersInterface $parameters)
    {
        if ($parameters->scheme !== 'unix') {
            if (!socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1)) {
                $this->emitSocketError();
            }

            if (!socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1)) {
                $this->emitSocketError();
            }
        }

        if (isset($parameters->read_write_timeout)) {
            $rwtimeout = (float) $parameters->read_write_timeout;
            $timeoutSec = floor($rwtimeout);
            $timeoutUsec = ($rwtimeout - $timeoutSec) * 1000000;

            $timeout = array(
                'sec' => $timeoutSec,
                'usec' => $timeoutUsec,
            );

            if (!socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout)) {
                $this->emitSocketError();
            }

            if (!socket_set_option($socket, SOL_SOCKET, SO_RCVTIMEO, $timeout)) {
                $this->emitSocketError();
            }
        }
    }

    /**
     * Opens the actual connection to the server with a timeout.
     *
     * @param resource            $socket     Socket resource.
     * @param string              $address    IP address (DNS-resolved from hostname)
     * @param ParametersInterface $parameters Parameters used to initialize the connection.
     *
     * @return string
     */
    private function connectWithTimeout($socket, $address, ParametersInterface $parameters)
    {
        socket_set_nonblock($socket);

        if (@socket_connect($socket, $address, (int) $parameters->port) === false) {
            $error = socket_last_error();

            if ($error != SOCKET_EINPROGRESS && $error != SOCKET_EALREADY) {
                $this->emitSocketError();
            }
        }

        socket_set_block($socket);

        $null = null;
        $selectable = array($socket);

        $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);
        $timeoutSecs = floor($timeout);
        $timeoutUSecs = ($timeout - $timeoutSecs) * 1000000;

        $selected = socket_select($selectable, $selectable, $null, $timeoutSecs, $timeoutUSecs);

        if ($selected === 2) {
            $this->onConnectionError('Connection refused.', SOCKET_ECONNREFUSED);
        }

        if ($selected === 0) {
            $this->onConnectionError('Connection timed out.', SOCKET_ETIMEDOUT);
        }

        if ($selected === false) {
            $this->emitSocketError();
        }
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        if (parent::connect() && $this->initCommands) {
            foreach ($this->initCommands as $command) {
                $response = $this->executeCommand($command);

                if ($response instanceof ErrorResponseInterface) {
                    $this->onConnectionError("`{$command->getId()}` failed: $response", 0);
                }
            }
        }
    }

    /**
     * {@inheritdoc}
     */
    public function disconnect()
    {
        if ($this->isConnected()) {
            socket_close($this->getResource());
            parent::disconnect();
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function write($buffer)
    {
        $socket = $this->getResource();

        while (($length = strlen($buffer)) > 0) {
            $written = socket_write($socket, $buffer, $length);

            if ($length === $written) {
                return;
            }

            if ($written === false) {
                $this->onConnectionError('Error while writing bytes to the server.');
            }

            $buffer = substr($buffer, $written);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $socket = $this->getResource();
        $reader = $this->reader;

        while (PHPIREDIS_READER_STATE_INCOMPLETE === $state = phpiredis_reader_get_state($reader)) {
            if (@socket_recv($socket, $buffer, 4096, 0) === false || $buffer === '' || $buffer === null) {
                $this->emitSocketError();
            }

            phpiredis_reader_feed($reader, $buffer);
        }

        if ($state === PHPIREDIS_READER_STATE_COMPLETE) {
            return phpiredis_reader_get_reply($reader);
        } else {
            $this->onProtocolError(phpiredis_reader_get_error($reader));

            return;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $arguments = $command->getArguments();
        array_unshift($arguments, $command->getId());

        $this->write(phpiredis_format_command($arguments));
    }

    /**
     * {@inheritdoc}
     */
    public function __wakeup()
    {
        $this->assertExtensions();
        $this->reader = $this->createReader();
    }
}


code:
<?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\Connection;

use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\Status as StatusResponse;

/**
 * This class provides the implementation of a Predis connection that uses PHP's
 * streams for network communication and wraps the phpiredis C extension (PHP
 * bindings for hiredis) to parse and serialize the Redis protocol.
 *
 * This class is intended to provide an optional low-overhead alternative for
 * processing responses from Redis compared to the standard pure-PHP classes.
 * Differences in speed when dealing with short inline responses are practically
 * nonexistent, the actual speed boost is for big multibulk responses when this
 * protocol processor can parse and return responses very fast.
 *
 * For instructions on how to build and install the phpiredis extension, please
 * consult the repository of the project.
 *
 * The connection parameters supported by this class are:
 *
 *  - scheme: it can be either 'redis', 'tcp' or 'unix'.
 *  - host: hostname or IP address of the server.
 *  - port: TCP port of the server.
 *  - path: path of a UNIX domain socket when scheme is 'unix'.
 *  - timeout: timeout to perform the connection.
 *  - read_write_timeout: timeout of read / write operations.
 *  - async_connect: performs the connection asynchronously.
 *  - tcp_nodelay: enables or disables Nagle's algorithm for coalescing.
 *  - persistent: the connection is left intact after a GC collection.
 *
 * @link https://github.com/nrk/phpiredis
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class PhpiredisStreamConnection extends StreamConnection
{
    private $reader;

    /**
     * {@inheritdoc}
     */
    public function __construct(ParametersInterface $parameters)
    {
        $this->assertExtensions();

        parent::__construct($parameters);

        $this->reader = $this->createReader();
    }

    /**
     * {@inheritdoc}
     */
    public function __destruct()
    {
        phpiredis_reader_destroy($this->reader);

        parent::__destruct();
    }

    /**
     * Checks if the phpiredis extension is loaded in PHP.
     */
    private function assertExtensions()
    {
        if (!extension_loaded('phpiredis')) {
            throw new NotSupportedException(
                'The "phpiredis" extension is required by this connection backend.'
            );
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function assertSslSupport(ParametersInterface $parameters)
    {
        throw new \InvalidArgumentException('SSL encryption is not supported by this connection backend.');
    }

    /**
     * {@inheritdoc}
     */
    protected function createStreamSocket(ParametersInterface $parameters, $address, $flags, $context = null)
    {
        $socket = null;
        $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);

        $resource = @stream_socket_client($address, $errno, $errstr, $timeout, $flags);

        if (!$resource) {
            $this->onConnectionError(trim($errstr), $errno);
        }

        if (isset($parameters->read_write_timeout) && function_exists('socket_import_stream')) {
            $rwtimeout = (float) $parameters->read_write_timeout;
            $rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;

            $timeout = array(
                'sec' => $timeoutSeconds = floor($rwtimeout),
                'usec' => ($rwtimeout - $timeoutSeconds) * 1000000,
            );

            $socket = $socket ?: socket_import_stream($resource);
            @socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout);
            @socket_set_option($socket, SOL_SOCKET, SO_RCVTIMEO, $timeout);
        }

        if (isset($parameters->tcp_nodelay) && function_exists('socket_import_stream')) {
            $socket = $socket ?: socket_import_stream($resource);
            socket_set_option($socket, SOL_TCP, TCP_NODELAY, (int) $parameters->tcp_nodelay);
        }

        return $resource;
    }

    /**
     * Creates a new instance of the protocol reader resource.
     *
     * @return resource
     */
    private function createReader()
    {
        $reader = phpiredis_reader_create();

        phpiredis_reader_set_status_handler($reader, $this->getStatusHandler());
        phpiredis_reader_set_error_handler($reader, $this->getErrorHandler());

        return $reader;
    }

    /**
     * Returns the underlying protocol reader resource.
     *
     * @return resource
     */
    protected function getReader()
    {
        return $this->reader;
    }

    /**
     * Returns the handler used by the protocol reader for inline responses.
     *
     * @return \Closure
     */
    protected function getStatusHandler()
    {
        static $statusHandler;

        if (!$statusHandler) {
            $statusHandler = function ($payload) {
                return StatusResponse::get($payload);
            };
        }

        return $statusHandler;
    }

    /**
     * Returns the handler used by the protocol reader for error responses.
     *
     * @return \Closure
     */
    protected function getErrorHandler()
    {
        static $errorHandler;

        if (!$errorHandler) {
            $errorHandler = function ($errorMessage) {
                return new ErrorResponse($errorMessage);
            };
        }

        return $errorHandler;
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $socket = $this->getResource();
        $reader = $this->reader;

        while (PHPIREDIS_READER_STATE_INCOMPLETE === $state = phpiredis_reader_get_state($reader)) {
            $buffer = stream_socket_recvfrom($socket, 4096);

            if ($buffer === false || $buffer === '') {
                $this->onConnectionError('Error while reading bytes from the server.');
            }

            phpiredis_reader_feed($reader, $buffer);
        }

        if ($state === PHPIREDIS_READER_STATE_COMPLETE) {
            return phpiredis_reader_get_reply($reader);
        } else {
            $this->onProtocolError(phpiredis_reader_get_error($reader));

            return;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $arguments = $command->getArguments();
        array_unshift($arguments, $command->getId());

        $this->write(phpiredis_format_command($arguments));
    }

    /**
     * {@inheritdoc}
     */
    public function __wakeup()
    {
        $this->assertExtensions();
        $this->reader = $this->createReader();
    }
}

repo_name: spookie/gnu

license: mit


code:
<?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\Connection;

use Predis\Command\CommandInterface;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\Status as StatusResponse;

/**
 * Standard connection to Redis servers implemented on top of PHP's streams.
 * The connection parameters supported by this class are:.
 *
 *  - scheme: it can be either 'redis', 'tcp', 'rediss', 'tls' or 'unix'.
 *  - host: hostname or IP address of the server.
 *  - port: TCP port of the server.
 *  - path: path of a UNIX domain socket when scheme is 'unix'.
 *  - timeout: timeout to perform the connection (default is 5 seconds).
 *  - read_write_timeout: timeout of read / write operations.
 *  - async_connect: performs the connection asynchronously.
 *  - tcp_nodelay: enables or disables Nagle's algorithm for coalescing.
 *  - persistent: the connection is left intact after a GC collection.
 *  - ssl: context options array (see http://php.net/manual/en/context.ssl.php)
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class StreamConnection extends AbstractConnection
{
    /**
     * Disconnects from the server and destroys the underlying resource when the
     * garbage collector kicks in only if the connection has not been marked as
     * persistent.
     */
    public function __destruct()
    {
        if (isset($this->parameters->persistent) && $this->parameters->persistent) {
            return;
        }

        $this->disconnect();
    }

    /**
     * {@inheritdoc}
     */
    protected function assertParameters(ParametersInterface $parameters)
    {
        switch ($parameters->scheme) {
            case 'tcp':
            case 'redis':
            case 'unix':
                break;

            case 'tls':
            case 'rediss':
                $this->assertSslSupport($parameters);
                break;

            default:
                throw new \InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
        }

        return $parameters;
    }

    /**
     * Checks needed conditions for SSL-encrypted connections.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @throws \InvalidArgumentException
     */
    protected function assertSslSupport(ParametersInterface $parameters)
    {
        if (
            filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN) &&
            version_compare(PHP_VERSION, '7.0.0beta') < 0
        ) {
            throw new \InvalidArgumentException('Persistent SSL connections require PHP >= 7.0.0.');
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function createResource()
    {
        switch ($this->parameters->scheme) {
            case 'tcp':
            case 'redis':
                return $this->tcpStreamInitializer($this->parameters);

            case 'unix':
                return $this->unixStreamInitializer($this->parameters);

            case 'tls':
            case 'rediss':
                return $this->tlsStreamInitializer($this->parameters);

            default:
                throw new \InvalidArgumentException("Invalid scheme: '{$this->parameters->scheme}'.");
        }
    }

    /**
     * Creates a connected stream socket resource.
     *
     * @param ParametersInterface $parameters Connection parameters.
     * @param string              $address    Address for stream_socket_client().
     * @param int                 $flags      Flags for stream_socket_client().
     *
     * @return resource
     */
    protected function createStreamSocket(ParametersInterface $parameters, $address, $flags)
    {
        $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);

        if (!$resource = @stream_socket_client($address, $errno, $errstr, $timeout, $flags)) {
            $this->onConnectionError(trim($errstr), $errno);
        }

        if (isset($parameters->read_write_timeout)) {
            $rwtimeout = (float) $parameters->read_write_timeout;
            $rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;
            $timeoutSeconds = floor($rwtimeout);
            $timeoutUSeconds = ($rwtimeout - $timeoutSeconds) * 1000000;
            stream_set_timeout($resource, $timeoutSeconds, $timeoutUSeconds);
        }

        if (isset($parameters->tcp_nodelay) && function_exists('socket_import_stream')) {
            $socket = socket_import_stream($resource);
            socket_set_option($socket, SOL_TCP, TCP_NODELAY, (int) $parameters->tcp_nodelay);
        }

        return $resource;
    }

    /**
     * Initializes a TCP stream resource.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return resource
     */
    protected function tcpStreamInitializer(ParametersInterface $parameters)
    {
        if (!filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
            $address = "tcp://$parameters->host:$parameters->port";
        } else {
            $address = "tcp://[$parameters->host]:$parameters->port";
        }

        $flags = STREAM_CLIENT_CONNECT;

        if (isset($parameters->async_connect) && $parameters->async_connect) {
            $flags |= STREAM_CLIENT_ASYNC_CONNECT;
        }

        if (isset($parameters->persistent)) {
            if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {
                $flags |= STREAM_CLIENT_PERSISTENT;

                if ($persistent === null) {
                    $address = "{$address}/{$parameters->persistent}";
                }
            }
        }

        $resource = $this->createStreamSocket($parameters, $address, $flags);

        return $resource;
    }

    /**
     * Initializes a UNIX stream resource.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return resource
     */
    protected function unixStreamInitializer(ParametersInterface $parameters)
    {
        if (!isset($parameters->path)) {
            throw new \InvalidArgumentException('Missing UNIX domain socket path.');
        }

        $flags = STREAM_CLIENT_CONNECT;

        if (isset($parameters->persistent)) {
            if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {
                $flags |= STREAM_CLIENT_PERSISTENT;

                if ($persistent === null) {
                    throw new \InvalidArgumentException(
                        'Persistent connection IDs are not supported when using UNIX domain sockets.'
                    );
                }
            }
        }

        $resource = $this->createStreamSocket($parameters, "unix://{$parameters->path}", $flags);

        return $resource;
    }

    /**
     * Initializes a SSL-encrypted TCP stream resource.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return resource
     */
    protected function tlsStreamInitializer(ParametersInterface $parameters)
    {
        $resource = $this->tcpStreamInitializer($parameters);
        $metadata = stream_get_meta_data($resource);

        // Detect if crypto mode is already enabled for this stream (PHP >= 7.0.0).
        if (isset($metadata['crypto'])) {
            return $resource;
        }

        if (is_array($parameters->ssl)) {
            $options = $parameters->ssl;
        } else {
            $options = array();
        }

        if (!isset($options['crypto_type'])) {
            $options['crypto_type'] = STREAM_CRYPTO_METHOD_TLS_CLIENT;
        }

        if (!stream_context_set_option($resource, array('ssl' => $options))) {
            $this->onConnectionError('Error while setting SSL context options');
        }

        if (!stream_socket_enable_crypto($resource, true, $options['crypto_type'])) {
            $this->onConnectionError('Error while switching to encrypted communication');
        }

        return $resource;
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        if (parent::connect() && $this->initCommands) {
            foreach ($this->initCommands as $command) {
                $response = $this->executeCommand($command);

                if ($response instanceof ErrorResponseInterface) {
                    $this->onConnectionError("`{$command->getId()}` failed: $response", 0);
                }
            }
        }
    }

    /**
     * {@inheritdoc}
     */
    public function disconnect()
    {
        if ($this->isConnected()) {
            fclose($this->getResource());
            parent::disconnect();
        }
    }

    /**
     * Performs a write operation over the stream of the buffer containing a
     * command serialized with the Redis wire protocol.
     *
     * @param string $buffer Representation of a command in the Redis wire protocol.
     */
    protected function write($buffer)
    {
        $socket = $this->getResource();

        while (($length = strlen($buffer)) > 0) {
            $written = @fwrite($socket, $buffer);

            if ($length === $written) {
                return;
            }

            if ($written === false || $written === 0) {
                $this->onConnectionError('Error while writing bytes to the server.');
            }

            $buffer = substr($buffer, $written);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $socket = $this->getResource();
        $chunk = fgets($socket);

        if ($chunk === false || $chunk === '') {
            $this->onConnectionError('Error while reading line from the server.');
        }

        $prefix = $chunk[0];
        $payload = substr($chunk, 1, -2);

        switch ($prefix) {
            case '+':
                return StatusResponse::get($payload);

            case '$':
                $size = (int) $payload;

                if ($size === -1) {
                    return;
                }

                $bulkData = '';
                $bytesLeft = ($size += 2);

                do {
                    $chunk = fread($socket, min($bytesLeft, 4096));

                    if ($chunk === false || $chunk === '') {
                        $this->onConnectionError('Error while reading bytes from the server.');
                    }

                    $bulkData .= $chunk;
                    $bytesLeft = $size - strlen($bulkData);
                } while ($bytesLeft > 0);

                return substr($bulkData, 0, -2);

            case '*':
                $count = (int) $payload;

                if ($count === -1) {
                    return;
                }

                $multibulk = array();

                for ($i = 0; $i < $count; ++$i) {
                    $multibulk[$i] = $this->read();
                }

                return $multibulk;

            case ':':
                $integer = (int) $payload;
                return $integer == $payload ? $integer : $payload;

            case '-':
                return new ErrorResponse($payload);

            default:
                $this->onProtocolError("Unknown response prefix: '$prefix'.");

                return;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $commandID = $command->getId();
        $arguments = $command->getArguments();

        $cmdlen = strlen($commandID);
        $reqlen = count($arguments) + 1;

        $buffer = "*{$reqlen}\r\n\${$cmdlen}\r\n{$commandID}\r\n";

        foreach ($arguments as $argument) {
            $arglen = strlen($argument);
            $buffer .= "\${$arglen}\r\n{$argument}\r\n";
        }

        $this->write($buffer);
    }
}


code:
<?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\Connection;

use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Protocol\ProtocolException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\Status as StatusResponse;

/**
 * This class implements a Predis connection that actually talks with Webdis
 * instead of connecting directly to Redis. It relies on the cURL extension to
 * communicate with the web server and the phpiredis extension to parse the
 * protocol for responses returned in the http response bodies.
 *
 * Some features are not yet available or they simply cannot be implemented:
 *   - Pipelining commands.
 *   - Publish / Subscribe.
 *   - MULTI / EXEC transactions (not yet supported by Webdis).
 *
 * The connection parameters supported by this class are:
 *
 *  - scheme: must be 'http'.
 *  - host: hostname or IP address of the server.
 *  - port: TCP port of the server.
 *  - timeout: timeout to perform the connection (default is 5 seconds).
 *  - user: username for authentication.
 *  - pass: password for authentication.
 *
 * @link http://webd.is
 * @link http://github.com/nicolasff/webdis
 * @link http://github.com/seppo0010/phpiredis
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class WebdisConnection implements NodeConnectionInterface
{
    private $parameters;
    private $resource;
    private $reader;

    /**
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @throws \InvalidArgumentException
     */
    public function __construct(ParametersInterface $parameters)
    {
        $this->assertExtensions();

        if ($parameters->scheme !== 'http') {
            throw new \InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
        }

        $this->parameters = $parameters;

        $this->resource = $this->createCurl();
        $this->reader = $this->createReader();
    }

    /**
     * Frees the underlying cURL and protocol reader resources when the garbage
     * collector kicks in.
     */
    public function __destruct()
    {
        curl_close($this->resource);
        phpiredis_reader_destroy($this->reader);
    }

    /**
     * Helper method used to throw on unsupported methods.
     *
     * @param string $method Name of the unsupported method.
     *
     * @throws NotSupportedException
     */
    private function throwNotSupportedException($method)
    {
        $class = __CLASS__;
        throw new NotSupportedException("The method $class::$method() is not supported.");
    }

    /**
     * Checks if the cURL and phpiredis extensions are loaded in PHP.
     */
    private function assertExtensions()
    {
        if (!extension_loaded('curl')) {
            throw new NotSupportedException(
                'The "curl" extension is required by this connection backend.'
            );
        }

        if (!extension_loaded('phpiredis')) {
            throw new NotSupportedException(
                'The "phpiredis" extension is required by this connection backend.'
            );
        }
    }

    /**
     * Initializes cURL.
     *
     * @return resource
     */
    private function createCurl()
    {
        $parameters = $this->getParameters();
        $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0) * 1000;

        if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP)) {
            $host = "[$host]";
        }

        $options = array(
            CURLOPT_FAILONERROR => true,
            CURLOPT_CONNECTTIMEOUT_MS => $timeout,
            CURLOPT_URL => "$parameters->scheme://$host:$parameters->port",
            CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
            CURLOPT_POST => true,
            CURLOPT_WRITEFUNCTION => array($this, 'feedReader'),
        );

        if (isset($parameters->user, $parameters->pass)) {
            $options[CURLOPT_USERPWD] = "{$parameters->user}:{$parameters->pass}";
        }

        curl_setopt_array($resource = curl_init(), $options);

        return $resource;
    }

    /**
     * Initializes the phpiredis protocol reader.
     *
     * @return resource
     */
    private function createReader()
    {
        $reader = phpiredis_reader_create();

        phpiredis_reader_set_status_handler($reader, $this->getStatusHandler());
        phpiredis_reader_set_error_handler($reader, $this->getErrorHandler());

        return $reader;
    }

    /**
     * Returns the handler used by the protocol reader for inline responses.
     *
     * @return \Closure
     */
    protected function getStatusHandler()
    {
        static $statusHandler;

        if (!$statusHandler) {
            $statusHandler = function ($payload) {
                return StatusResponse::get($payload);
            };
        }

        return $statusHandler;
    }

    /**
     * Returns the handler used by the protocol reader for error responses.
     *
     * @return \Closure
     */
    protected function getErrorHandler()
    {
        static $errorHandler;

        if (!$errorHandler) {
            $errorHandler = function ($errorMessage) {
                return new ErrorResponse($errorMessage);
            };
        }

        return $errorHandler;
    }

    /**
     * Feeds the phpredis reader resource with the data read from the network.
     *
     * @param resource $resource Reader resource.
     * @param string   $buffer   Buffer of data read from a connection.
     *
     * @return int
     */
    protected function feedReader($resource, $buffer)
    {
        phpiredis_reader_feed($this->reader, $buffer);

        return strlen($buffer);
    }

    /**
     * {@inheritdoc}
     */
    public function connect()
    {
        // NOOP
    }

    /**
     * {@inheritdoc}
     */
    public function disconnect()
    {
        // NOOP
    }

    /**
     * {@inheritdoc}
     */
    public function isConnected()
    {
        return true;
    }

    /**
     * Checks if the specified command is supported by this connection class.
     *
     * @param CommandInterface $command Command instance.
     *
     * @throws NotSupportedException
     *
     * @return string
     */
    protected function getCommandId(CommandInterface $command)
    {
        switch ($commandID = $command->getId()) {
            case 'AUTH':
            case 'SELECT':
            case 'MULTI':
            case 'EXEC':
            case 'WATCH':
            case 'UNWATCH':
            case 'DISCARD':
            case 'MONITOR':
                throw new NotSupportedException("Command '$commandID' is not allowed by Webdis.");

            default:
                return $commandID;
        }
    }

    /**
     * {@inheritdoc}
     */
    public function writeRequest(CommandInterface $command)
    {
        $this->throwNotSupportedException(__FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function readResponse(CommandInterface $command)
    {
        $this->throwNotSupportedException(__FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function executeCommand(CommandInterface $command)
    {
        $resource = $this->resource;
        $commandId = $this->getCommandId($command);

        if ($arguments = $command->getArguments()) {
            $arguments = implode('/', array_map('urlencode', $arguments));
            $serializedCommand = "$commandId/$arguments.raw";
        } else {
            $serializedCommand = "$commandId.raw";
        }

        curl_setopt($resource, CURLOPT_POSTFIELDS, $serializedCommand);

        if (curl_exec($resource) === false) {
            $error = curl_error($resource);
            $errno = curl_errno($resource);

            throw new ConnectionException($this, trim($error), $errno);
        }

        if (phpiredis_reader_get_state($this->reader) !== PHPIREDIS_READER_STATE_COMPLETE) {
            throw new ProtocolException($this, phpiredis_reader_get_error($this->reader));
        }

        return phpiredis_reader_get_reply($this->reader);
    }

    /**
     * {@inheritdoc}
     */
    public function getResource()
    {
        return $this->resource;
    }

    /**
     * {@inheritdoc}
     */
    public function getParameters()
    {
        return $this->parameters;
    }

    /**
     * {@inheritdoc}
     */
    public function addConnectCommand(CommandInterface $command)
    {
        $this->throwNotSupportedException(__FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $this->throwNotSupportedException(__FUNCTION__);
    }

    /**
     * {@inheritdoc}
     */
    public function __toString()
    {
        return "{$this->parameters->host}:{$this->parameters->port}";
    }

    /**
     * {@inheritdoc}
     */
    public function __sleep()
    {
        return array('parameters');
    }

    /**
     * {@inheritdoc}
     */
    public function __wakeup()
    {
        $this->assertExtensions();

        $this->resource = $this->createCurl();
        $this->reader = $this->createReader();
    }
}


code:
<?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\Monitor;

use Predis\ClientInterface;
use Predis\Connection\AggregateConnectionInterface;
use Predis\NotSupportedException;

/**
 * Redis MONITOR consumer.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class Consumer implements \Iterator
{
    private $client;
    private $valid;
    private $position;

    /**
     * @param ClientInterface $client Client instance used by the consumer.
     */
    public function __construct(ClientInterface $client)
    {
        $this->assertClient($client);

        $this->client = $client;

        $this->start();
    }

    /**
     * Automatically stops the consumer when the garbage collector kicks in.
     */
    public function __destruct()
    {
        $this->stop();
    }

    /**
     * Checks if the passed client instance satisfies the required conditions
     * needed to initialize a monitor consumer.
     *
     * @param ClientInterface $client Client instance used by the consumer.
     *
     * @throws NotSupportedException
     */
    private function assertClient(ClientInterface $client)
    {
        if ($client->getConnection() instanceof AggregateConnectionInterface) {
            throw new NotSupportedException(
                'Cannot initialize a monitor consumer over aggregate connections.'
            );
        }

        if ($client->getProfile()->supportsCommand('MONITOR') === false) {
            throw new NotSupportedException("The current profile does not support 'MONITOR'.");
        }
    }

    /**
     * Initializes the consumer and sends the MONITOR command to the server.
     */
    protected function start()
    {
        $this->client->executeCommand(
            $this->client->createCommand('MONITOR')
        );
        $this->valid = true;
    }

    /**
     * Stops the consumer. Internally this is done by disconnecting from server
     * since there is no way to terminate the stream initialized by MONITOR.
     */
    public function stop()
    {
        $this->client->disconnect();
        $this->valid = false;
    }

    /**
     * {@inheritdoc}
     */
    public function rewind()
    {
        // NOOP
    }

    /**
     * Returns the last message payload retrieved from the server.
     *
     * @return object
     */
    public function current()
    {
        return $this->getValue();
    }

    /**
     * {@inheritdoc}
     */
    public function key()
    {
        return $this->position;
    }

    /**
     * {@inheritdoc}
     */
    public function next()
    {
        ++$this->position;
    }

    /**
     * Checks if the the consumer is still in a valid state to continue.
     *
     * @return bool
     */
    public function valid()
    {
        return $this->valid;
    }

    /**
     * Waits for a new message from the server generated by MONITOR and returns
     * it when available.
     *
     * @return object
     */
    private function getValue()
    {
        $database = 0;
        $client = null;
        $event = $this->client->getConnection()->read();

        $callback = function ($matches) use (&$database, &$client) {
            if (2 === $count = count($matches)) {
                // Redis <= 2.4
                $database = (int) $matches[1];
            }

            if (4 === $count) {
                // Redis >= 2.6
                $database = (int) $matches[2];
                $client = $matches[3];
            }

            return ' ';
        };

        $event = preg_replace_callback('/ \(db (\d+)\) | \[(\d+) (.*?)\] /', $callback, $event, 1);
        @list($timestamp, $command, $arguments) = explode(' ', $event, 3);

        return (object) array(
            'timestamp' => (float) $timestamp,
            'database' => $database,
            'client' => $client,
            'command' => substr($command, 1, -1),
            'arguments' => $arguments,
        );
    }
}


code:
<?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;

/**
 * Exception class thrown when trying to use features not supported by certain
 * classes or abstractions of Predis.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class NotSupportedException extends PredisException
{
}


code:
<?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->getProfile()->supportsCommands(array('multi', 'exec', 'discard'))) {
            throw new ClientException(
                "The current profile does not support 'MULTI', 'EXEC' and 'DISCARD'."
            );
        }

        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)
    {
        $profile = $this->getClient()->getProfile();
        $connection->executeCommand($profile->createCommand('multi'));

        foreach ($commands as $command) {
            $connection->writeRequest($command);
        }

        foreach ($commands as $command) {
            $response = $connection->readResponse($command);

            if ($response instanceof ErrorResponseInterface) {
                $connection->executeCommand($profile->createCommand('discard'));
                throw new ServerException($response->getMessage());
            }
        }

        $executed = $connection->executeCommand($profile->createCommand('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;
    }
}


code:
<?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\Aggregate\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->getConnection($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->getConnection($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;
    }
}


code:
<?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();
    }
}


code:
<?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\ClientContextInterface;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Command\CommandInterface;
use Predis\Connection\Aggregate\ReplicationInterface;
use Predis\Connection\ConnectionInterface;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ResponseInterface;
use Predis\Response\ServerException;

/**
 * Implementation of a command pipeline in which write and read operations of
 * Redis commands are pipelined to alleviate the effects of network round-trips.
 *
 * {@inheritdoc}
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class Pipeline implements ClientContextInterface
{
    private $client;
    private $pipeline;

    private $responses = array();
    private $running = false;

    /**
     * @param ClientInterface $client Client instance used by the context.
     */
    public function __construct(ClientInterface $client)
    {
        $this->client = $client;
        $this->pipeline = new \SplQueue();
    }

    /**
     * Queues a command into the pipeline buffer.
     *
     * @param string $method    Command ID.
     * @param array  $arguments Arguments for the command.
     *
     * @return $this
     */
    public function __call($method, $arguments)
    {
        $command = $this->client->createCommand($method, $arguments);
        $this->recordCommand($command);

        return $this;
    }

    /**
     * Queues a command instance into the pipeline buffer.
     *
     * @param CommandInterface $command Command to be queued in the buffer.
     */
    protected function recordCommand(CommandInterface $command)
    {
        $this->pipeline->enqueue($command);
    }

    /**
     * Queues a command instance into the pipeline buffer.
     *
     * @param CommandInterface $command Command instance to be queued in the buffer.
     *
     * @return $this
     */
    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);
    }

    /**
     * Returns the underlying connection to be used by the pipeline.
     *
     * @return ConnectionInterface
     */
    protected function getConnection()
    {
        $connection = $this->getClient()->getConnection();

        if ($connection instanceof ReplicationInterface) {
            $connection->switchTo('master');
        }

        return $connection;
    }

    /**
     * Implements the logic to flush the queued commands and read the responses
     * from the current connection.
     *
     * @param ConnectionInterface $connection Current connection instance.
     * @param \SplQueue           $commands   Queued commands.
     *
     * @return array
     */
    protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
    {
        foreach ($commands as $command) {
            $connection->writeRequest($command);
        }

        $responses = array();
        $exceptions = $this->throwServerExceptions();

        while (!$commands->isEmpty()) {
            $command = $commands->dequeue();
            $response = $connection->readResponse($command);

            if (!$response instanceof ResponseInterface) {
                $responses[] = $command->parseResponse($response);
            } elseif ($response instanceof ErrorResponseInterface && $exceptions) {
                $this->exception($connection, $response);
            } else {
                $responses[] = $response;
            }
        }

        return $responses;
    }

    /**
     * Flushes the buffer holding all of the commands queued so far.
     *
     * @param bool $send Specifies if the commands in the buffer should be sent to Redis.
     *
     * @return $this
     */
    public function flushPipeline($send = true)
    {
        if ($send && !$this->pipeline->isEmpty()) {
            $responses = $this->executePipeline($this->getConnection(), $this->pipeline);
            $this->responses = array_merge($this->responses, $responses);
        } else {
            $this->pipeline = new \SplQueue();
        }

        return $this;
    }

    /**
     * Marks the running status of the pipeline.
     *
     * @param bool $bool Sets the running status of the pipeline.
     *
     * @throws ClientException
     */
    private function setRunning($bool)
    {
        if ($bool && $this->running) {
            throw new ClientException('The current pipeline context is already being executed.');
        }

        $this->running = $bool;
    }

    /**
     * Handles the actual execution of the whole pipeline.
     *
     * @param mixed $callable Optional callback for execution.
     *
     * @throws \Exception
     * @throws \InvalidArgumentException
     *
     * @return array
     */
    public function execute($callable = null)
    {
        if ($callable && !is_callable($callable)) {
            throw new \InvalidArgumentException('The argument must be a callable object.');
        }

        $exception = null;
        $this->setRunning(true);

        try {
            if ($callable) {
                call_user_func($callable, $this);
            }

            $this->flushPipeline();
        } catch (\Exception $exception) {
            // NOOP
        }

        $this->setRunning(false);

        if ($exception) {
            throw $exception;
        }

        return $this->responses;
    }

    /**
     * Returns if the pipeline should throw exceptions on server errors.
     *
     * @return bool
     */
    protected function throwServerExceptions()
    {
        return (bool) $this->client->getOptions()->exceptions;
    }

    /**
     * Returns the underlying client instance used by the pipeline object.
     *
     * @return ClientInterface
     */
    public function getClient()
    {
        return $this->client;
    }
}


code:
<?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;

/**
 * Base exception class for Predis-related errors.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
abstract class PredisException extends \Exception
{
}


code:
<?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\Profile;

use Predis\ClientException;

/**
 * Factory class for creating profile instances from strings.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
final class Factory
{
    private static $profiles = array(
        '2.0' => 'Predis\Profile\RedisVersion200',
        '2.2' => 'Predis\Profile\RedisVersion220',
        '2.4' => 'Predis\Profile\RedisVersion240',
        '2.6' => 'Predis\Profile\RedisVersion260',
        '2.8' => 'Predis\Profile\RedisVersion280',
        '3.0' => 'Predis\Profile\RedisVersion300',
        '3.2' => 'Predis\Profile\RedisVersion320',
        'dev' => 'Predis\Profile\RedisUnstable',
        'default' => 'Predis\Profile\RedisVersion320',
    );

    /**
     *
     */
    private function __construct()
    {
        // NOOP
    }

    /**
     * Returns the default server profile.
     *
     * @return ProfileInterface
     */
    public static function getDefault()
    {
        return self::get('default');
    }

    /**
     * Returns the development server profile.
     *
     * @return ProfileInterface
     */
    public static function getDevelopment()
    {
        return self::get('dev');
    }

    /**
     * Registers a new server profile.
     *
     * @param string $alias Profile version or alias.
     * @param string $class FQN of a class implementing Predis\Profile\ProfileInterface.
     *
     * @throws \InvalidArgumentException
     */
    public static function define($alias, $class)
    {
        $reflection = new \ReflectionClass($class);

        if (!$reflection->isSubclassOf('Predis\Profile\ProfileInterface')) {
            throw new \InvalidArgumentException("The class '$class' is not a valid profile class.");
        }

        self::$profiles[$alias] = $class;
    }

    /**
     * Returns the specified server profile.
     *
     * @param string $version Profile version or alias.
     *
     * @throws ClientException
     *
     * @return ProfileInterface
     */
    public static function get($version)
    {
        if (!isset(self::$profiles[$version])) {
            throw new ClientException("Unknown server profile: '$version'.");
        }

        $profile = self::$profiles[$version];

        return new $profile();
    }
}


code:
<?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\Profile;

use Predis\Command\CommandInterface;

/**
 * A profile defines all the features and commands supported by certain versions
 * of Redis. Instances of Predis\Client should use a server profile matching the
 * version of Redis being used.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
interface ProfileInterface
{
    /**
     * Returns the profile version corresponding to the Redis version.
     *
     * @return string
     */
    public function getVersion();

    /**
     * Checks if the profile supports the specified command.
     *
     * @param string $commandID Command ID.
     *
     * @return bool
     */
    public function supportsCommand($commandID);

    /**
     * Checks if the profile supports the specified list of commands.
     *
     * @param array $commandIDs List of command IDs.
     *
     * @return string
     */
    public function supportsCommands(array $commandIDs);

    /**
     * Creates a new command instance.
     *
     * @param string $commandID Command ID.
     * @param array  $arguments Arguments for the command.
     *
     * @return CommandInterface
     */
    public function createCommand($commandID, array $arguments = array());
}


code:
<?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\Profile;

use Predis\ClientException;
use Predis\Command\Processor\ProcessorInterface;

/**
 * Base class implementing common functionalities for Redis server profiles.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
abstract class RedisProfile implements ProfileInterface
{
    private $commands;
    private $processor;

    /**
     *
     */
    public function __construct()
    {
        $this->commands = $this->getSupportedCommands();
    }

    /**
     * Returns a map of all the commands supported by the profile and their
     * actual PHP classes.
     *
     * @return array
     */
    abstract protected function getSupportedCommands();

    /**
     * {@inheritdoc}
     */
    public function supportsCommand($commandID)
    {
        return isset($this->commands[strtoupper($commandID)]);
    }

    /**
     * {@inheritdoc}
     */
    public function supportsCommands(array $commandIDs)
    {
        foreach ($commandIDs as $commandID) {
            if (!$this->supportsCommand($commandID)) {
                return false;
            }
        }

        return true;
    }

    /**
     * Returns the fully-qualified name of a class representing the specified
     * command ID registered in the current server profile.
     *
     * @param string $commandID Command ID.
     *
     * @return string|null
     */
    public function getCommandClass($commandID)
    {
        if (isset($this->commands[$commandID = strtoupper($commandID)])) {
            return $this->commands[$commandID];
        }
    }

    /**
     * {@inheritdoc}
     */
    public function createCommand($commandID, array $arguments = array())
    {
        $commandID = strtoupper($commandID);

        if (!isset($this->commands[$commandID])) {
            throw new ClientException("Command '$commandID' is not a registered Redis command.");
        }

        $commandClass = $this->commands[$commandID];
        $command = new $commandClass();
        $command->setArguments($arguments);

        if (isset($this->processor)) {
            $this->processor->process($command);
        }

        return $command;
    }

    /**
     * Defines a new command in the server profile.
     *
     * @param string $commandID Command ID.
     * @param string $class     Fully-qualified name of a Predis\Command\CommandInterface.
     *
     * @throws \InvalidArgumentException
     */
    public function defineCommand($commandID, $class)
    {
        $reflection = new \ReflectionClass($class);

        if (!$reflection->isSubclassOf('Predis\Command\CommandInterface')) {
            throw new \InvalidArgumentException("The class '$class' is not a valid command class.");
        }

        $this->commands[strtoupper($commandID)] = $class;
    }

    /**
     * {@inheritdoc}
     */
    public function setProcessor(ProcessorInterface $processor = null)
    {
        $this->processor = $processor;
    }

    /**
     * {@inheritdoc}
     */
    public function getProcessor()
    {
        return $this->processor;
    }

    /**
     * Returns the version of server profile as its string representation.
     *
     * @return string
     */
    public function __toString()
    {
        return $this->getVersion();
    }
}


code:
<?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\Profile;

/**
 * Server profile for the current unstable version of Redis.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class RedisUnstable extends RedisVersion320
{
    /**
     * {@inheritdoc}
     */
    public function getVersion()
    {
        return '3.2';
    }

    /**
     * {@inheritdoc}
     */
    public function getSupportedCommands()
    {
        return array_merge(parent::getSupportedCommands(), array(
            // EMPTY
        ));
    }
}


code:
<?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\Profile;

/**
 * Server profile for Redis 2.0.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class RedisVersion200 extends RedisProfile
{
    /**
     * {@inheritdoc}
     */
    public function getVersion()
    {
        return '2.0';
    }

    /**
     * {@inheritdoc}
     */
    public function getSupportedCommands()
    {
        return array(
            /* ---------------- Redis 1.2 ---------------- */

            /* commands operating on the key space */
            'EXISTS' => 'Predis\Command\KeyExists',
            'DEL' => 'Predis\Command\KeyDelete',
            'TYPE' => 'Predis\Command\KeyType',
            'KEYS' => 'Predis\Command\KeyKeys',
            'RANDOMKEY' => 'Predis\Command\KeyRandom',
            'RENAME' => 'Predis\Command\KeyRename',
            'RENAMENX' => 'Predis\Command\KeyRenamePreserve',
            'EXPIRE' => 'Predis\Command\KeyExpire',
            'EXPIREAT' => 'Predis\Command\KeyExpireAt',
            'TTL' => 'Predis\Command\KeyTimeToLive',
            'MOVE' => 'Predis\Command\KeyMove',
            'SORT' => 'Predis\Command\KeySort',

            /* commands operating on string values */
            'SET' => 'Predis\Command\StringSet',
            'SETNX' => 'Predis\Command\StringSetPreserve',
            'MSET' => 'Predis\Command\StringSetMultiple',
            'MSETNX' => 'Predis\Command\StringSetMultiplePreserve',
            'GET' => 'Predis\Command\StringGet',
            'MGET' => 'Predis\Command\StringGetMultiple',
            'GETSET' => 'Predis\Command\StringGetSet',
            'INCR' => 'Predis\Command\StringIncrement',
            'INCRBY' => 'Predis\Command\StringIncrementBy',
            'DECR' => 'Predis\Command\StringDecrement',
            'DECRBY' => 'Predis\Command\StringDecrementBy',

            /* commands operating on lists */
            'RPUSH' => 'Predis\Command\ListPushTail',
            'LPUSH' => 'Predis\Command\ListPushHead',
            'LLEN' => 'Predis\Command\ListLength',
            'LRANGE' => 'Predis\Command\ListRange',
            'LTRIM' => 'Predis\Command\ListTrim',
            'LINDEX' => 'Predis\Command\ListIndex',
            'LSET' => 'Predis\Command\ListSet',
            'LREM' => 'Predis\Command\ListRemove',
            'LPOP' => 'Predis\Command\ListPopFirst',
            'RPOP' => 'Predis\Command\ListPopLast',
            'RPOPLPUSH' => 'Predis\Command\ListPopLastPushHead',

            /* commands operating on sets */
            'SADD' => 'Predis\Command\SetAdd',
            'SREM' => 'Predis\Command\SetRemove',
            'SPOP' => 'Predis\Command\SetPop',
            'SMOVE' => 'Predis\Command\SetMove',
            'SCARD' => 'Predis\Command\SetCardinality',
            'SISMEMBER' => 'Predis\Command\SetIsMember',
            'SINTER' => 'Predis\Command\SetIntersection',
            'SINTERSTORE' => 'Predis\Command\SetIntersectionStore',
            'SUNION' => 'Predis\Command\SetUnion',
            'SUNIONSTORE' => 'Predis\Command\SetUnionStore',
            'SDIFF' => 'Predis\Command\SetDifference',
            'SDIFFSTORE' => 'Predis\Command\SetDifferenceStore',
            'SMEMBERS' => 'Predis\Command\SetMembers',
            'SRANDMEMBER' => 'Predis\Command\SetRandomMember',

            /* commands operating on sorted sets */
            'ZADD' => 'Predis\Command\ZSetAdd',
            'ZINCRBY' => 'Predis\Command\ZSetIncrementBy',
            'ZREM' => 'Predis\Command\ZSetRemove',
            'ZRANGE' => 'Predis\Command\ZSetRange',
            'ZREVRANGE' => 'Predis\Command\ZSetReverseRange',
            'ZRANGEBYSCORE' => 'Predis\Command\ZSetRangeByScore',
            'ZCARD' => 'Predis\Command\ZSetCardinality',
            'ZSCORE' => 'Predis\Command\ZSetScore',
            'ZREMRANGEBYSCORE' => 'Predis\Command\ZSetRemoveRangeByScore',

            /* connection related commands */
            'PING' => 'Predis\Command\ConnectionPing',
            'AUTH' => 'Predis\Command\ConnectionAuth',
            'SELECT' => 'Predis\Command\ConnectionSelect',
            'ECHO' => 'Predis\Command\ConnectionEcho',
            'QUIT' => 'Predis\Command\ConnectionQuit',

            /* remote server control commands */
            'INFO' => 'Predis\Command\ServerInfo',
            'SLAVEOF' => 'Predis\Command\ServerSlaveOf',
            'MONITOR' => 'Predis\Command\ServerMonitor',
            'DBSIZE' => 'Predis\Command\ServerDatabaseSize',
            'FLUSHDB' => 'Predis\Command\ServerFlushDatabase',
            'FLUSHALL' => 'Predis\Command\ServerFlushAll',
            'SAVE' => 'Predis\Command\ServerSave',
            'BGSAVE' => 'Predis\Command\ServerBackgroundSave',
            'LASTSAVE' => 'Predis\Command\ServerLastSave',
            'SHUTDOWN' => 'Predis\Command\ServerShutdown',
            'BGREWRITEAOF' => 'Predis\Command\ServerBackgroundRewriteAOF',

            /* ---------------- Redis 2.0 ---------------- */

            /* commands operating on string values */
            'SETEX' => 'Predis\Command\StringSetExpire',
            'APPEND' => 'Predis\Command\StringAppend',
            'SUBSTR' => 'Predis\Command\StringSubstr',

            /* commands operating on lists */
            'BLPOP' => 'Predis\Command\ListPopFirstBlocking',
            'BRPOP' => 'Predis\Command\ListPopLastBlocking',

            /* commands operating on sorted sets */
            'ZUNIONSTORE' => 'Predis\Command\ZSetUnionStore',
            'ZINTERSTORE' => 'Predis\Command\ZSetIntersectionStore',
            'ZCOUNT' => 'Predis\Command\ZSetCount',
            'ZRANK' => 'Predis\Command\ZSetRank',
            'ZREVRANK' => 'Predis\Command\ZSetReverseRank',
            'ZREMRANGEBYRANK' => 'Predis\Command\ZSetRemoveRangeByRank',

            /* commands operating on hashes */
            'HSET' => 'Predis\Command\HashSet',
            'HSETNX' => 'Predis\Command\HashSetPreserve',
            'HMSET' => 'Predis\Command\HashSetMultiple',
            'HINCRBY' => 'Predis\Command\HashIncrementBy',
            'HGET' => 'Predis\Command\HashGet',
            'HMGET' => 'Predis\Command\HashGetMultiple',
            'HDEL' => 'Predis\Command\HashDelete',
            'HEXISTS' => 'Predis\Command\HashExists',
            'HLEN' => 'Predis\Command\HashLength',
            'HKEYS' => 'Predis\Command\HashKeys',
            'HVALS' => 'Predis\Command\HashValues',
            'HGETALL' => 'Predis\Command\HashGetAll',

            /* transactions */
            'MULTI' => 'Predis\Command\TransactionMulti',
            'EXEC' => 'Predis\Command\TransactionExec',
            'DISCARD' => 'Predis\Command\TransactionDiscard',

            /* publish - subscribe */
            'SUBSCRIBE' => 'Predis\Command\PubSubSubscribe',
            'UNSUBSCRIBE' => 'Predis\Command\PubSubUnsubscribe',
            'PSUBSCRIBE' => 'Predis\Command\PubSubSubscribeByPattern',
            'PUNSUBSCRIBE' => 'Predis\Command\PubSubUnsubscribeByPattern',
            'PUBLISH' => 'Predis\Command\PubSubPublish',

            /* remote server control commands */
            'CONFIG' => 'Predis\Command\ServerConfig',
        );
    }
}


code:
<?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\Profile;

/**
 * Server profile for Redis 2.2.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class RedisVersion220 extends RedisProfile
{
    /**
     * {@inheritdoc}
     */
    public function getVersion()
    {
        return '2.2';
    }

    /**
     * {@inheritdoc}
     */
    public function getSupportedCommands()
    {
        return array(
            /* ---------------- Redis 1.2 ---------------- */

            /* commands operating on the key space */
            'EXISTS' => 'Predis\Command\KeyExists',
            'DEL' => 'Predis\Command\KeyDelete',
            'TYPE' => 'Predis\Command\KeyType',
            'KEYS' => 'Predis\Command\KeyKeys',
            'RANDOMKEY' => 'Predis\Command\KeyRandom',
            'RENAME' => 'Predis\Command\KeyRename',
            'RENAMENX' => 'Predis\Command\KeyRenamePreserve',
            'EXPIRE' => 'Predis\Command\KeyExpire',
            'EXPIREAT' => 'Predis\Command\KeyExpireAt',
            'TTL' => 'Predis\Command\KeyTimeToLive',
            'MOVE' => 'Predis\Command\KeyMove',
            'SORT' => 'Predis\Command\KeySort',

            /* commands operating on string values */
            'SET' => 'Predis\Command\StringSet',
            'SETNX' => 'Predis\Command\StringSetPreserve',
            'MSET' => 'Predis\Command\StringSetMultiple',
            'MSETNX' => 'Predis\Command\StringSetMultiplePreserve',
            'GET' => 'Predis\Command\StringGet',
            'MGET' => 'Predis\Command\StringGetMultiple',
            'GETSET' => 'Predis\Command\StringGetSet',
            'INCR' => 'Predis\Command\StringIncrement',
            'INCRBY' => 'Predis\Command\StringIncrementBy',
            'DECR' => 'Predis\Command\StringDecrement',
            'DECRBY' => 'Predis\Command\StringDecrementBy',

            /* commands operating on lists */
            'RPUSH' => 'Predis\Command\ListPushTail',
            'LPUSH' => 'Predis\Command\ListPushHead',
            'LLEN' => 'Predis\Command\ListLength',
            'LRANGE' => 'Predis\Command\ListRange',
            'LTRIM' => 'Predis\Command\ListTrim',
            'LINDEX' => 'Predis\Command\ListIndex',
            'LSET' => 'Predis\Command\ListSet',
            'LREM' => 'Predis\Command\ListRemove',
            'LPOP' => 'Predis\Command\ListPopFirst',
            'RPOP' => 'Predis\Command\ListPopLast',
            'RPOPLPUSH' => 'Predis\Command\ListPopLastPushHead',

            /* commands operating on sets */
            'SADD' => 'Predis\Command\SetAdd',
            'SREM' => 'Predis\Command\SetRemove',
            'SPOP' => 'Predis\Command\SetPop',
            'SMOVE' => 'Predis\Command\SetMove',
            'SCARD' => 'Predis\Command\SetCardinality',
            'SISMEMBER' => 'Predis\Command\SetIsMember',
            'SINTER' => 'Predis\Command\SetIntersection',
            'SINTERSTORE' => 'Predis\Command\SetIntersectionStore',
            'SUNION' => 'Predis\Command\SetUnion',
            'SUNIONSTORE' => 'Predis\Command\SetUnionStore',
            'SDIFF' => 'Predis\Command\SetDifference',
            'SDIFFSTORE' => 'Predis\Command\SetDifferenceStore',
            'SMEMBERS' => 'Predis\Command\SetMembers',
            'SRANDMEMBER' => 'Predis\Command\SetRandomMember',

            /* commands operating on sorted sets */
            'ZADD' => 'Predis\Command\ZSetAdd',
            'ZINCRBY' => 'Predis\Command\ZSetIncrementBy',
            'ZREM' => 'Predis\Command\ZSetRemove',
            'ZRANGE' => 'Predis\Command\ZSetRange',
            'ZREVRANGE' => 'Predis\Command\ZSetReverseRange',
            'ZRANGEBYSCORE' => 'Predis\Command\ZSetRangeByScore',
            'ZCARD' => 'Predis\Command\ZSetCardinality',
            'ZSCORE' => 'Predis\Command\ZSetScore',
            'ZREMRANGEBYSCORE' => 'Predis\Command\ZSetRemoveRangeByScore',

            /* connection related commands */
            'PING' => 'Predis\Command\ConnectionPing',
            'AUTH' => 'Predis\Command\ConnectionAuth',
            'SELECT' => 'Predis\Command\ConnectionSelect',
            'ECHO' => 'Predis\Command\ConnectionEcho',
            'QUIT' => 'Predis\Command\ConnectionQuit',

            /* remote server control commands */
            'INFO' => 'Predis\Command\ServerInfo',
            'SLAVEOF' => 'Predis\Command\ServerSlaveOf',
            'MONITOR' => 'Predis\Command\ServerMonitor',
            'DBSIZE' => 'Predis\Command\ServerDatabaseSize',
            'FLUSHDB' => 'Predis\Command\ServerFlushDatabase',
            'FLUSHALL' => 'Predis\Command\ServerFlushAll',
            'SAVE' => 'Predis\Command\ServerSave',
            'BGSAVE' => 'Predis\Command\ServerBackgroundSave',
            'LASTSAVE' => 'Predis\Command\ServerLastSave',
            'SHUTDOWN' => 'Predis\Command\ServerShutdown',
            'BGREWRITEAOF' => 'Predis\Command\ServerBackgroundRewriteAOF',

            /* ---------------- Redis 2.0 ---------------- */

            /* commands operating on string values */
            'SETEX' => 'Predis\Command\StringSetExpire',
            'APPEND' => 'Predis\Command\StringAppend',
            'SUBSTR' => 'Predis\Command\StringSubstr',

            /* commands operating on lists */
            'BLPOP' => 'Predis\Command\ListPopFirstBlocking',
            'BRPOP' => 'Predis\Command\ListPopLastBlocking',

            /* commands operating on sorted sets */
            'ZUNIONSTORE' => 'Predis\Command\ZSetUnionStore',
            'ZINTERSTORE' => 'Predis\Command\ZSetIntersectionStore',
            'ZCOUNT' => 'Predis\Command\ZSetCount',
            'ZRANK' => 'Predis\Command\ZSetRank',
            'ZREVRANK' => 'Predis\Command\ZSetReverseRank',
            'ZREMRANGEBYRANK' => 'Predis\Command\ZSetRemoveRangeByRank',

            /* commands operating on hashes */
            'HSET' => 'Predis\Command\HashSet',
            'HSETNX' => 'Predis\Command\HashSetPreserve',
            'HMSET' => 'Predis\Command\HashSetMultiple',
            'HINCRBY' => 'Predis\Command\HashIncrementBy',
            'HGET' => 'Predis\Command\HashGet',
            'HMGET' => 'Predis\Command\HashGetMultiple',
            'HDEL' => 'Predis\Command\HashDelete',
            'HEXISTS' => 'Predis\Command\HashExists',
            'HLEN' => 'Predis\Command\HashLength',
            'HKEYS' => 'Predis\Command\HashKeys',
            'HVALS' => 'Predis\Command\HashValues',
            'HGETALL' => 'Predis\Command\HashGetAll',

            /* transactions */
            'MULTI' => 'Predis\Command\TransactionMulti',
            'EXEC' => 'Predis\Command\TransactionExec',
            'DISCARD' => 'Predis\Command\TransactionDiscard',

            /* publish - subscribe */
            'SUBSCRIBE' => 'Predis\Command\PubSubSubscribe',
            'UNSUBSCRIBE' => 'Predis\Command\PubSubUnsubscribe',
            'PSUBSCRIBE' => 'Predis\Command\PubSubSubscribeByPattern',
            'PUNSUBSCRIBE' => 'Predis\Command\PubSubUnsubscribeByPattern',
            'PUBLISH' => 'Predis\Command\PubSubPublish',

            /* remote server control commands */
            'CONFIG' => 'Predis\Command\ServerConfig',

            /* ---------------- Redis 2.2 ---------------- */

            /* commands operating on the key space */
            'PERSIST' => 'Predis\Command\KeyPersist',

            /* commands operating on string values */
            'STRLEN' => 'Predis\Command\StringStrlen',
            'SETRANGE' => 'Predis\Command\StringSetRange',
            'GETRANGE' => 'Predis\Command\StringGetRange',
            'SETBIT' => 'Predis\Command\StringSetBit',
            'GETBIT' => 'Predis\Command\StringGetBit',

            /* commands operating on lists */
            'RPUSHX' => 'Predis\Command\ListPushTailX',
            'LPUSHX' => 'Predis\Command\ListPushHeadX',
            'LINSERT' => 'Predis\Command\ListInsert',
            'BRPOPLPUSH' => 'Predis\Command\ListPopLastPushHeadBlocking',

            /* commands operating on sorted sets */
            'ZREVRANGEBYSCORE' => 'Predis\Command\ZSetReverseRangeByScore',

            /* transactions */
            'WATCH' => 'Predis\Command\TransactionWatch',
            'UNWATCH' => 'Predis\Command\TransactionUnwatch',

            /* remote server control commands */
            'OBJECT' => 'Predis\Command\ServerObject',
            'SLOWLOG' => 'Predis\Command\ServerSlowlog',
        );
    }
}


code:
<?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\Profile;

/**
 * Server profile for Redis 2.4.
 *
 * @author Daniele Alessandri <suppakilla@gmail.com>
 */
class RedisVersion240 extends RedisProfile
{
    /**
     * {@inheritdoc}
     */
    public function getVersion()
    {
        return '2.4';
    }

    /**
     * {@inheritdoc}
     */
    public function getSupportedCommands()
    {
        return array(
            /* ---------------- Redis 1.2 ---------------- */

            /* commands operating on the key space */
            'EXISTS' => 'Predis\Command\KeyExists',
            'DEL' => 'Predis\Command\KeyDelete',
            'TYPE' => 'Predis\Command\KeyType',
            'KEYS' => 'Predis\Command\KeyKeys',
            'RANDOMKEY' => 'Predis\Command\KeyRandom',
            'RENAME' => 'Predis\Command\KeyRename',
            'RENAMENX' => 'Predis\Command\KeyRenamePreserve',
            'EXPIRE' => 'Predis\Command\KeyExpire',
            'EXPIREAT' => 'Predis\Command\KeyExpireAt',
            'TTL' => 'Predis\Command\KeyTimeToLive',
            'MOVE' => 'Predis\Command\KeyMove',
            'SORT' => 'Predis\Command\KeySort',

            /* commands operating on string values */
            'SET' => 'Predis\Command\StringSet',
            'SETNX' => 'Predis\Command\StringSetPreserve',
            'MSET' => 'Predis\Command\StringSetMultiple',
            'MSETNX' => 'Predis\Command\StringSetMultiplePreserve',
            'GET' => 'Predis\Command\StringGet',
            'MGET' => 'Predis\Command\StringGetMultiple',
            'GETSET' => 'Predis\Command\StringGetSet',
            'INCR' => 'Predis\Command\StringIncrement',
            'INCRBY' => 'Predis\Command\StringIncrementBy',
            'DECR' => 'Predis\Command\StringDecrement',
            'DECRBY' => 'Predis\Command\StringDecrementBy',

            /* commands operating on lists */
            'RPUSH' => 'Predis\Command\ListPushTail',
            'LPUSH' => 'Predis\Command\ListPushHead',
            'LLEN' => 'Predis\Command\ListLength',
            'LRANGE' => 'Predis\Command\ListRange',
            'LTRIM' => 'Predis\Command\ListTrim',
            'LINDEX' => 'Predis\Command\ListIndex',
            'LSET' => 'Predis\Command\ListSet',
            'LREM' => 'Predis\Command\ListRemove',
            'LPOP' => 'Predis\Command\ListPopFirst',
            'RPOP' => 'Predis\Command\ListPopLast',
            'RPOPLPUSH' => 'Predis\Command\ListPopLastPushHead',

            /* commands operating on sets */
            'SADD' => 'Predis\Command\SetAdd',
            'SREM' => 'Predis\Command\SetRemove',
            'SPOP' => 'Predis\Command\SetPop',
            'SMOVE' => 'Predis\Command\SetMove',
            'SCARD' => 'Predis\Command\SetCardinality',
            'SISMEMBER' => 'Predis\Command\SetIsMember',
            'SINTER' => 'Predis\Command\SetIntersection',
            'SINTERSTORE' => 'Predis\Command\SetIntersectionStore',
            'SUNION' => 'Predis\Command\SetUnion',
            'SUNIONSTORE' => 'Predis\Command\SetUnionStore',
            'SDIFF' => 'Predis\Command\SetDifference',
            'SDIFFSTORE' => 'Predis\Command\SetDifferenceStore',
            'SMEMBERS' => 'Predis\Command\SetMembers',
            'SRANDMEMBER' => 'Predis\Command\SetRandomMember',

            /* commands operating on sorted sets */
            'ZADD' => 'Predis\Command\ZSetAdd',
            'ZINCRBY' => 'Predis\Command\ZSetIncrementBy',
            'ZREM' => 'Predis\Command\ZSetRemove',
            'ZRANGE' => 'Predis\Command\ZSetRange',
            'ZREVRANGE' => 'Predis\Command\ZSetReverseRange',
            'ZRANGEBYSCORE' => 'Predis\Command\ZSetRangeByScore',
            'ZCARD' => 'Predis\Command\ZSetCardinality',
            'ZSCORE' => 'Predis\Command\ZSetScore',
            'ZREMRANGEBYSCORE' => 'Predis\Command\ZSetRemoveRangeByScore',

            /* connection related commands */
            'PING' => 'Predis\Command\ConnectionPing',
            'AUTH' => 'Predis\Command\ConnectionAuth',
            'SELECT' => 'Predis\Command\ConnectionSelect',
            'ECHO' => 'Predis\Command\ConnectionEcho',
            'QUIT' => 'Predis\Command\ConnectionQuit',

            /* remote server control commands */
            'INFO' => 'Predis\Command\ServerInfo',
            'SLAVEOF' => 'Predis\Command\ServerSlaveOf',
            'MONITOR' => 'Predis\Command\ServerMonitor',
            'DBSIZE' => 'Predis\Command\ServerDatabaseSize',
            'FLUSHDB' => 'Predis\Command\ServerFlushDatabase',
            'FLUSHALL' => 'Predis\Command\ServerFlushAll',
            'SAVE' => 'Predis\Command\ServerSave',
            'BGSAVE' => 'Predis\Command\ServerBackgroundSave',
            'LASTSAVE' => 'Predis\Command\ServerLastSave',
            'SHUTDOWN' => 'Predis\Command\ServerShutdown',
            'BGREWRITEAOF' => 'Predis\Command\ServerBackgroundRewriteAOF',

            /* ---------------- Redis 2.0 ---------------- */

            /* commands operating on string values */
            'SETEX' => 'Predis\Command\StringSetExpire',
            'APPEND' => 'Predis\Command\StringAppend',
            'SUBSTR' => 'Predis\Command\StringSubstr',

            /* commands operating on lists */
            'BLPOP' => 'Predis\Command\ListPopFirstBlocking',
            'BRPOP' => 'Predis\Command\ListPopLastBlocking',

            /* commands operating on sorted sets */
            'ZUNIONSTORE' => 'Predis\Command\ZSetUnionStore',
            'ZINTERSTORE' => 'Predis\Command\ZSetIntersectionStore',
            'ZCOUNT' => 'Predis\Command\ZSetCount',
            'ZRANK' => 'Predis\Command\ZSetRank',
            'ZREVRANK' => 'Predis\Command\ZSetReverseRank',
            'ZREMRANGEBYRANK' => 'Predis\Command\ZSetRemoveRangeByRank',

            /* commands operating on hashes */
            'HSET' => 'Predis\Command\HashSet',
            'HSETNX' => 'Predis\Command\HashSetPreserve',
            'HMSET' => 'Predis\Command\HashSetMultiple',
            'HINCRBY' => 'Predis\Command\HashIncrementBy',
            'HGET' => 'Predis\Command\HashGet',
            'HMGET' => 'Predis\Command\HashGetMultiple',
            'HDEL' => 'Predis\Command\HashDelete',
            'HEXISTS' => 'Predis\Command\HashExists',
            'HLEN' => 'Predis\Command\HashLength',
            'HKEYS' => 'Predis\Command\HashKeys',
            'HVALS' => 'Predis\Command\HashValues',
            'HGETALL' => 'Predis\Command\HashGetAll',

            /* transactions */
            'MULTI' => 'Predis\Command\TransactionMulti',
            'EXEC' => 'Predis\Command\TransactionExec',
            'DISCARD' => 'Predis\Command\TransactionDiscard',

            /* publish - subscribe */
            'SUBSCRIBE' => 'Predis\Command\PubSubSubscribe',
            'UNSUBSCRIBE' => 'Predis\Command\PubSubUnsubscribe',
            'PSUBSCRIBE' => 'Predis\Command\PubSubSubscribeByPattern',
            'PUNSUBSCRIBE' => 'Predis\Command\PubSubUnsubscribeByPattern',
            'PUBLISH' => 'Predis\Command\PubSubPublish',

            /* remote server control commands */
            'CONFIG' => 'Predis\Command\ServerConfig',

            /* ---------------- Redis 2.2 ---------------- */

            /* commands operating on the key space */
            'PERSIST' => 'Predis\Command\KeyPersist',

            /* commands operating on string values */
            'STRLEN' => 'Predis\Command\StringStrlen',
            'SETRANGE' => 'Predis\Command\StringSetRange',
            'GETRANGE' => 'Predis\Command\StringGetRange',
            'SETBIT' => 'Predis\Command\StringSetBit',
            'GETBIT' => 'Predis\Command\StringGetBit',

            /* commands operating on lists */
            'RPUSHX' => 'Predis\Command\ListPushTailX',
            'LPUSHX' => 'Predis\Command\ListPushHeadX',
            'LINSERT' => 'Predis\Command\ListInsert',
            'BRPOPLPUSH' => 'Predis\Command\ListPopLastPushHeadBlocking',

            /* commands operating on sorted sets */
            'ZREVRANGEBYSCORE' => 'Predis\Command\ZSetReverseRangeByScore',

            /* transactions */
            'WATCH' => 'Predis\Command\TransactionWatch',
            'UNWATCH' => 'Predis\Command\TransactionUnwatch',

            /* remote server control commands */
            'OBJECT' => 'Predis\Command\ServerObject',
            'SLOWLOG' => 'Predis\Command\ServerSlowlog',

            /* ---------------- Redis 2.4 ---------------- */

            /* remote server control commands */
            'CLIENT' => 'Predis\Command\ServerClient',
        );
    }
}