diff --git a/lib/Predis.php b/lib/Predis.php index b3f3eab4..5b2ec7fe 100644 --- a/lib/Predis.php +++ b/lib/Predis.php @@ -805,19 +805,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->_checkAndSet = false; - $this->_insideBlock = false; + $this->reset(); $this->_redisClient = $redisClient; $this->_options = $options ?: array(); - $this->_commands = array(); } private function checkCapabilities(Client $redisClient) { @@ -843,6 +839,14 @@ class MultiExecBlock { } } + private function reset() { + $this->_initialized = false; + $this->_discarded = false; + $this->_checkAndSet = false; + $this->_insideBlock = false; + $this->_commands = array(); + } + private function initialize() { if ($this->_initialized === false) { $options = &$this->_options; @@ -920,8 +924,7 @@ class MultiExecBlock { public function discard() { $this->_redisClient->discard(); - $this->_commands = array(); - $this->_initialized = false; + $this->reset(); $this->_discarded = true; return $this; } @@ -941,40 +944,51 @@ class MultiExecBlock { throw new \InvalidArgumentException('Argument passed must be a callable object'); } - $blockException = null; - $returnValues = array(); + $reply = null; + $returnValues = array(); + $attemptsLeft = isset($this->_options['retry']) ? (int)$this->_options['retry'] : 0; + do { + $blockException = null; - 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(); + 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(); + } + } + $this->setInsideBlock(false); + if ($blockException !== null) { + throw $blockException; } } - $this->setInsideBlock(false); - if ($blockException !== null) { - throw $blockException; + + if ($this->_initialized === false) { + 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(); + continue; + } + break; + } while ($attemptsLeft-- > 0); $execReply = $reply instanceof \Iterator ? iterator_to_array($reply) : $reply; $commands = &$this->_commands; diff --git a/test/PredisClientFeatures.php b/test/PredisClientFeatures.php index 17dd32d8..70fc4110 100644 --- a/test/PredisClientFeatures.php +++ b/test/PredisClientFeatures.php @@ -637,11 +637,60 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase { $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')); + } + function testMultiExecBlock_CheckAndSet_Discard() { $client = RC::getConnection(); $client->flushdb(); - $client->set('foo', 'bar'); + $client->set('foo', 'bar'); $options = array('watch' => 'foo', 'cas' => true); $replies = $client->multiExec($options, function($tx) { $tx->watch('foobar'); @@ -653,6 +702,26 @@ class PredisClientFeaturesTestSuite extends PHPUnit_Framework_TestCase { }); $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); } } ?>