Bump VERSION, update CHANGELOG and release.

CHANGELOG

@@ -1,3 +1,25 @@
+v0.6.4 (2011-02-12)
+  * Various performance improvements (15% ~ 25%) especially when dealing with 
+    long multibulk replies or when using clustered connections.
+
+  * Added the "on_retry" option to Predis\MultiExecBlock that can be used to 
+    specify an external callback (or any callable object) that gets invoked 
+    whenever a transaction is aborted by the server.
+
+  * Added inline (p)subscribtion via options when initializing an instance of 
+    Predis\PubSubContext.
+
+v0.6.3 (2011-01-01)
+  * New commands available in the Redis v2.2 profile (dev):
+      - Strings: SETRANGE, GETRANGE, SETBIT, GETBIT
+      - Lists  : BRPOPLPUSH
+
+  * The abstraction for MULTI/EXEC transactions has been dramatically improved 
+    by providing support for check-and-set (CAS) operations when using Redis >= 
+    2.2. Aborted transactions can also be optionally replayed in automatic up 
+    to a user-defined number of times, after which a Predis\AbortedMultiExec 
+    exception is thrown.
+
 v0.6.2 (2010-11-28)
   * Minor internal improvements and clean ups.
 

LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2009-2010 Daniele Alessandri
+Copyright (c) 2009-2011 Daniele Alessandri
 
 Permission is hereby granted, free of charge, to any person
 obtaining a copy of this software and associated documentation

README.markdown

@@ -17,9 +17,10 @@ to be implemented soon in Predis.
 
 ## Main features ##
 
-- Full support for Redis 2.0. Different versions of Redis are supported via server profiles.
+- Full support for Redis 2.0 and 2.2. Different versions of Redis are supported via server profiles.
 - Client-side sharding (support for consistent hashing and custom distribution strategies).
 - Command pipelining on single and multiple connections (transparent).
+- Abstraction for Redis transactions (>= 2.0) with support for CAS operations (>= 2.2).
 - Lazy connections (connections to Redis instances are only established just in time).
 - Flexible system to define and register your own set of commands to a client instance.
 
@@ -60,10 +61,10 @@ Furthermore, a pipeline can be initialized on a cluster of redis instances in th
 same exact way they are created on single connection. Sharding is still transparent 
 to the user:
 
-    $redis = Predis\Client::create(
+    $redis = new Predis\Client(array(
         array('host' => '10.0.0.1', 'port' => 6379),
         array('host' => '10.0.0.2', 'port' => 6379)
-    );
+    ));
 
     $replies = $redis->pipeline(function($pipe) {
         for ($i = 0; $i < 1000; $i++) {

VERSION

@@ -1 +1 @@
-0.6.2
+0.6.4

examples/MultiExecTransactionsWithCAS.php

@@ -0,0 +1,41 @@
+<?php
+require_once 'SharedConfigurations.php';
+
+/*
+This is an implementation of an atomic client-side ZPOP using the support for 
+check-and-set (CAS) operations with MULTI/EXEC transactions, as described in 
+"WATCH explained" from http://redis.io/topics/transactions
+
+First, populate your database with a tiny sample data set:
+
+./redis-cli
+SELECT 15
+ZADD zset 1 a
+ZADD zset 2 b
+ZADD zset 3 c
+*/
+
+function zpop($client, $zsetKey) {
+    $element = null;
+    $options = array(
+        'cas'   => true,     // Initialize with support for CAS operations
+        'watch' => $zsetKey, // Key that needs to be WATCHed to detect changes
+        'retry' => 3,        // Number of retries on aborted transactions, after
+                             // which the client bails out with an exception.
+    );
+
+    $txReply = $client->multiExec($options, function($tx) 
+        use ($zsetKey, &$element) {
+        @list($element) = $tx->zrange($zsetKey, 0, 0);
+        if (isset($element)) {
+            $tx->multi();     // With CAS, MULTI *must* be explicitly invoked.
+            $tx->zrem($zsetKey, $element);
+        }
+    });
+    return $element;
+}
+
+$redis = new Predis\Client($single_server, 'dev');
+$zpopped = zpop($redis, 'zset');
+echo isset($zpopped) ? "ZPOPed $zpopped" : "Nothing to ZPOP!", "\n";
+?>

lib/Predis.php

@@ -287,8 +287,8 @@ class Client {
         return $transBlock !== null ? $multi->execute($transBlock) : $multi;
     }
 
-    public function pubSubContext() {
-        return new PubSubContext($this);
+    public function pubSubContext(Array $options = null) {
+        return new PubSubContext($this, $options);
     }
 }
 
@@ -417,7 +417,8 @@ class Protocol {
 }
 
 abstract class Command {
-    private $_arguments, $_hash;
+    private $_hash;
+    private $_arguments = array();
 
     public abstract function getCommandId();
 
@@ -431,22 +432,24 @@ abstract class Command {
         if (isset($this->_hash)) {
             return $this->_hash;
         }
-        else {
-            if (isset($this->_arguments[0])) {
-                // TODO: should we throw an exception if the command does 
-                //       not support sharding?
-                $key = $this->_arguments[0];
 
-                $start = strpos($key, '{');
-                $end   = strpos($key, '}');
-                if ($start !== false && $end !== false) {
+        if (isset($this->_arguments[0])) {
+            // TODO: should we throw an exception if the command does 
+            //       not support sharding?
+            $key = $this->_arguments[0];
+
+            $start = strpos($key, '{');
+            if ($start !== false) {
+                $end = strpos($key, '}', $start);
+                if ($end !== false) {
                     $key = substr($key, ++$start, $end - $start);
                 }
-
-                $this->_hash = $distributor->generateKey($key);
-                return $this->_hash;
             }
+
+            $this->_hash = $distributor->generateKey($key);
+            return $this->_hash;
         }
+
         return null;
     }
 
@@ -469,11 +472,13 @@ abstract class Command {
     }
 
     public function getArguments() {
-        return isset($this->_arguments) ? $this->_arguments : array();
+        return $this->_arguments;
     }
 
     public function getArgument($index = 0) {
-        return isset($this->_arguments[$index]) ? $this->_arguments[$index] : null;
+        if (isset($this->_arguments[$index]) === true) {
+            return $this->_arguments[$index];
+        }
     }
 
     public function parseResponse($data) {
@@ -491,8 +496,7 @@ abstract class InlineCommand extends Command {
             $arguments[0] = implode($arguments[0], ' ');
         }
         return $command . (count($arguments) > 0
-            ? ' ' . implode($arguments, ' ') . Protocol::NEWLINE 
-            : Protocol::NEWLINE
+            ? ' ' . implode($arguments, ' ') . "\r\n" : "\r\n"
         );
     }
 }
@@ -504,7 +508,7 @@ abstract class BulkCommand extends Command {
             $data = implode($data, ' ');
         }
         return $command . ' ' . implode($arguments, ' ') . ' ' . strlen($data) . 
-            Protocol::NEWLINE . $data . Protocol::NEWLINE;
+            "\r\n" . $data . "\r\n";
     }
 }
 
@@ -521,14 +525,14 @@ abstract class MultiBulkCommand extends Command {
             $cmd_args = $arguments;
         }
 
-        $newline = Protocol::NEWLINE;
         $cmdlen  = strlen($command);
         $reqlen  = $argsc + 1;
 
-        $buffer = "*{$reqlen}{$newline}\${$cmdlen}{$newline}{$command}{$newline}";
-        foreach ($cmd_args as $argument) {
+        $buffer = "*{$reqlen}\r\n\${$cmdlen}\r\n{$command}\r\n";
+        for ($i = 0; $i < $reqlen - 1; $i++) {
+            $argument = $cmd_args[$i];
             $arglen  = strlen($argument);
-            $buffer .= "\${$arglen}{$newline}{$argument}{$newline}";
+            $buffer .= "\${$arglen}\r\n{$argument}\r\n";
         }
 
         return $buffer;
@@ -543,10 +547,10 @@ interface IResponseHandler {
 
 class ResponseStatusHandler implements IResponseHandler {
     public function handle(Connection $connection, $status) {
-        if ($status === Protocol::OK) {
+        if ($status === 'OK') {
             return true;
         }
-        else if ($status === Protocol::QUEUED) {
+        if ($status === 'QUEUED') {
             return new ResponseQueued();
         }
         return $status;
@@ -566,44 +570,31 @@ class ResponseErrorSilentHandler implements IResponseHandler {
 }
 
 class ResponseBulkHandler implements IResponseHandler {
-    public function handle(Connection $connection, $dataLength) {
-        if (!is_numeric($dataLength)) {
+    public function handle(Connection $connection, $lengthString) {
+        $length = (int) $lengthString;
+        if ($length != $lengthString) {
             Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                $connection, "Cannot parse '$dataLength' as data length"
+                $connection, "Cannot parse '$length' as data length"
             ));
         }
-
-        if ($dataLength > 0) {
-            $value = $connection->readBytes($dataLength);
-            self::discardNewLine($connection);
-            return $value;
+        if ($length >= 0) {
+            return $length > 0 ? substr($connection->readBytes($length + 2), 0, -2) : '';
         }
-        else if ($dataLength == 0) {
-            self::discardNewLine($connection);
-            return '';
-        }
-
-        return null;
-    }
-
-    private static function discardNewLine(Connection $connection) {
-        if ($connection->readBytes(2) !== Protocol::NEWLINE) {
-            Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                $connection, 'Did not receive a new-line at the end of a bulk response'
-            ));
+        if ($length == -1) {
+            return null;
         }
     }
 }
 
 class ResponseMultiBulkHandler implements IResponseHandler {
-    public function handle(Connection $connection, $rawLength) {
-        if (!is_numeric($rawLength)) {
+    public function handle(Connection $connection, $lengthString) {
+        $listLength = (int) $lengthString;
+        if ($listLength != $lengthString) {
             Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                $connection, "Cannot parse '$rawLength' as data length"
+                $connection, "Cannot parse '$lengthString' as data length"
             ));
         }
 
-        $listLength = (int) $rawLength;
         if ($listLength === -1) {
             return null;
         }
@@ -611,9 +602,19 @@ class ResponseMultiBulkHandler implements IResponseHandler {
         $list = array();
 
         if ($listLength > 0) {
+            $handlers = array();
             $reader = $connection->getResponseReader();
             for ($i = 0; $i < $listLength; $i++) {
-                $list[] = $reader->read($connection);
+                $header = $connection->readLine();
+                $prefix = $header[0];
+                if (isset($handlers[$prefix])) {
+                    $handler = $handlers[$prefix];
+                }
+                else {
+                    $handler = $reader->getHandler($prefix);
+                    $handlers[$prefix] = $handler;
+                }
+                $list[$i] = $handler->handle($connection, substr($header, 1));
             }
         }
 
@@ -622,13 +623,14 @@ class ResponseMultiBulkHandler implements IResponseHandler {
 }
 
 class ResponseMultiBulkStreamHandler implements IResponseHandler {
-    public function handle(Connection $connection, $rawLength) {
-        if (!is_numeric($rawLength)) {
+    public function handle(Connection $connection, $lengthString) {
+        $listLength = (int) $lengthString;
+        if ($listLength != $lengthString) {
             Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                $connection, "Cannot parse '$rawLength' as data length"
+                $connection, "Cannot parse '$lengthString' as data length"
             ));
         }
-        return new Shared\MultiBulkResponseIterator($connection, (int)$rawLength);
+        return new Shared\MultiBulkResponseIterator($connection, $lengthString);
     }
 }
 
@@ -637,14 +639,12 @@ class ResponseIntegerHandler implements IResponseHandler {
         if (is_numeric($number)) {
             return (int) $number;
         }
-        else {
-            if ($number !== Protocol::NULL) {
-                Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                    $connection, "Cannot parse '$number' as numeric response"
-                ));
-            }
-            return null;
+        if ($number !== 'nil') {
+            Shared\Utils::onCommunicationException(new MalformedServerResponse(
+                $connection, "Cannot parse '$number' as numeric response"
+            ));
         }
+        return null;
     }
 }
 
@@ -678,26 +678,26 @@ class ResponseReader {
     public function read(Connection $connection) {
         $header = $connection->readLine();
         if ($header === '') {
-            Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                $connection, 'Unexpected empty header'
-            ));
+            $this->throwMalformedResponse($connection, 'Unexpected empty header');
         }
 
         $prefix  = $header[0];
-        $payload = strlen($header) > 1 ? substr($header, 1) : '';
-
         if (!isset($this->_prefixHandlers[$prefix])) {
-            Shared\Utils::onCommunicationException(new MalformedServerResponse(
-                $connection, "Unknown prefix '$prefix'"
-            ));
+            $this->throwMalformedResponse($connection, "Unknown prefix '$prefix'");
         }
-
         $handler = $this->_prefixHandlers[$prefix];
-        return $handler->handle($connection, $payload);
+        return $handler->handle($connection, substr($header, 1));
+    }
+
+    private function throwMalformedResponse(Connection $connection, $message) {
+        Shared\Utils::onCommunicationException(new MalformedServerResponse(
+            $connection, $message
+        ));
     }
 }
 
 class ResponseError {
+    public $skipParse = true;
     private $_message;
 
     public function __construct($message) {
@@ -705,10 +705,10 @@ class ResponseError {
     }
 
     public function __get($property) {
-        if ($property == 'error') {
+        if ($property === 'error') {
             return true;
         }
-        if ($property == 'message') {
+        if ($property === 'message') {
             return $this->_message;
         }
     }
@@ -723,11 +723,21 @@ class ResponseError {
 }
 
 class ResponseQueued {
-    public $queued = true;
+    public $skipParse = true;
 
     public function __toString() {
         return Protocol::QUEUED;
     }
+
+    public function __get($property) {
+        if ($property === 'queued') {
+            return true;
+        }
+    }
+
+    public function __isset($property) {
+        return $property === 'queued';
+    }
 }
 
 /* ------------------------------------------------------------------------- */
@@ -805,18 +815,15 @@ class CommandPipeline {
 }
 
 class MultiExecBlock {
-    private $_initialized, $_discarded, $_insideBlock;
+    private $_initialized, $_discarded, $_insideBlock, $_checkAndSet;
     private $_redisClient, $_options, $_commands;
     private $_supportsWatch;
 
     public function __construct(Client $redisClient, Array $options = null) {
         $this->checkCapabilities($redisClient);
-        $this->_initialized = false;
-        $this->_discarded   = false;
-        $this->_insideBlock = false;
+        $this->_options = $options ?: array();
         $this->_redisClient = $redisClient;
-        $this->_options     = $options ?: array();
-        $this->_commands    = array();
+        $this->reset();
     }
 
     private function checkCapabilities(Client $redisClient) {
@@ -842,55 +849,66 @@ class MultiExecBlock {
         }
     }
 
-    private function initialize() {
-        if ($this->_initialized === false) {
-            if (isset($this->_options['watch'])) {
-                $this->watch($this->_options['watch']);
-            }
-            $this->_redisClient->multi();
-            $this->_initialized = true;
-            $this->_discarded   = false;
-        }
+    private function reset() {
+        $this->_initialized = false;
+        $this->_discarded   = false;
+        $this->_checkAndSet = false;
+        $this->_insideBlock = false;
+        $this->_commands    = array();
     }
 
-    private function setInsideBlock($value) {
-        $this->_insideBlock = $value;
+    private function initialize() {
+        if ($this->_initialized === true) {
+            return;
+        }
+        $options = &$this->_options;
+        $this->_checkAndSet = isset($options['cas']) && $options['cas'];
+        if (isset($options['watch'])) {
+            $this->watch($options['watch']);
+        }
+        if (!$this->_checkAndSet || ($this->_discarded && $this->_checkAndSet)) {
+            $this->_redisClient->multi();
+            if ($this->_discarded) {
+                $this->_checkAndSet = false;
+            }
+        }
+        $this->_initialized = true;
+        $this->_discarded   = false;
     }
 
     public function __call($method, $arguments) {
         $this->initialize();
-        $command  = $this->_redisClient->createCommand($method, $arguments);
-        $response = $this->_redisClient->executeCommand($command);
-        if (isset($response->queued)) {
-            $this->_commands[] = $command;
-            return $this;
+        $client = $this->_redisClient;
+        if ($this->_checkAndSet) {
+            return call_user_func_array(array($client, $method), $arguments);
         }
-        else {
-            $this->malformedServerResponse('The server did not respond with a QUEUED status reply');
+        $command  = $client->createCommand($method, $arguments);
+        $response = $client->executeCommand($command);
+        if (!$response instanceof \Predis\ResponseQueued) {
+            $this->malformedServerResponse(
+                'The server did not respond with a QUEUED status reply'
+            );
         }
+        $this->_commands[] = $command;
+        return $this;
     }
 
     public function watch($keys) {
         $this->isWatchSupported();
-        if ($this->_initialized === true) {
-            throw new \Predis\ClientException('WATCH inside MULTI is not allowed');
+        if ($this->_initialized && !$this->_checkAndSet) {
+            throw new ClientException('WATCH inside MULTI is not allowed');
         }
-
-        $reply = null;
-        if (is_array($keys)) {
-            $reply = array();
-            foreach ($keys as $key) {
-                $reply = $this->_redisClient->watch($keys);
-            }
-        }
-        else {
-            $reply = $this->_redisClient->watch($keys);
-        }
-        return $reply;
+        return $this->_redisClient->watch($keys);
     }
 
     public function multi() {
+        if ($this->_initialized && $this->_checkAndSet) {
+            $this->_checkAndSet = false;
+            $this->_redisClient->multi();
+            return $this;
+        }
         $this->initialize();
+        return $this;
     }
 
     public function unwatch() {
@@ -901,9 +919,8 @@ class MultiExecBlock {
 
     public function discard() {
         $this->_redisClient->discard();
-        $this->_commands    = array();
-        $this->_initialized = false;
-        $this->_discarded   = true;
+        $this->reset();
+        $this->_discarded = true;
         return $this;
     }
 
@@ -911,60 +928,92 @@ class MultiExecBlock {
         return $this->execute();
     }
 
-    public function execute($block = null) {
+    private function checkBeforeExecution($block) {
         if ($this->_insideBlock === true) {
             throw new \Predis\ClientException(
                 "Cannot invoke 'execute' or 'exec' inside an active client transaction block"
             );
         }
-
-        if ($block && !is_callable($block)) {
-            throw new \InvalidArgumentException('Argument passed must be a callable object');
+        if ($block) {
+            if (!is_callable($block)) {
+                throw new \InvalidArgumentException(
+                    'Argument passed must be a callable object'
+                );
+            }
+            if (count($this->_commands) > 0) {
+                throw new ClientException(
+                    'Cannot execute a transaction block after using fluent interface'
+                );
+            }
         }
+        if (isset($this->_options['retry']) && !isset($block)) {
+            $this->discard();
+            throw new \InvalidArgumentException(
+                'Automatic retries can be used only when a transaction block is provided'
+            );
+        }
+    }
 
-        $blockException = null;
-        $returnValues   = array();
+    public function execute($block = null) {
+        $this->checkBeforeExecution($block);
 
-        if ($block !== null) {
-            $this->setInsideBlock(true);
-            try {
-                $block($this);
-            }
-            catch (CommunicationException $exception) {
-                $blockException = $exception;
-            }
-            catch (ServerException $exception) {
-                $blockException = $exception;
-            }
-            catch (\Exception $exception) {
-                $blockException = $exception;
-                if ($this->_initialized === true) {
-                    $this->discard();
+        $reply = null;
+        $returnValues = array();
+        $attemptsLeft = isset($this->_options['retry']) ? (int)$this->_options['retry'] : 0;
+        do {
+            $blockException = null;
+            if ($block !== null) {
+                $this->_insideBlock = true;
+                try {
+                    $block($this);
+                }
+                catch (CommunicationException $exception) {
+                    $blockException = $exception;
+                }
+                catch (ServerException $exception) {
+                    $blockException = $exception;
+                }
+                catch (\Exception $exception) {
+                    $blockException = $exception;
+                    if ($this->_initialized === true) {
+                        $this->discard();
+                    }
+                }
+                $this->_insideBlock = false;
+                if ($blockException !== null) {
+                    throw $blockException;
                 }
             }
-            $this->setInsideBlock(false);
-            if ($blockException !== null) {
-                throw $blockException;
+
+            if ($this->_initialized === false || count($this->_commands) == 0) {
+                return;
             }
-        }
 
-        if ($this->_initialized === false) {
-            return;
-        }
-
-        $reply = $this->_redisClient->exec();
-        if ($reply === null) {
-            throw new AbortedMultiExec('The current transaction has been aborted by the server');
-        }
+            $reply = $this->_redisClient->exec();
+            if ($reply === null) {
+                if ($attemptsLeft === 0) {
+                    throw new AbortedMultiExec(
+                        'The current transaction has been aborted by the server'
+                    );
+                }
+                $this->reset();
+                if (isset($this->_options['on_retry']) && is_callable($this->_options['on_retry'])) {
+                    call_user_func($this->_options['on_retry'], $this, $attemptsLeft);
+                }
+                continue;
+            }
+            break;
+        } while ($attemptsLeft-- > 0);
 
         $execReply = $reply instanceof \Iterator ? iterator_to_array($reply) : $reply;
-        $commands  = &$this->_commands;
         $sizeofReplies = count($execReply);
 
+        $commands = &$this->_commands;
         if ($sizeofReplies !== count($commands)) {
-            $this->malformedServerResponse('Unexpected number of responses for a MultiExecBlock');
+            $this->malformedServerResponse(
+                'Unexpected number of responses for a MultiExecBlock'
+            );
         }
-
         for ($i = 0; $i < $sizeofReplies; $i++) {
             $returnValues[] = $commands[$i]->parseResponse($execReply[$i] instanceof \Iterator
                 ? iterator_to_array($execReply[$i])
@@ -977,9 +1026,9 @@ class MultiExecBlock {
     }
 
     private function malformedServerResponse($message) {
-        // NOTE: a MULTI/EXEC block cannot be initialized on a clustered 
-        //       connection, which means that Predis\Client::getConnection 
-        //       will always return an instance of Predis\Connection.
+        // Since a MULTI/EXEC block cannot be initialized over a clustered 
+        // connection, we can safely assume that Predis\Client::getConnection() 
+        // will always return an instance of Predis\Connection.
         Shared\Utils::onCommunicationException(new MalformedServerResponse(
             $this->_redisClient->getConnection(), $message
         ));
@@ -998,12 +1047,16 @@ class PubSubContext implements \Iterator {
     const STATUS_SUBSCRIBED  = 0x0010;
     const STATUS_PSUBSCRIBED = 0x0100;
 
-    private $_redisClient, $_subscriptions, $_isStillValid, $_position;
+    private $_redisClient, $_position, $_options;
 
-    public function __construct(Client $redisClient) {
+    public function __construct(Client $redisClient, Array $options = null) {
         $this->checkCapabilities($redisClient);
+        $this->_options = $options ?: array();
         $this->_redisClient = $redisClient;
         $this->_statusFlags = self::STATUS_VALID;
+
+        $this->genericSubscribeInit('subscribe');
+        $this->genericSubscribeInit('psubscribe');
     }
 
     public function __destruct() {
@@ -1025,6 +1078,19 @@ class PubSubContext implements \Iterator {
         }
     }
 
+    private function genericSubscribeInit($subscribeAction) {
+        if (isset($this->_options[$subscribeAction])) {
+            if (is_array($this->_options[$subscribeAction])) {
+                foreach ($this->_options[$subscribeAction] as $subscription) {
+                    $this->$subscribeAction($subscription);
+                }
+            }
+            else {
+                $this->$subscribeAction($this->_options[$subscribeAction]);
+            }
+        }
+    }
+
     private function isFlagSet($value) {
         return ($this->_statusFlags & $value) === $value;
     }
@@ -1306,8 +1372,7 @@ class Connection implements IConnection {
 
     public function readResponse(Command $command) {
         $response = $this->_reader->read($this);
-        $skipparse = isset($response->queued) || isset($response->error);
-        return $skipparse ? $response : $command->parseResponse($response);
+        return isset($response->skipParse) ? $response : $command->parseResponse($response);
     }
 
     public function executeCommand(Command $command) {
@@ -1343,7 +1408,7 @@ class Connection implements IConnection {
     }
 
     public function readBytes($length) {
-        if ($length == 0) {
+        if ($length <= 0) {
             throw new \InvalidArgumentException('Length parameter must be greater than 0');
         }
         $socket = $this->getSocket();
@@ -1364,12 +1429,12 @@ class Connection implements IConnection {
         $value  = '';
         do {
             $chunk = fgets($socket);
-            if ($chunk === false || strlen($chunk) == 0) {
+            if ($chunk === false || $chunk === '') {
                 $this->onCommunicationException('Error while reading line from the server');
             }
             $value .= $chunk;
         }
-        while (substr($value, -2) !== Protocol::NEWLINE);
+        while (substr($value, -2) !== "\r\n");
         return substr($value, 0, -2);
     }
 
@@ -1492,6 +1557,7 @@ abstract class RedisServerProfile {
         return array(
             '1.2'     => '\Predis\RedisServer_v1_2',
             '2.0'     => '\Predis\RedisServer_v2_0',
+            '2.2'     => '\Predis\RedisServer_v2_2',
             'default' => '\Predis\RedisServer_v2_0',
             'dev'     => '\Predis\RedisServer_vNext',
         );
@@ -1619,7 +1685,7 @@ class RedisServer_v1_2 extends RedisServerProfile {
             'type'                    => '\Predis\Commands\Type',
 
             /* commands operating on the key space */
-            'keys'               => '\Predis\Commands\Keys',
+            'keys'               => '\Predis\Commands\Keys_v1_2',
             'randomkey'          => '\Predis\Commands\RandomKey',
                 'randomKey'      => '\Predis\Commands\RandomKey',
             'rename'             => '\Predis\Commands\Rename',
@@ -1753,6 +1819,9 @@ class RedisServer_v2_0 extends RedisServer_v1_2 {
             'append'                    => '\Predis\Commands\Append',
             'substr'                    => '\Predis\Commands\Substr',
 
+            /* commands operating on the key space */
+            'keys'                      => '\Predis\Commands\Keys',
+
             /* commands operating on lists */
             'blpop'                     => '\Predis\Commands\ListPopFirstBlocking',
                 'popFirstBlocking'      => '\Predis\Commands\ListPopFirstBlocking',
@@ -1813,8 +1882,8 @@ class RedisServer_v2_0 extends RedisServer_v1_2 {
     }
 }
 
-class RedisServer_vNext extends RedisServer_v2_0 {
-    public function getVersion() { return '2.1'; }
+class RedisServer_v2_2 extends RedisServer_v2_0 {
+    public function getVersion() { return '2.2'; }
     public function getSupportedCommands() {
         return array_merge(parent::getSupportedCommands(), array(
             /* transactions */
@@ -1823,6 +1892,10 @@ class RedisServer_vNext extends RedisServer_v2_0 {
 
             /* commands operating on string values */
             'strlen'                    => '\Predis\Commands\Strlen',
+            'setrange'                  => '\Predis\Commands\SetRange',
+            'getrange'                  => '\Predis\Commands\Substr',
+            'setbit'                    => '\Predis\Commands\SetBit',
+            'getbit'                    => '\Predis\Commands\GetBit',
 
             /* commands operating on the key space */
             'persist'                   => '\Predis\Commands\Persist',
@@ -1831,6 +1904,7 @@ class RedisServer_vNext extends RedisServer_v2_0 {
             'rpushx'                    => '\Predis\Commands\ListPushTailX',
             'lpushx'                    => '\Predis\Commands\ListPushHeadX',
             'linsert'                   => '\Predis\Commands\ListInsert',
+            'brpoplpush'                => '\Predis\Commands\ListPopLastPushHeadBlocking',
 
             /* commands operating on sorted sets */
             'zrevrangebyscore'          => '\Predis\Commands\ZSetReverseRangeByScore',
@@ -1838,6 +1912,10 @@ class RedisServer_vNext extends RedisServer_v2_0 {
     }
 }
 
+class RedisServer_vNext extends RedisServer_v2_2 {
+    public function getVersion() { return 'DEV'; }
+}
+
 /* ------------------------------------------------------------------------- */
 
 namespace Predis\Pipeline;
@@ -2357,10 +2435,22 @@ class Append extends \Predis\MultiBulkCommand {
     public function getCommandId() { return 'APPEND'; }
 }
 
+class SetRange extends \Predis\MultiBulkCommand {
+    public function getCommandId() { return 'SETRANGE'; }
+}
+
 class Substr extends \Predis\MultiBulkCommand {
     public function getCommandId() { return 'SUBSTR'; }
 }
 
+class SetBit extends \Predis\MultiBulkCommand {
+    public function getCommandId() { return 'SETBIT'; }
+}
+
+class GetBit extends \Predis\MultiBulkCommand {
+    public function getCommandId() { return 'GETBIT'; }
+}
+
 class Strlen extends \Predis\MultiBulkCommand {
     public function getCommandId() { return 'STRLEN'; }
 }
@@ -2369,12 +2459,11 @@ class Strlen extends \Predis\MultiBulkCommand {
 class Keys extends \Predis\MultiBulkCommand {
     public function canBeHashed()  { return false; }
     public function getCommandId() { return 'KEYS'; }
-    public function parseResponse($data) { 
-        // TODO: is this behaviour correct?
-        if (is_array($data) || $data instanceof \Iterator) {
-            return $data;
-        }
-        return strlen($data) > 0 ? explode(' ', $data) : array();
+}
+
+class Keys_v1_2 extends Keys {
+    public function parseResponse($data) {
+        return explode(' ', $data);
     }
 }
 
@@ -2464,8 +2553,8 @@ class ListPopLastPushHead extends \Predis\MultiBulkCommand {
     public function getCommandId() { return 'RPOPLPUSH'; }
 }
 
-class ListPopLastPushHeadBulk extends \Predis\MultiBulkCommand {
-    public function getCommandId() { return 'RPOPLPUSH'; }
+class ListPopLastPushHeadBlocking extends \Predis\MultiBulkCommand {
+    public function getCommandId() { return 'BRPOPLPUSH'; }
 }
 
 class ListPopFirst extends \Predis\MultiBulkCommand {

test/PredisClientFeatures.php

@@ -216,6 +216,7 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase {
 
     function testResponseQueued() {
         $response = new \Predis\ResponseQueued();
+        $this->assertTrue($response->skipParse);
         $this->assertTrue($response->queued);
         $this->assertEquals(\Predis\Protocol::QUEUED, (string)$response);
     }
@@ -227,6 +228,7 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase {
         $errorMessage = 'ERROR MESSAGE';
         $response = new \Predis\ResponseError($errorMessage);
 
+        $this->assertTrue($response->skipParse);
         $this->assertTrue($response->error);
         $this->assertEquals($errorMessage, $response->message);
         $this->assertEquals($errorMessage, (string)$response);
@@ -524,12 +526,29 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase {
         $this->assertEquals('bar', $replies[2]);
     }
 
+    /**
+     * @expectedException Predis\ClientException
+     */
+    function testMultiExecBlock_CannotMixFluentInterfaceAndAnonymousBlock() {
+        $emptyBlock = function($tx) { };
+        $tx = RC::getConnection()->multiExec()->get('foo')->execute($emptyBlock);
+    }
+
     function testMultiExecBlock_EmptyCallableBlock() {
         $client = RC::getConnection();
         $client->flushdb();
 
         $replies = $client->multiExec(function($multi) { });
+        $this->assertEquals(0, count($replies));
 
+        $options = array('cas' => true);
+        $replies = $client->multiExec($options, function($multi) { });
+        $this->assertEquals(0, count($replies));
+
+        $options = array('cas' => true);
+        $replies = $client->multiExec($options, function($multi) {
+            $multi->multi();
+        });
         $this->assertEquals(0, count($replies));
     }
 
@@ -609,5 +628,128 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase {
 
         $this->assertEquals('client2', $client1->get('sentinel'));
     }
+
+    function testMultiExecBlock_CheckAndSet() {
+        $client = RC::getConnection();
+        $client->flushdb();
+        $client->set('foo', 'bar');
+
+        $options = array('watch' => 'foo', 'cas' => true);
+        $replies = $client->multiExec($options, function($tx) {
+            $tx->watch('foobar');
+            $foo = $tx->get('foo');
+            $tx->multi();
+            $tx->set('foobar', $foo);
+            $tx->mget('foo', 'foobar');
+        });
+        $this->assertType('array', $replies);
+        $this->assertEquals(array(true, array('bar', 'bar')), $replies);
+
+        $tx = $client->multiExec($options);
+        $tx->watch('foobar');
+        $foo = $tx->get('foo');
+        $replies = $tx->multi()
+                      ->set('foobar', $foo)
+                      ->mget('foo', 'foobar')
+                      ->execute();
+        $this->assertType('array', $replies);
+        $this->assertEquals(array(true, array('bar', 'bar')), $replies);
+    }
+
+    function testMultiExecBlock_RetryOnServerAbort() {
+        $client1 = RC::getConnection();
+        $client2 = RC::getConnection(true);
+        $client1->flushdb();
+
+        $retry = 3;
+        $attempts = 0;
+        RC::testForAbortedMultiExecException($this, function()
+            use($client1, $client2, $retry, &$attempts) {
+
+            $options = array('watch' => 'sentinel', 'retry' => $retry);
+            $client1->multiExec($options, function($tx)
+                use ($client2, &$attempts) {
+
+                $attempts++;
+                $tx->set('sentinel', 'client1');
+                $tx->get('sentinel');
+                $client2->set('sentinel', 'client2');
+            });
+        });
+        $this->assertEquals('client2', $client1->get('sentinel'));
+        $this->assertEquals($retry + 1, $attempts);
+
+        $retry = 3;
+        $attempts = 0;
+        RC::testForAbortedMultiExecException($this, function()
+            use($client1, $client2, $retry, &$attempts) {
+
+            $options = array(
+                'watch' => 'sentinel',
+                'cas'   => true,
+                'retry' => $retry
+            );
+            $client1->multiExec($options, function($tx)
+                use ($client2, &$attempts) {
+
+                $attempts++;
+                $tx->incr('attempts');
+                $tx->multi();
+                $tx->set('sentinel', 'client1');
+                $tx->get('sentinel');
+                $client2->set('sentinel', 'client2');
+            });
+        });
+        $this->assertEquals('client2', $client1->get('sentinel'));
+        $this->assertEquals($retry + 1, $attempts);
+        $this->assertEquals($attempts, $client1->get('attempts'));
+    }
+
+    /**
+     * @expectedException InvalidArgumentException
+     */
+    function testMultiExecBlock_RetryNotAvailableWithoutBlock() {
+        $options = array('watch' => 'foo', 'retry' => 1);
+        $tx = RC::getConnection()->multiExec($options);
+        $tx->multi()->get('foo')->exec();
+    }
+
+    function testMultiExecBlock_CheckAndSet_Discard() {
+        $client = RC::getConnection();
+        $client->flushdb();
+
+        $client->set('foo', 'bar');
+        $options = array('watch' => 'foo', 'cas' => true);
+        $replies = $client->multiExec($options, function($tx) {
+            $tx->watch('foobar');
+            $foo = $tx->get('foo');
+            $tx->multi();
+            $tx->set('foobar', $foo);
+            $tx->discard();
+            $tx->mget('foo', 'foobar');
+        });
+        $this->assertType('array', $replies);
+        $this->assertEquals(array(array('bar', null)), $replies);
+
+        $hijack = true;
+        $client->set('foo', 'bar');
+        $client2 = RC::getConnection(true);
+        $options = array('watch' => 'foo', 'cas' => true, 'retry' => 1);
+        $replies = $client->multiExec($options, function($tx)
+            use ($client2, &$hijack) {
+
+            $foo = $tx->get('foo');
+            $tx->multi();
+            $tx->set('foobar', $foo);
+            $tx->discard();
+            if ($hijack) {
+                $hijack = false;
+                $client2->set('foo', 'hijacked!');
+            }
+            $tx->mget('foo', 'foobar');
+        });
+        $this->assertType('array', $replies);
+        $this->assertEquals(array(array('hijacked!', null)), $replies);
+    }
 }
 ?>

test/PredisShared.php

@@ -20,11 +20,14 @@ class RC {
     const EXCEPTION_WRONG_TYPE     = 'Operation against a key holding the wrong kind of value';
     const EXCEPTION_NO_SUCH_KEY    = 'no such key';
     const EXCEPTION_OUT_OF_RANGE   = 'index out of range';
+    const EXCEPTION_OFFSET_RANGE   = 'offset is out of range';
     const EXCEPTION_INVALID_DB_IDX = 'invalid DB index';
     const EXCEPTION_VALUE_NOT_INT  = 'value is not an integer';
     const EXCEPTION_EXEC_NO_MULTI  = 'EXEC without MULTI';
     const EXCEPTION_SETEX_TTL      = 'invalid expire time in SETEX';
     const EXCEPTION_HASH_VALNOTINT = 'hash value is not an integer';
+    const EXCEPTION_BIT_VALUE      = 'bit is not an integer or out of range';
+    const EXCEPTION_BIT_OFFSET     = 'bit offset is not an integer or out of range';
 
     private static $_connection;
 

test/RedisCommandsTest.php

@@ -223,6 +223,28 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
         });
     }
 
+    function testSetRange() {
+        $this->assertEquals(6, $this->redis->setrange('var', 0, 'foobar'));
+        $this->assertEquals('foobar', $this->redis->get('var'));
+        $this->assertEquals(6, $this->redis->setrange('var', 3, 'foo'));
+        $this->assertEquals('foofoo', $this->redis->get('var'));
+        $this->assertEquals(16, $this->redis->setrange('var', 10, 'barbar'));
+        $this->assertEquals("foofoo\x00\x00\x00\x00barbar", $this->redis->get('var'));
+
+        $this->assertEquals(4, $this->redis->setrange('binary', 0, pack('l', -2147483648)));
+        list($unpacked) = array_values(unpack('l', $this->redis->get('binary')));
+        $this->assertEquals(-2147483648, $unpacked);
+
+        RC::testForServerException($this, RC::EXCEPTION_OFFSET_RANGE, function($test) {
+            $test->redis->setrange('var', -1, 'bogus');
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, function($test) {
+            $test->redis->rpush('metavars', 'foo');
+            $test->redis->setrange('metavars', 0, 'hoge');
+        });
+    }
+
     function testSubstr() {
         $this->redis->set('var', 'foobar');
         $this->assertEquals('foo', $this->redis->substr('var', 0, 2));
@@ -252,6 +274,61 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
         });
     }
 
+    function testSetBit() {
+        $this->assertEquals(0, $this->redis->setbit('binary', 31, 1));
+        $this->assertEquals(0, $this->redis->setbit('binary', 0, 1));
+        $this->assertEquals(4, $this->redis->strlen('binary'));
+        $this->assertEquals("\x80\x00\00\x01", $this->redis->get('binary'));
+
+        $this->assertEquals(1, $this->redis->setbit('binary', 0, 0));
+        $this->assertEquals(0, $this->redis->setbit('binary', 0, 0));
+        $this->assertEquals("\x00\x00\00\x01", $this->redis->get('binary'));
+
+        RC::testForServerException($this, RC::EXCEPTION_BIT_OFFSET, function($test) {
+            $test->redis->setbit('binary', -1, 1);
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_BIT_OFFSET, function($test) {
+            $test->redis->setbit('binary', 'invalid', 1);
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_BIT_VALUE, function($test) {
+            $test->redis->setbit('binary', 15, 255);
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_BIT_VALUE, function($test) {
+            $test->redis->setbit('binary', 15, 'invalid');
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, function($test) {
+            $test->redis->rpush('metavars', 'foo');
+            $test->redis->setbit('metavars', 0, 1);
+        });
+    }
+
+    function testGetBit() {
+        $this->redis->set('binary', "\x80\x00\00\x01");
+
+        $this->assertEquals(1, $this->redis->getbit('binary', 0));
+        $this->assertEquals(0, $this->redis->getbit('binary', 15));
+        $this->assertEquals(1, $this->redis->getbit('binary', 31));
+        $this->assertEquals(0, $this->redis->getbit('binary', 63));
+
+        RC::testForServerException($this, RC::EXCEPTION_BIT_OFFSET, function($test) {
+            $test->redis->getbit('binary', -1);
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_BIT_OFFSET, function($test) {
+            $test->redis->getbit('binary', 'invalid');
+        });
+
+        RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, function($test) {
+            $test->redis->rpush('metavars', 'foo');
+            $test->redis->getbit('metavars', 0);
+        });
+    }
+
+
 
     /* commands operating on the key space */
 
@@ -358,7 +435,8 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
         sleep(2);
         $this->assertFalse($this->redis->exists('hoge'));
 
-        RC::testForServerException($this, RC::EXCEPTION_VALUE_NOT_INT, function($test) {
+        // TODO: do not check the error message RC::EXCEPTION_VALUE_NOT_INT for now
+        RC::testForServerException($this, null, function($test) {
             $test->redis->setex('hoge', 2.5, 'piyo');
         });
         RC::testForServerException($this, RC::EXCEPTION_SETEX_TTL, function($test) {
@@ -733,6 +811,31 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
         $this->assertEquals((float)(time() - $start), 2, '', 1);
     }
 
+    function testListBlockingPopLastPushHead() {
+        // TODO: this test does not cover all the aspects of BLPOP/BRPOP as it
+        //       does not run with a concurrent client pushing items on lists.
+        $numbers = RC::pushTailAndReturn($this->redis, 'numbers', array(1, 2, 3));
+        $src_count = count($numbers);
+        $dst_count = 0;
+
+        while ($item = $this->redis->brpoplpush('numbers', 'temporary', 1)) {
+            $this->assertEquals(--$src_count, $this->redis->llen('numbers'));
+            $this->assertEquals(++$dst_count, $this->redis->llen('temporary'));
+            $this->assertEquals(array_pop($numbers), $this->redis->lindex('temporary', 0));
+        }
+
+        $start = time();
+        $this->assertNull($this->redis->brpoplpush('numbers', 'temporary', 2));
+        $this->assertEquals(2, (float)(time() - $start), '', 1);
+
+        RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, function($test) {
+            $test->redis->del('numbers');
+            $test->redis->del('temporary');
+            $test->redis->set('numbers', 'foobar');
+            $test->redis->brpoplpush('numbers', 'temporary', 1);
+        });
+    }
+
     function testListInsert() {
         $numbers = RC::pushTailAndReturn($this->redis, 'numbers', RC::getArrayOfNumbers());
 
@@ -745,7 +848,7 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
 
         RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, function($test) {
             $test->redis->set('foo', 'bar');
-            $test->redis->lset('foo', 0, 0);
+            $test->redis->linsert('foo', 'before', 0, 0);
         });
     }