Compare commits

..

2 Commits

4 changed files with 347 additions and 57 deletions
+39
View File
@@ -1,3 +1,42 @@
v0.6.1 (2010-07-11)
* Minor internal improvements and clean ups.
* New commands available in the Redis v2.2 profile (dev):
- Misc. : WATCH, UNWATCH
* Optional modifiers for ZRANGE, ZREVRANGE and ZRANGEBYSCORE queries are
supported using an associative array passed as the last argument of their
respective methods.
* The LIMIT modifier for ZRANGEBYSCORE can be specified using either:
- an indexed array: array($offset, $count)
- an associative array: array('offset' => $offset, 'count' => $count)
* The method Predis_Client::__construct() now accepts also instances of
Predis_ConnectionParameters.
* Predis_MultiExecBlock and Predis_PubSubContext now throw an exception
when trying to create their instances using a profile that does not
support the required Redis commands or when the client is connected to
a cluster of connections.
* Various improvements to Predis_MultiExecBlock:
- fixes and more consistent behaviour across various usage cases.
- support for WATCH and UNWATCH when using the current development
profile (Redis v2.2) and aborted transactions.
* New signature for Predis_Client::multiExec() which is now able to accept
an array of options for the underlying instance of Predis_MultiExecBlock.
Backwards compatibility with previous releases of Predis is ensured.
* New signature for Predis_Client::pipeline() which is now able to accept
an array of options for the underlying instance of Predis_CommandPipeline.
Backwards compatibility with previous releases of Predis is ensured.
The method Predis_Client::pipelineSafe() is to be considered deprecated.
* FIX: The WEIGHT modifier for ZUNIONSTORE and ZINTERSTORE was handled
incorrectly with more than two weights specified.
v0.6.0 (2010-05-24)
* Switched to the new multi-bulk request protocol for all of the commands
in the Redis 1.2 and Redis 2.0 profiles. Inline and bulk requests are now
+1 -1
View File
@@ -1 +1 @@
0.6.0
0.6.1
+272 -56
View File
@@ -4,6 +4,9 @@ class PredisException extends Exception { }
// Client-side errors
class Predis_ClientException extends PredisException { }
// Aborted multi/exec
class Predis_AbortedMultiExec extends PredisException { }
// Server-side errors
class Predis_ServerException extends PredisException {
public function toResponseError() {
@@ -53,7 +56,7 @@ class Predis_Client {
throw new Predis_ClientException('Missing connection parameters');
}
return new Predis_Client($argc === 1 ? $argv[0] : $argv, $serverProfile);
return new Predis_Client($argc === 1 ? $argv[0] : $argv, $options);
}
private static function filterClientOptions($options) {
@@ -113,7 +116,9 @@ class Predis_Client {
}
private function createConnection($parameters) {
$params = new Predis_ConnectionParameters($parameters);
$params = $parameters instanceof Predis_ConnectionParameters
? $parameters
: new Predis_ConnectionParameters($parameters);
$connection = new Predis_Connection($params, $this->_responseReader);
if ($params->password !== null) {
@@ -155,7 +160,7 @@ class Predis_Client {
}
public function getClientFor($connectionAlias) {
if (!($this->_connection instanceof Predis_ConnectionCluster)) {
if (!Predis_Shared_Utils::isCluster($this->_connection)) {
throw new Predis_ClientException(
'This method is supported only when the client is connected to a cluster of connections'
);
@@ -191,7 +196,7 @@ class Predis_Client {
return $this->_connection;
}
else {
return $this->_connection instanceof Predis_ConnectionCluster
return Predis_Shared_Utils::isCluster($this->_connection)
? $this->_connection->getConnectionById($id)
: $this->_connection;
}
@@ -212,7 +217,7 @@ class Predis_Client {
public function executeCommandOnShards(Predis_Command $command) {
$replies = array();
if ($this->_connection instanceof Predis_ConnectionCluster) {
if (Predis_Shared_Utils::isCluster($this->_connection)) {
foreach($this->_connection as $connection) {
$replies[] = $connection->executeCommand($command);
}
@@ -224,22 +229,50 @@ class Predis_Client {
}
public function rawCommand($rawCommandData, $closesConnection = false) {
if ($this->_connection instanceof Predis_ConnectionCluster) {
if (Predis_Shared_Utils::isCluster($this->_connection)) {
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($pipelineBlock = null) {
return $this->pipelineExecute(new Predis_CommandPipeline($this), $pipelineBlock);
public function pipeline(/* arguments */) {
$argv = func_get_args();
$argc = func_num_args();
if ($argc === 0) {
return $this->initPipeline();
}
else if ($argc === 1) {
list($arg0) = $argv;
return is_array($arg0) ? $this->initPipeline($arg0) : $this->initPipeline(null, $arg0);
}
else if ($argc === 2) {
list($arg0, $arg1) = $argv;
return $this->initPipeline($arg0, $arg1);
}
}
public function pipelineSafe($pipelineBlock = null) {
$connection = $this->getConnection();
$pipeline = new Predis_CommandPipeline($this, $connection instanceof Predis_Connection
? new Predis_Pipeline_SafeExecutor($connection)
: new Predis_Pipeline_SafeClusterExecutor($connection)
);
return $this->initPipeline(array('safe' => true), $pipelineBlock);
}
private function initPipeline(Array $options = null, $pipelineBlock = null) {
$pipeline = null;
if (isset($options)) {
if (isset($options['safe']) && $options['safe'] == true) {
$connection = $this->getConnection();
$pipeline = new Predis_CommandPipeline($this, $connection instanceof Predis_Connection
? new Predis_Pipeline_SafeExecutor($connection)
: new Predis_Pipeline_SafeClusterExecutor($connection)
);
}
else {
$pipeline = new Predis_CommandPipeline($this);
}
}
else {
$pipeline = new Predis_CommandPipeline($this);
}
return $this->pipelineExecute($pipeline, $pipelineBlock);
}
@@ -247,9 +280,26 @@ class Predis_Client {
return $block !== null ? $pipeline->execute($block) : $pipeline;
}
public function multiExec($multiExecBlock = null) {
$multiExec = new Predis_MultiExecBlock($this);
return $multiExecBlock !== null ? $multiExec->execute($multiExecBlock) : $multiExec;
public function multiExec(/* arguments */) {
$argv = func_get_args();
$argc = func_num_args();
if ($argc === 0) {
return $this->initMultiExec();
}
else if ($argc === 1) {
list($arg0) = $argv;
return is_array($arg0) ? $this->initMultiExec($arg0) : $this->initMultiExec(null, $arg0);
}
else if ($argc === 2) {
list($arg0, $arg1) = $argv;
return $this->initMultiExec($arg0, $arg1);
}
}
private function initMultiExec(Array $options = null, $transBlock = null) {
$multi = isset($options) ? new Predis_MultiExecBlock($this, $options) : new Predis_MultiExecBlock($this);
return $transBlock !== null ? $multi->execute($transBlock) : $multi;
}
public function pubSubContext() {
@@ -433,8 +483,7 @@ abstract class Predis_Command {
$this->_hash = null;
}
protected function getArguments() {
// TODO: why getArguments is protected?
public function getArguments() {
return isset($this->_arguments) ? $this->_arguments : array();
}
@@ -577,8 +626,9 @@ class Predis_ResponseMultiBulkHandler implements Predis_IResponseHandler {
$list = array();
if ($listLength > 0) {
$reader = $connection->getResponseReader();
for ($i = 0; $i < $listLength; $i++) {
$list[] = $connection->getResponseReader()->read($connection);
$list[] = $reader->read($connection);
}
}
@@ -770,23 +820,58 @@ class Predis_CommandPipeline {
}
class Predis_MultiExecBlock {
private $_redisClient, $_commands, $_initialized, $_discarded;
private $_initialized, $_discarded, $_insideBlock;
private $_redisClient, $_options, $_commands;
private $_supportsWatch;
public function __construct(Predis_Client $redisClient) {
public function __construct(Predis_Client $redisClient, Array $options = null) {
$this->checkCapabilities($redisClient);
$this->_initialized = false;
$this->_discarded = false;
$this->_insideBlock = false;
$this->_redisClient = $redisClient;
$this->_options = isset($options) ? $options : array();
$this->_commands = array();
}
private function checkCapabilities(Predis_Client $redisClient) {
if (Predis_Shared_Utils::isCluster($redisClient->getConnection())) {
throw new Predis_ClientException(
'Cannot initialize a MULTI/EXEC context over a cluster of connections'
);
}
$profile = $redisClient->getProfile();
if ($profile->supportsCommands(array('multi', 'exec', 'discard')) === false) {
throw new Predis_ClientException(
'The current profile does not support MULTI, EXEC and DISCARD commands'
);
}
$this->_supportsWatch = $profile->supportsCommands(array('watch', 'unwatch'));
}
private function isWatchSupported() {
if ($this->_supportsWatch === false) {
throw new Predis_ClientException(
'The current profile does not support WATCH and UNWATCH commands'
);
}
}
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 setInsideBlock($value) {
$this->_insideBlock = $value;
}
public function __call($method, $arguments) {
$this->initialize();
$command = $this->_redisClient->createCommand($method, $arguments);
@@ -800,6 +885,34 @@ class Predis_MultiExecBlock {
}
}
public function watch($keys) {
$this->isWatchSupported();
if ($this->_initialized === true) {
throw new Predis_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;
}
public function multi() {
$this->initialize();
}
public function unwatch() {
$this->isWatchSupported();
$this->_redisClient->unwatch();
}
public function discard() {
$this->_redisClient->discard();
$this->_commands = array();
@@ -807,7 +920,17 @@ class Predis_MultiExecBlock {
$this->_discarded = true;
}
public function exec() {
return $this->execute();
}
public function execute($block = null) {
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');
}
@@ -817,17 +940,21 @@ class Predis_MultiExecBlock {
try {
if ($block !== null) {
$this->setInsideBlock(true);
$block($this);
$this->setInsideBlock(false);
}
if ($this->_discarded === true) {
return;
}
$execReply = (($reply = $this->_redisClient->exec()) instanceof Iterator
? iterator_to_array($reply)
: $reply
);
$reply = $this->_redisClient->exec();
if ($reply === null) {
throw new Predis_AbortedMultiExec('The current transaction has been aborted by the server');
}
$execReply = $reply instanceof Iterator ? iterator_to_array($reply) : $reply;
$commands = &$this->_commands;
$sizeofReplies = count($execReply);
@@ -844,6 +971,7 @@ class Predis_MultiExecBlock {
}
}
catch (Exception $exception) {
$this->setInsideBlock(false);
$blockException = $exception;
}
@@ -863,12 +991,16 @@ class Predis_PubSubContext implements Iterator {
const MESSAGE = 'message';
const PMESSAGE = 'pmessage';
private $_redisClient, $_subscriptions, $_isStillValid, $_position;
const STATUS_VALID = 0x0001;
const STATUS_SUBSCRIBED = 0x0010;
const STATUS_PSUBSCRIBED = 0x0100;
private $_redisClient, $_position;
public function __construct(Predis_Client $redisClient) {
$this->_redisClient = $redisClient;
$this->_isStillValid = true;
$this->_subscriptions = false;
$this->checkCapabilities($redisClient);
$this->_redisClient = $redisClient;
$this->_statusFlags = self::STATUS_VALID;
}
public function __destruct() {
@@ -878,10 +1010,29 @@ class Predis_PubSubContext implements Iterator {
}
}
private function checkCapabilities(Predis_Client $redisClient) {
if (Predis_Shared_Utils::isCluster($redisClient->getConnection())) {
throw new Predis_ClientException(
'Cannot initialize a PUB/SUB context over a cluster of connections'
);
}
$profile = $redisClient->getProfile();
$commands = array('publish', 'subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe');
if ($profile->supportsCommands($commands) === false) {
throw new Predis_ClientException(
'The current profile does not support PUB/SUB related commands'
);
}
}
private function isFlagSet($value) {
return ($this->_statusFlags & $value) === $value;
}
public function subscribe(/* arguments */) {
$args = func_get_args();
$this->writeCommand(self::SUBSCRIBE, $args);
$this->_subscriptions = true;
$this->_statusFlags |= self::STATUS_SUBSCRIBED;
}
public function unsubscribe(/* arguments */) {
@@ -892,7 +1043,7 @@ class Predis_PubSubContext implements Iterator {
public function psubscribe(/* arguments */) {
$args = func_get_args();
$this->writeCommand(self::PSUBSCRIBE, $args);
$this->_subscriptions = true;
$this->_statusFlags |= self::STATUS_PSUBSCRIBED;
}
public function punsubscribe(/* arguments */) {
@@ -902,10 +1053,12 @@ class Predis_PubSubContext implements Iterator {
public function closeContext() {
if ($this->valid()) {
// TODO: as an optimization, we should not send both
// commands if one of them has not been issued.
$this->unsubscribe();
$this->punsubscribe();
if ($this->isFlagSet(self::STATUS_SUBSCRIBED)) {
$this->unsubscribe();
}
if ($this->isFlagSet(self::STATUS_PSUBSCRIBED)) {
$this->punsubscribe();
}
}
}
@@ -930,19 +1083,20 @@ class Predis_PubSubContext implements Iterator {
}
public function next() {
if ($this->_isStillValid) {
if ($this->isFlagSet(self::STATUS_VALID)) {
$this->_position++;
}
return $this->_position;
}
public function valid() {
return $this->_subscriptions && $this->_isStillValid;
$subscriptions = self::STATUS_SUBSCRIBED + self::STATUS_PSUBSCRIBED;
return $this->isFlagSet(self::STATUS_VALID)
&& ($this->_statusFlags & $subscriptions) > 0;
}
private function invalidate() {
$this->_isStillValid = false;
$this->_subscriptions = false;
$this->_statusFlags = 0x0000;
}
private function getValue() {
@@ -1379,6 +1533,15 @@ abstract class Predis_RedisServerProfile {
return new $profile();
}
public function supportsCommands(Array $commands) {
foreach ($commands as $command) {
if ($this->supportsCommand($command) === false) {
return false;
}
}
return true;
}
public function supportsCommand($command) {
return isset($this->_registeredCommands[$command]);
}
@@ -1656,6 +1819,13 @@ class Predis_RedisServer_v2_0 extends Predis_RedisServer_v1_2 {
class Predis_RedisServer_vNext extends Predis_RedisServer_v2_0 {
public function getVersion() { return '2.1'; }
public function getSupportedCommands() {
return array_merge(parent::getSupportedCommands(), array(
/* transactions */
'watch' => 'Predis_Commands_Watch',
'unwatch' => 'Predis_Commands_Unwatch',
));
}
}
/* ------------------------------------------------------------------------- */
@@ -1693,7 +1863,6 @@ class Predis_Pipeline_StandardExecutor implements Predis_Pipeline_IPipelineExecu
class Predis_Pipeline_SafeExecutor implements Predis_Pipeline_IPipelineExecutor {
public function execute(Predis_IConnection $connection, &$commands) {
$firstServerException = null;
$sizeofPipe = count($commands);
$values = array();
@@ -1720,9 +1889,6 @@ class Predis_Pipeline_SafeExecutor implements Predis_Pipeline_IPipelineExecutor
$values[] = $exception->toResponseError();
}
catch (Predis_CommunicationException $exception) {
if ($throwExceptions) {
throw $exception;
}
$toAdd = count($commands) - count($values);
$values = array_merge($values, array_fill(0, $toAdd, $exception));
break;
@@ -1952,6 +2118,10 @@ class Predis_Distribution_KetamaPureRing extends Predis_Distribution_HashRing {
/* ------------------------------------------------------------------------- */
class Predis_Shared_Utils {
public static function isCluster(Predis_IConnection $connection) {
return $connection instanceof Predis_ConnectionCluster;
}
public static function onCommunicationException(Predis_CommunicationException $exception) {
if ($exception->shouldResetConnection()) {
$connection = $exception->getConnection();
@@ -2376,8 +2546,9 @@ class Predis_Commands_ZSetUnionStore extends Predis_MultiBulkCommand {
$finalizedOpts = array();
if (isset($opts['WEIGHTS']) && is_array($opts['WEIGHTS'])) {
$finalizedOpts[] = 'WEIGHTS';
$finalizedOpts[] = $opts['WEIGHTS'][0];
$finalizedOpts[] = $opts['WEIGHTS'][1];
foreach ($opts['WEIGHTS'] as $weight) {
$finalizedOpts[] = $weight;
}
}
if (isset($opts['AGGREGATE'])) {
$finalizedOpts[] = 'AGGREGATE';
@@ -2392,20 +2563,42 @@ class Predis_Commands_ZSetIntersectionStore extends Predis_Commands_ZSetUnionSto
}
class Predis_Commands_ZSetRange extends Predis_MultiBulkCommand {
private $_withScores = false;
public function getCommandId() { return 'ZRANGE'; }
public function parseResponse($data) {
$arguments = $this->getArguments();
public function filterArguments(Array $arguments) {
if (count($arguments) === 4) {
if (strtolower($arguments[3]) === 'withscores') {
if ($data instanceof Iterator) {
return new Predis_Shared_MultiBulkResponseKVIterator($data);
}
$result = array();
for ($i = 0; $i < count($data); $i++) {
$result[] = array($data[$i], $data[++$i]);
}
return $result;
$lastType = gettype($arguments[3]);
if ($lastType === 'string' && strtolower($arguments[3]) === 'withscores') {
// used for compatibility with older versions
$arguments[3] = array('WITHSCORES' => true);
$lastType = 'array';
}
if ($lastType === 'array') {
$options = $this->prepareOptions(array_pop($arguments));
return array_merge($arguments, $options);
}
}
return $arguments;
}
protected function prepareOptions($options) {
$opts = array_change_key_case($options, CASE_UPPER);
$finalizedOpts = array();
if (isset($opts['WITHSCORES'])) {
$finalizedOpts[] = 'WITHSCORES';
$this->_withScores = true;
}
return $finalizedOpts;
}
public function parseResponse($data) {
if ($this->_withScores) {
if ($data instanceof Iterator) {
return new Predis_Shared_MultiBulkResponseKVIterator($data);
}
$result = array();
for ($i = 0; $i < count($data); $i++) {
$result[] = array($data[$i], $data[++$i]);
}
return $result;
}
return $data;
}
@@ -2417,6 +2610,17 @@ class Predis_Commands_ZSetReverseRange extends Predis_Commands_ZSetRange {
class Predis_Commands_ZSetRangeByScore extends Predis_Commands_ZSetRange {
public function getCommandId() { return 'ZRANGEBYSCORE'; }
protected function prepareOptions($options) {
$opts = array_change_key_case($options, CASE_UPPER);
$finalizedOpts = array();
if (isset($opts['LIMIT']) && is_array($opts['LIMIT'])) {
$limit = array_change_key_case($opts['LIMIT'], CASE_UPPER);
$finalizedOpts[] = 'LIMIT';
$finalizedOpts[] = isset($limit['OFFSET']) ? $limit['OFFSET'] : $limit[0];
$finalizedOpts[] = isset($limit['COUNT']) ? $limit['COUNT'] : $limit[1];
}
return array_merge($finalizedOpts, parent::prepareOptions($options));
}
}
class Predis_Commands_ZSetCount extends Predis_MultiBulkCommand {
@@ -2646,6 +2850,18 @@ class Predis_Commands_Publish extends Predis_MultiBulkCommand {
public function getCommandId() { return 'PUBLISH'; }
}
class Predis_Commands_Watch extends Predis_MultiBulkCommand {
public function canBeHashed() { return false; }
public function getCommandId() { return 'WATCH'; }
public function parseResponse($data) { return (bool) $data; }
}
class Predis_Commands_Unwatch extends Predis_MultiBulkCommand {
public function canBeHashed() { return false; }
public function getCommandId() { return 'UNWATCH'; }
public function parseResponse($data) { return (bool) $data; }
}
/* persistence control commands */
class Predis_Commands_Save extends Predis_MultiBulkCommand {
public function canBeHashed() { return false; }
+35
View File
@@ -1097,6 +1097,11 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
$this->redis->zrange('zset', 0, 2, 'withscores')
);
$this->assertEquals(
array(array('a', -10), array('b', 0), array('c', 10)),
$this->redis->zrange('zset', 0, 2, array('withscores' => true))
);
RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, p_anon("\$test", "
\$test->redis->set('foo', 'bar');
\$test->redis->zrange('foo', 0, -1);
@@ -1151,6 +1156,36 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
$this->redis->zrevrange('zset', 0, 2, 'withscores')
);
$this->assertEquals(
array(array('f', 30), array('e', 20), array('d', 20)),
$this->redis->zrevrange('zset', 0, 2, array('withscores' => true))
);
$this->assertEquals(
array(array('c', 10), array('d', 20), array('e', 20)),
$this->redis->zrangebyscore('zset', 10, 20, array('withscores' => true))
);
$this->assertEquals(
array('d', 'e'),
$this->redis->zrangebyscore('zset', 10, 20, array('limit' => array(1, 2)))
);
$this->assertEquals(
array('d', 'e'),
$this->redis->zrangebyscore('zset', 10, 20, array(
'limit' => array('offset' => 1, 'count' => 2)
))
);
$this->assertEquals(
array(array('d', 20), array('e', 20)),
$this->redis->zrangebyscore('zset', 10, 20, array(
'limit' => array(1, 2),
'withscores' => true,
))
);
RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, p_anon("\$test", "
\$test->redis->set('foo', 'bar');
\$test->redis->zrevrange('foo', 0, -1);