Backported changes from the mainline library to the PHP 5.2 branch (up to commit 35016ba)

LICENSE

@@ -1,5 +1,5 @@
-Copyright (c) 2009 Daniele Alessandri
-
+Copyright (c) 2009-2010 Daniele Alessandri
+qq
 Permission is hereby granted, free of charge, to any person
 obtaining a copy of this software and associated documentation
 files (the "Software"), to deal in the Software without
@@ -19,4 +19,4 @@ NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
-OTHER DEALINGS IN THE SOFTWARE.
+OTHER DEALINGS IN THE SOFTWARE.

README.markdown

@@ -7,7 +7,7 @@ database.
 
 Predis is currently a work-in-progress and it comes in two flavors:
 
- - the mainline client library, which targets PHP 5.3.x and exploits a lot of the 
+ - the mainline client library, which targets PHP 5.3.x and leverages a lot of the 
    features introduced in this new version of the PHP interpreter.
  - a backport to PHP 5.2.x for those who can not upgrade their environment yet 
    (it admittedly has a lower priority compared to the mainline library, although we 
@@ -27,6 +27,9 @@ to be implemented soon in Predis.
 
 ## Quick examples ##
 
+See the [official wiki](http://wiki.github.com/nrk/predis) of the project for a more 
+complete coverage of all the features available in Predis.
+
 ### Connecting to a local instance of Redis ###
 
 You don't have to specify a tcp host and port when connecting to Redis instances 
@@ -109,13 +112,15 @@ variable set to E_ALL.
 
 ## Dependencies ##
 
-- PHP >= 5.2
+- PHP >= 5.3.0 (for the mainline client library)
+- PHP >= 5.2.6 (for the backported client library)
 - PHPUnit (needed to run the test suite)
 
 ## Links ##
 
 ### Project ###
 - [Source code](http://github.com/nrk/predis/)
+- [Wiki](http://wiki.github.com/nrk/predis/)
 - [Issue tracker](http://github.com/nrk/predis/issues)
 
 ### Related ###

lib/Predis.php

@@ -7,13 +7,10 @@ class Predis_MalformedServerResponse extends Predis_ServerException { }
 /* ------------------------------------------------------------------------- */
 
 class Predis_Client {
-    // TODO: command arguments should be sanitized or checked for bad arguments 
-    //       (e.g. CRLF in keys for inline commands)
-
     private $_connection, $_serverProfile;
 
     public function __construct($parameters = null, Predis_RedisServerProfile $serverProfile = null) {
-        $this->setServerProfile(
+        $this->setProfile(
             $serverProfile === null 
                 ? Predis_RedisServerProfile::getDefault() 
                 : $serverProfile
@@ -82,10 +79,14 @@ class Predis_Client {
         $this->_connection = $connection;
     }
 
-    public function setServerProfile(Predis_RedisServerProfile $serverProfile) {
+    public function setProfile(Predis_RedisServerProfile $serverProfile) {
         $this->_serverProfile = $serverProfile;
     }
 
+    public function getProfile() {
+        return $this->_serverProfile;
+    }
+
     public function connect() {
         $this->_connection->connect();
     }
@@ -137,28 +138,20 @@ class Predis_Client {
     }
 
     public function rawCommand($rawCommandData, $closesConnection = false) {
-        // TODO: rather than check the type of a connection instance, we should 
-        //       check if it does respond to the rawCommand method.
         if (is_a($this->_connection, 'Predis_ConnectionCluster')) {
             throw new Predis_ClientException('Cannot send raw commands when connected to a cluster of Redis servers');
         }
         return $this->_connection->rawCommand($rawCommandData, $closesConnection);
     }
 
-    public function pipeline() {
-        return new Predis_CommandPipeline($this);
-    }
-
-    public function multiExec() {
-        return new Predis_MultiExecBlock($this);
+    public function pipeline($pipelineBlock = null) {
+        $pipeline = new Predis_CommandPipeline($this);
+        return $pipelineBlock !== null ? $pipeline->execute($pipelineBlock) : $pipeline;
     }
 
-    public function registerCommands(Array $commands) {
-        $this->_serverProfile->registerCommands($commands);
-    }
-
-    public function registerCommand($command, $aliases) {
-        $this->_serverProfile->registerCommand($command, $aliases);
+    public function multiExec($multiExecBlock = null) {
+        $multiExec = new Predis_MultiExecBlock($this);
+        return $multiExecBlock !== null ? $multiExec->execute($multiExecBlock) : $multiExec;
     }
 }
 
@@ -305,20 +298,7 @@ class Predis_Response {
         );
     }
 
-    public static function getPrefixHandler($prefix) {
-        if (self::$_prefixHandlers === null) {
-            self::$_prefixHandlers = self::initializePrefixHandlers();
-        }
-
-        $handler = self::$_prefixHandlers[$prefix];
-        if ($handler === null) {
-            throw new Predis_MalformedServerResponse("Unknown prefix '$prefix'");
-        }
-        return $handler;
-    }
-
-    public static function handleStatus($socket) {
-        $status = rtrim(fgets($socket), Predis_Response::NEWLINE);
+    public static function handleStatus($socket, $prefix, $status) {
         if ($status === Predis_Response::OK) {
             return true;
         }
@@ -328,20 +308,20 @@ class Predis_Response {
         return $status;
     }
 
-    public static function handleError($socket) {
-        $errorMessage = rtrim(fgets($socket), Predis_Response::NEWLINE);
+    public static function handleError($socket, $prefix, $errorMessage) {
         throw new Predis_ServerException(substr($errorMessage, 4));
     }
 
-    public static function handleBulk($socket) {
-        $dataLength = rtrim(fgets($socket), Predis_Response::NEWLINE);
-
+    public static function handleBulk($socket, $prefix, $dataLength) {
         if (!is_numeric($dataLength)) {
             throw new Predis_ClientException("Cannot parse '$dataLength' as data length");
         }
 
         if ($dataLength > 0) {
             $value = stream_get_contents($socket, $dataLength);
+            if ($value === false) {
+                throw new Predis_ClientException('An error has occurred while reading from the network stream');
+            }
             fread($socket, 2);
             return $value;
         }
@@ -353,8 +333,7 @@ class Predis_Response {
         return null;
     }
 
-    public static function handleMultiBulk($socket) {
-        $rawLength = rtrim(fgets($socket), Predis_Response::NEWLINE);
+    public static function handleMultiBulk($socket, $prefix, $rawLength) {
         if (!is_numeric($rawLength)) {
             throw new Predis_ClientException("Cannot parse '$rawLength' as data length");
         }
@@ -368,26 +347,44 @@ class Predis_Response {
 
         if ($listLength > 0) {
             for ($i = 0; $i < $listLength; $i++) {
-                $handler = Predis_Response::getPrefixHandler(fgetc($socket));
-                $list[] = call_user_func($handler, $socket);
+                $list[] = Predis_Response::read($socket);
             }
         }
 
         return $list;
     }
 
-    public static function handleInteger($socket) {
-        $number = rtrim(fgets($socket), Predis_Response::NEWLINE);
+    public static function handleInteger($socket, $prefix, $number) {
         if (is_numeric($number)) {
             return (int) $number;
         }
         else {
-            if ($number !== Predis_Response::NULL) {
+            if ($number !== Response::NULL) {
                 throw new Predis_ClientException("Cannot parse '$number' as numeric response");
             }
             return null;
         }
     }
+
+    public static function read($socket) {
+        $header  = fgets($socket);
+        if ($header === false) {
+           throw new Predis_ClientException('An error has occurred while reading from the network stream');
+        }
+
+        $prefix  = $header[0];
+        $payload = substr($header, 1, -2);
+
+        if (!isset(self::$_prefixHandlers)) {
+            self::$_prefixHandlers = self::initializePrefixHandlers();
+        }
+        if (!isset(self::$_prefixHandlers[$prefix])) {
+            throw new Predis_MalformedServerResponse("Unknown prefix '$prefix'");
+        }
+
+        $handler = self::$_prefixHandlers[$prefix];
+        return call_user_func($handler, $socket, $prefix, $payload);
+    }
 }
 
 class Predis_ResponseQueued {
@@ -421,25 +418,24 @@ class Predis_CommandPipeline {
     }
 
     public function flushPipeline() {
-        if (count($this->_pipelineBuffer) === 0) {
+        $sizeofPipe = count($this->_pipelineBuffer);
+        if ($sizeofPipe === 0) {
             return;
         }
 
         $connection = $this->_redisClient->getConnection();
-        $commands   = $this->getRecordedCommands();
+        $commands   = &$this->_pipelineBuffer;
 
         foreach ($commands as $command) {
             $connection->writeCommand($command);
         }
-        foreach ($commands as $command) {
-            $this->_returnValues[] = $connection->readResponse($command);
+        for ($i = 0; $i < $sizeofPipe; $i++) {
+            $this->_returnValues[] = $connection->readResponse($commands[$i]);
+            unset($commands[$i]);
         }
-
-        $this->_pipelineBuffer = array();
     }
 
     private function setRunning($bool) {
-        // TODO: I am honest when I say that I don't like this approach.
         if ($bool == true && $this->_running == true) {
             throw new Predis_ClientException("This pipeline is already opened");
         }
@@ -447,11 +443,18 @@ class Predis_CommandPipeline {
         $this->_running = $bool;
     }
 
-    public function execute() {
+    public function execute($block = null) {
+        if ($block && !is_callable($block)) {
+            throw new RuntimeException('Argument passed must be a callable object');
+        }
+
         $this->setRunning(true);
         $pipelineBlockException = null;
 
         try {
+            if ($block !== null) {
+                $block($this);
+            }
             $this->flushPipeline();
         }
         catch (Exception $exception) {
@@ -497,14 +500,31 @@ class Predis_MultiExecBlock {
         }
     }
 
-    public function execute() {
+    public function execute($block = null) {
+        if ($block && !is_callable($block)) {
+            throw new RuntimeException('Argument passed must be a callable object');
+        }
+
         $blockException = null;
         $returnValues   = array();
 
         try {
+            if ($block !== null) {
+                $block($this);
+            }
+
             $execReply = $this->_redisClient->exec();
-            for ($i = 0; $i < count($execReply); $i++) {
-                $returnValues[] = $this->_commands[$i]->parseResponse($execReply[$i]);
+            $commands  = &$this->_commands;
+            $sizeofReplies = count($execReply);
+
+            if ($sizeofReplies !== count($commands)) {
+                // TODO: think of a better exception message
+                throw new Predis_ClientException("Out-of-sync");
+            }
+
+            for ($i = 0; $i < $sizeofReplies; $i++) {
+                $returnValues[] = $commands[$i]->parseResponse($execReply[$i]);
+                unset($commands[$i]);
             }
         }
         catch (Exception $exception) {
@@ -657,24 +677,30 @@ class Predis_Connection implements Predis_IConnection {
     }
 
     public function writeCommand(Predis_Command $command) {
-        fwrite($this->getSocket(), $command->invoke());
+        $written = fwrite($this->getSocket(), $command->invoke());
+        if ($written === false){
+           throw new Predis_ClientException(sprintf(
+               'An error has occurred while writing command %s on the network stream'),
+               $command->getCommandId()
+           );
+        }
     }
 
     public function readResponse(Predis_Command $command) {
-        $socket   = $this->getSocket();
-        $handler  = Predis_Response::getPrefixHandler(fgetc($socket));
-        $response = call_user_func($handler, $socket);
+        $response = Predis_Response::read($this->getSocket());
         return isset($response->queued) ? $response : $command->parseResponse($response);
     }
 
     public function rawCommand($rawCommandData, $closesConnection = false) {
         $socket = $this->getSocket();
-        fwrite($socket, $rawCommandData);
+        $written = fwrite($socket, $rawCommandData);
+        if ($written === false){
+           throw new Predis_ClientException('An error has occurred while writing a raw command on the network stream');
+        }
         if ($closesConnection) {
             return;
         }
-        $handler = Predis_Response::getPrefixHandler(fgetc($socket));
-        return call_user_func($handler, $socket);
+        return Predis_Response::read($socket);
     }
 
     public function getSocket() {
@@ -758,7 +784,7 @@ class Predis_ConnectionCluster implements Predis_IConnection, IteratorAggregate
 /* ------------------------------------------------------------------------- */
 
 abstract class Predis_RedisServerProfile {
-    const DEFAULT_SERVER_PROFILE = 'Predis_RedisServer__V1_2';
+    const DEFAULT_SERVER_PROFILE = 'Predis_RedisServer_v1_2';
     private $_registeredCommands;
 
     public function __construct() {
@@ -774,17 +800,27 @@ abstract class Predis_RedisServerProfile {
         return new $defaultProfile();
     }
 
+    public function compareWith($version, $operator = null) {
+        // one could expect that PHP's version_compare would behave 
+        // the same way if invoked with 2 arguments or 3 arguments 
+        // with the third being NULL, but it is not like that.
+        // TODO: since version_compare considers 1 < 1.0 < 1.0.0, 
+        //       we might need to revise the behavior of this method.
+        return ($operator === null 
+            ? version_compare($this, $version)
+            : version_compare($this, $version, $operator)
+        );
+    }
+
     public function supportsCommand($command) {
         return isset($this->_registeredCommands[$command]);
     }
 
     public function createCommand($method, $arguments = array()) {
-        $commandClass = $this->_registeredCommands[$method];
-
-        if ($commandClass === null) {
+        if (!isset($this->_registeredCommands[$method])) {
             throw new Predis_ClientException("'$method' is not a registered Redis command");
         }
-
+        $commandClass = $this->_registeredCommands[$method];
         $command = new $commandClass();
         $command->setArgumentsArray($arguments);
         return $command;
@@ -812,10 +848,14 @@ abstract class Predis_RedisServerProfile {
             $this->_registeredCommands[$aliases] = $command;
         }
     }
+
+    public function __toString() {
+        return $this->getVersion();
+    }
 }
 
-class Predis_RedisServer__V1_0 extends Predis_RedisServerProfile {
-    public function getVersion() { return 1.0; }
+class Predis_RedisServer_v1_0 extends Predis_RedisServerProfile {
+    public function getVersion() { return '1.0'; }
     public function getSupportedCommands() {
         return array(
             /* miscellaneous commands */
@@ -939,13 +979,13 @@ class Predis_RedisServer__V1_0 extends Predis_RedisServerProfile {
                 'backgroundSave'    => 'Predis_Commands_BackgroundSave',
             'lastsave'              => 'Predis_Commands_LastSave', 
                 'lastSave'          => 'Predis_Commands_LastSave',
-            'shutdown'              => 'Predis_Commands_Shutdown'
+            'shutdown'              => 'Predis_Commands_Shutdown',
         );
     }
 }
 
-class Predis_RedisServer__V1_2 extends Predis_RedisServer__V1_0 {
-    public function getVersion() { return 1.2; }
+class Predis_RedisServer_v1_2 extends Predis_RedisServer_v1_0 {
+    public function getVersion() { return '1.2'; }
     public function getSupportedCommands() {
         return array_merge(parent::getSupportedCommands(), array(
             /* commands operating on string values */
@@ -976,17 +1016,24 @@ class Predis_RedisServer__V1_2 extends Predis_RedisServer__V1_0 {
             'zscore'                        => 'Predis_Commands_ZSetScore',
                 'zsetScore'                 => 'Predis_Commands_ZSetScore',
             'zremrangebyscore'              => 'Predis_Commands_ZSetRemoveRangeByScore',
-                'zsetRemoveRangeByScore'    => 'Predis_Commands_ZSetRemoveRangeByScore'
+                'zsetRemoveRangeByScore'    => 'Predis_Commands_ZSetRemoveRangeByScore',
         ));
     }
 }
 
-class Predis_RedisServer__Futures extends Predis_RedisServer__V1_2 {
-    public function getVersion() { return 0; }
+class Predis_RedisServer_vNext extends Predis_RedisServer_v1_2 {
+    public function getVersion() { return '1.3'; }
     public function getSupportedCommands() {
         return array_merge(parent::getSupportedCommands(), array(
+            /* miscellaneous commands */
             'multi'     => 'Predis_Commands_Multi',
-            'exec'      => 'Predis_Commands_Exec'
+            'exec'      => 'Predis_Commands_Exec',
+
+            /* commands operating on lists */
+            'blpop'                     => 'Predis_Commands_ListPopFirstBlocking',
+                'popFirstBlocking'      => 'Predis_Commands_ListPopFirstBlocking',
+            'brpop'                     => 'Predis_Commands_ListPopLastBlocking',
+                'popLastBlocking'       => 'Predis_Commands_ListPopLastBlocking',
         ));
     }
 }
@@ -1167,7 +1214,6 @@ class Predis_Commands_RandomKey extends Predis_InlineCommand {
 }
 
 class Predis_Commands_Rename extends Predis_InlineCommand {
-    // TODO: doesn't RENAME break the hash-based client-side sharding?
     public function canBeHashed()  { return false; }
     public function getCommandId() { return 'RENAME'; }
 }
@@ -1242,6 +1288,14 @@ class Predis_Commands_ListPopLast extends Predis_InlineCommand {
     public function getCommandId() { return 'RPOP'; }
 }
 
+class Predis_Commands_ListPopFirstBlocking extends Predis_InlineCommand {
+    public function getCommandId() { return 'BLPOP'; }
+}
+
+class Predis_Commands_ListPopLastBlocking extends Predis_InlineCommand {
+    public function getCommandId() { return 'BRPOP'; }
+}
+
 /* commands operating on sets */
 class Predis_Commands_SetAdd extends Predis_BulkCommand {
     public function getCommandId() { return 'SADD'; }
@@ -1422,6 +1476,12 @@ class Predis_Commands_Save extends Predis_InlineCommand {
 class Predis_Commands_BackgroundSave extends Predis_InlineCommand {
     public function canBeHashed()  { return false; }
     public function getCommandId() { return 'BGSAVE'; }
+    public function parseResponse($data) {
+        if ($data == 'Background saving started') {
+            return true;
+        }
+        return $data;
+    }
 }
 
 class Predis_Commands_LastSave extends Predis_InlineCommand {

test/PredisShared.php

@@ -30,7 +30,7 @@ class RC {
     private static $_connection;
 
     private static function createConnection() {
-        $serverProfile = new Predis_RedisServer__Futures();
+        $serverProfile = new Predis_RedisServer_vNext();
         $connection = new Predis_Client(array('host' => RC::SERVER_HOST, 'port' => RC::SERVER_PORT), $serverProfile);
         $connection->connect();
         $connection->selectDatabase(RC::DEFAULT_DATABASE);
@@ -51,6 +51,22 @@ class RC {
         }
     }
 
+    public static function helperForBlockingPops($op) {
+        // TODO: I admit that this helper is kinda lame and it does not run 
+        //       in a separate process to properly test BLPOP/BRPOP
+        $redisUri = sprintf('redis://%s:%d/?database=%d', RC::SERVER_HOST, RC::SERVER_PORT, RC::DEFAULT_DATABASE);
+        $handle = popen('php', 'w');
+        fwrite($handle, "<?php
+        require '../lib/Predis.php';
+        \$redis = Predis_Client::create('$redisUri');
+        \$redis->rpush('{$op}1', 'a');
+        \$redis->rpush('{$op}2', 'b');
+        \$redis->rpush('{$op}3', 'c');
+        \$redis->rpush('{$op}1', 'd');
+        ?>");
+        pclose($handle);
+    }
+
     public static function getArrayOfNumbers() {
         return array(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
     }

test/RedisCommandsTest.php

@@ -47,7 +47,7 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
 
     function testMultiExec() {
         // NOTE: due to a limitation in the current implementation of Predis_Client, 
-        //       the replies returned by Predis_Command\Exec are not parsed by their 
+        //       the replies returned by Predis_Command_Exec are not parsed by their 
         //       respective Predis_Command::parseResponse methods. If you need that 
         //       kind of behaviour, you should use an instance of Predis_MultiExecBlock.
         $this->assertTrue($this->redis->multi());
@@ -553,6 +553,60 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
         "));
     }
 
+    function testListBlockingPopFirst() {
+        // 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. 
+        RC::helperForBlockingPops('blpop');
+
+        // BLPOP on one key
+        $start = time();
+        $item = $this->redis->blpop('blpop3', 5);
+        $this->assertEquals((float)(time() - $start), 0, '', 1);
+        $this->assertEquals($item, array('blpop3', 'c'));
+
+        // BLPOP on more than one key
+        $poppedItems = array();
+        while ($item = $this->redis->blpop('blpop1', 'blpop2', 1)) {
+            $poppedItems[] = $item;
+        }
+        $this->assertEquals(
+            array(array('blpop1', 'a'), array('blpop1', 'd'), array('blpop2', 'b')),
+            $poppedItems
+        );
+
+        // check if BLPOP timeouts as expected on empty lists
+        $start = time();
+        $this->redis->blpop('blpop4', 2);
+        $this->assertEquals((float)(time() - $start), 2, '', 1);
+    }
+
+    function testListBlockingPopLast() {
+        // 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. 
+        RC::helperForBlockingPops('brpop');
+
+        // BRPOP on one key
+        $start = time();
+        $item = $this->redis->brpop('brpop3', 5);
+        $this->assertEquals((float)(time() - $start), 0, '', 1);
+        $this->assertEquals($item, array('brpop3', 'c'));
+
+        // BRPOP on more than one key
+        $poppedItems = array();
+        while ($item = $this->redis->brpop('brpop1', 'brpop2', 1)) {
+            $poppedItems[] = $item;
+        }
+        $this->assertEquals(
+            array(array('brpop1', 'd'), array('brpop1', 'a'), array('brpop2', 'b')),
+            $poppedItems
+        );
+
+        // check if BRPOP timeouts as expected on empty lists
+        $start = time();
+        $this->redis->brpop('brpop4', 2);
+        $this->assertEquals((float)(time() - $start), 2, '', 1);
+    }
+
 
     /* commands operating on sets */