Merge branch '16-predis-cluster'

Author: Willem Stuursma

.travis.yml

@@ -16,7 +16,7 @@ php:
 env: 
   global:
     - MEMCACHE_HOST=localhost
-    - REDIS_URIS=redis://localhost,redis://localhost:63791
+    - REDIS_URIS=redis://localhost:63791,redis://localhost:63792,redis://localhost:63793
     - MYSQL_DSN="mysql:host=localhost;dbname=test"
     - MYSQL_USER=travis
     - PGSQL_DSN="pgsql:host=localhost;dbname=test;"
@@ -36,6 +36,8 @@ matrix:
 before_install:
   - phpenv config-add tests/php-travis.ini
   - redis-server --port 63791 &
+  - redis-server --port 63792 &
+  - redis-server --port 63793 &
 
 install:
   - composer install --no-scripts --no-suggest --no-interaction

README.md

@@ -150,6 +150,9 @@ which uses the [`phpredis` extension](https://github.com/phpredis/phpredis).
 
 This implementation requires at least `phpredis-2.2.4`.
 
+If used with a cluster of Redis servers, acquiring and releasing locks will continue to function as 
+long as a majority of the servers still works.  
+
 Example:
 ```php
 $redis = new Redis();

tests/mutex/MutexConcurrencyTest.php

@@ -281,6 +281,7 @@ function ($uri) {
             }];
 
             $cases["PHPRedisMutex"] = [function ($timeout = 3) use ($uris) {
+                /** @var Redis[] $apis */
                 $apis = array_map(
                     function ($uri) {
                         $redis = new Redis();

tests/mutex/MutexTest.php

@@ -110,6 +110,7 @@ function ($uri) {
             }];
 
             $cases["PHPRedisMutex"] = [function () use ($uris) {
+                /** @var Redis[] $apis */
                 $apis = array_map(
                     function ($uri) {
                         $redis = new Redis();

tests/mutex/PHPRedisMutexTest.php

@@ -21,10 +21,10 @@
 class PHPRedisMutexTest extends \PHPUnit_Framework_TestCase
 {
     /**
-     * @var Redis The Redis API.
+     * @var Redis[]
      */
-    private $redis;
-    
+    private $connections = [];
+
     /**
      * @var PHPRedisMutex The SUT.
      */
@@ -34,31 +34,53 @@ protected function setUp()
     {
         parent::setUp();
 
-        $this->redis = new Redis();
-
         $uris = explode(",", getenv("REDIS_URIS") ?: "redis://localhost");
-        $uri  = parse_url($uris[0]);
-        if (!empty($uri["port"])) {
-            $this->redis->connect($uri["host"], $uri["port"]);
-        } else {
-            $this->redis->connect($uri["host"]);
+
+        foreach ($uris as $redisUri) {
+            $uri  = parse_url($redisUri);
+
+            $connection = new Redis();
+
+            if (!empty($uri["port"])) {
+                $connection->connect($uri["host"], $uri["port"]);
+            } else {
+                $connection->connect($uri["host"]);
+            }
+
+            $connection->flushAll(); // Clear any existing locks.
+
+            $this->connections[] = $connection;
         }
 
-        $this->redis->flushAll(); // Clear any existing locks.
+        $this->mutex = new PHPRedisMutex($this->connections, "test");
+    }
+
+    private function closeMajorityConnections()
+    {
+        $numberToClose = ceil(count($this->connections) / 2);
+
+        foreach (array_rand($this->connections, $numberToClose) as $keyToClose) {
+            $this->connections[$keyToClose]->close();
+        }
+    }
+
+    private function closeMinortyConnections()
+    {
+        $numberToClose = ceil(count($this->connections) / 2) - 1;
 
-        $this->mutex = new PHPRedisMutex([$this->redis], "test");
+        foreach ((array) array_rand($this->connections, $numberToClose) as $keyToClose) {
+            $this->connections[$keyToClose]->close();
+        }
     }
 
     /**
-     * Tests add() fails.
-     *
-     * @test
      * @expectedException \malkusch\lock\exception\LockAcquireException
      * @expectedExceptionCode \malkusch\lock\exception\MutexException::REDIS_NOT_ENOUGH_SERVERS
      */
     public function testAddFails()
     {
-        $this->redis->close();
+        $this->closeMajorityConnections();
+
         $this->mutex->synchronized(function () {
             $this->fail("Code execution is not expected");
         });
@@ -73,21 +95,40 @@ public function testAddFails()
     public function testEvalScriptFails()
     {
         $this->mutex->synchronized(function () {
-            $this->redis->close();
+            $this->closeMajorityConnections();
         });
     }
 
     /**
      * @param $serialization
      * @dataProvider dpSerializationModes
      */
-    public function testSyncronizedWorks($serialization)
+    public function testSynchronizedWorks($serialization)
     {
-        $this->redis->setOption(Redis::OPT_SERIALIZER, $serialization);
+        foreach ($this->connections as $connection) {
+            $connection->setOption(Redis::OPT_SERIALIZER, $serialization);
+        }
 
-        $this->mutex->synchronized(function () {
-            $this->assertTrue(true);
-        });
+        $this->assertNull($this->mutex->synchronized(function () {
+            return null;
+        }));
+    }
+
+    public function testResistantToPartialClusterFailuresForAcquiringLock()
+    {
+        $this->closeMinortyConnections();
+
+        $this->assertNull($this->mutex->synchronized(function () {
+            return null;
+        }));
+    }
+
+    public function testResistantToPartialClusterFailuresForReleasingLock()
+    {
+        $this->assertNull($this->mutex->synchronized(function () {
+            $this->closeMinortyConnections();
+            return null;
+        }));
     }
 
     public function dpSerializationModes()