geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-09-16 05:17:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Daniele Alessandri
2011-01-01 16:27:21 +01:00
parent ccb48a1857
commit ddb5109805
6 changed files with 359 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG
+11
View File
@@ -1,3 +1,14 @@
v0.6.3 (201x-xx-xx)
* 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.
README.markdown
+3 -2
View File
@@ -20,6 +20,7 @@ to be implemented soon in Predis.
- Full support for Redis 2.0. 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)
);
));
$pipe = $redis->pipeline();
for ($i = 0; $i < 1000; $i++) {
lib/Predis.php
+144 -77
View File
@@ -812,18 +812,15 @@ class Predis_CommandPipeline {
}
class Predis_MultiExecBlock {
private $_initialized, $_discarded, $_insideBlock;
private $_initialized, $_discarded, $_insideBlock, $_checkAndSet;
private $_redisClient, $_options, $_commands;
private $_supportsWatch;
public function __construct(Predis_Client $redisClient, Array $options = null) {
$this->checkCapabilities($redisClient);
$this->_initialized = false;
$this->_discarded = false;
$this->_insideBlock = false;
$this->_options = isset($options) ? $options : array();
$this->_redisClient = $redisClient;
$this->_options = isset($options) ? $options : array();
$this->_commands = array();
$this->reset();
}
private function checkCapabilities(Predis_Client $redisClient) {
@@ -849,127 +846,169 @@ class Predis_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 (!isset($response->queued)) {
$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) {
if ($this->_initialized && !$this->_checkAndSet) {
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;
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() {
$this->isWatchSupported();
$this->_redisClient->unwatch();
return $this;
}
public function discard() {
$this->_redisClient->discard();
$this->_commands = array();
$this->_initialized = false;
$this->_discarded = true;
$this->reset();
$this->_discarded = true;
return $this;
}
public function exec() {
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 Predis_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 (Predis_CommunicationException $exception) {
$blockException = $exception;
}
catch (Predis_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 (Predis_CommunicationException $exception) {
$blockException = $exception;
}
catch (Predis_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) {
if ($attemptsLeft === 0) {
throw new Predis_AbortedMultiExec(
'The current transaction has been aborted by the server'
);
}
$this->reset();
continue;
}
break;
} while ($attemptsLeft-- > 0);
$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);
$commands = &$this->_commands;
if ($sizeofReplies !== count($commands)) {
$this->malformedServerResponse('Unexpected number of responses for a Predis_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])
@@ -980,6 +1019,17 @@ class Predis_MultiExecBlock {
return $returnValues;
}
private function malformedServerResponse($message) {
// 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.
Predis_Shared_Utils::onCommunicationException(
new Predis_MalformedServerResponse(
$this->_redisClient->getConnection(), $message
)
);
}
}
class Predis_PubSubContext implements Iterator {
@@ -1825,6 +1875,10 @@ class Predis_RedisServer_vNext extends Predis_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',
@@ -1833,6 +1887,7 @@ class Predis_RedisServer_vNext extends Predis_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',
@@ -2355,10 +2410,22 @@ class Predis_Commands_Append extends Predis_MultiBulkCommand {
public function getCommandId() { return 'APPEND'; }
}
class Predis_Commands_SetRange extends Predis_MultiBulkCommand {
public function getCommandId() { return 'SETRANGE'; }
}
class Predis_Commands_Substr extends Predis_MultiBulkCommand {
public function getCommandId() { return 'SUBSTR'; }
}
class Predis_Commands_SetBit extends Predis_MultiBulkCommand {
public function getCommandId() { return 'SETBIT'; }
}
class Predis_Commands_GetBit extends Predis_MultiBulkCommand {
public function getCommandId() { return 'GETBIT'; }
}
class Predis_Commands_Strlen extends Predis_MultiBulkCommand {
public function getCommandId() { return 'STRLEN'; }
}
@@ -2462,8 +2529,8 @@ class Predis_Commands_ListPopLastPushHead extends Predis_MultiBulkCommand {
public function getCommandId() { return 'RPOPLPUSH'; }
}
class Predis_Commands_ListPopLastPushHeadBulk extends Predis_MultiBulkCommand {
public function getCommandId() { return 'RPOPLPUSH'; }
class Predis_Commands_ListPopLastPushHeadBlocking extends Predis_MultiBulkCommand {
public function getCommandId() { return 'BRPOPLPUSH'; }
}
class Predis_Commands_ListPopFirst extends Predis_MultiBulkCommand {
test/PredisClientFeatures.php
+94
View File
@@ -552,12 +552,29 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase {
$this->assertEquals('bar', $replies[2]);
}
/**
* @expectedException Predis_ClientException
*/
function testMultiExecBlock_CannotMixFluentInterfaceAndAnonymousBlock() {
$emptyBlock = p_anon("\$tx", "");
$tx = RC::getConnection()->multiExec()->get('foo')->execute($emptyBlock);
}
function testMultiExecBlock_EmptyCallableBlock() {
$client = RC::getConnection();
$client->flushdb();
$replies = $client->multiExec(p_anon("\$multi", ""));
$this->assertEquals(0, count($replies));
$options = array('cas' => true);
$replies = $client->multiExec($options, p_anon("\$multi", ""));
$this->assertEquals(0, count($replies));
$options = array('cas' => true);
$replies = $client->multiExec($options, p_anon("\$multi", "
\$multi->multi();
"));
$this->assertEquals(0, count($replies));
}
@@ -644,5 +661,82 @@ 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, p_anon("\$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();
$client1->flushdb();
$retry = 3;
$thrownException = null;
try {
$options = array('watch' => 'sentinel', 'retry' => $retry);
$client1->multiExec($options, p_anon("\$tx", "
\$tx->set('sentinel', 'client1');
\$tx->get('sentinel');
\$client2 = RC::getConnection(true);
\$client2->incr('attempts');
\$client2->set('sentinel', 'client2');
"));
}
catch (Predis_AbortedMultiExec $exception) {
$thrownException = $exception;
}
$this->assertType('Predis_AbortedMultiExec', $thrownException);
$this->assertEquals('The current transaction has been aborted by the server', $thrownException->getMessage());
$this->assertEquals('client2', $client1->get('sentinel'));
$this->assertEquals($retry + 1, $client1->get('attempts'));
$client1->del('attempts', 'sentinel');
$thrownException = null;
try {
$options = array(
'watch' => 'sentinel',
'cas' => true,
'retry' => $retry
);
$client1->multiExec($options, p_anon("\$tx", "
\$tx->incr('attempts');
\$tx->multi();
\$tx->set('sentinel', 'client1');
\$tx->get('sentinel');
\$client2 = RC::getConnection(true);
\$client2->set('sentinel', 'client2');
"));
}
catch (Predis_AbortedMultiExec $exception) {
$thrownException = $exception;
}
$this->assertType('Predis_AbortedMultiExec', $thrownException);
$this->assertEquals('The current transaction has been aborted by the server', $thrownException->getMessage());
$this->assertEquals('client2', $client1->get('sentinel'));
$this->assertEquals($retry + 1, $client1->get('attempts'));
}
}
?>
test/PredisShared.php
+3
View File
@@ -24,11 +24,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
+104 -2
View File
@@ -224,6 +224,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, p_anon("\$test", "
\$test->redis->setrange('var', -1, 'bogus');
"));
RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, p_anon("\$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));
@@ -253,6 +275,60 @@ 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, p_anon("\$test", "
\$test->redis->setbit('binary', -1, 1);
"));
RC::testForServerException($this, RC::EXCEPTION_BIT_OFFSET, p_anon("\$test", "
\$test->redis->setbit('binary', 'invalid', 1);
"));
RC::testForServerException($this, RC::EXCEPTION_BIT_VALUE, p_anon("\$test", "
\$test->redis->setbit('binary', 15, 255);
"));
RC::testForServerException($this, RC::EXCEPTION_BIT_VALUE, p_anon("\$test", "
\$test->redis->setbit('binary', 15, 'invalid');
"));
RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, p_anon("\$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 */
@@ -347,7 +423,8 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
sleep(2);
$this->assertFalse($this->redis->exists('hoge'));
RC::testForServerException($this, RC::EXCEPTION_VALUE_NOT_INT, p_anon("\$test", "
// TODO: do not check the error message RC::EXCEPTION_VALUE_NOT_INT for now
RC::testForServerException($this, null, p_anon("\$test", "
\$test->redis->setex('hoge', 2.5, 'piyo');
"));
RC::testForServerException($this, RC::EXCEPTION_SETEX_TTL, p_anon("\$test", "
@@ -722,6 +799,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, p_anon("\$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());
@@ -734,7 +836,7 @@ class RedisCommandTestSuite extends PHPUnit_Framework_TestCase {
RC::testForServerException($this, RC::EXCEPTION_WRONG_TYPE, p_anon("\$test", "
\$test->redis->set('foo', 'bar');
\$test->redis->lset('foo', 0, 0);
\$test->redis->linsert('foo', 'before', 0, 0);
"));
}