Add support for optional automatic retries upon aborted transactions.

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;

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);
     }
 }
 ?>