Feature sentinel

.github/workflows/ci.yml

@@ -27,9 +27,10 @@ jobs:
           ini-file: development
       - run: composer install
       - run: docker run --net=host -d redis
-      - run: REDIS_URI=localhost:6379 vendor/bin/phpunit --coverage-text --coverage-clover=clover.xml
+        run: docker run --net=host -d -e REDIS_MASTER_HOST=localhost bitnami/redis-sentinel
+      - run: REDIS_URI=localhost:6379 REDIS_URIS=localhost:26379 REDIS_SENTINEL_MASTER=mymaster vendor/bin/phpunit --coverage-text --coverage-clover=clover.xml
         if: ${{ matrix.php >= 7.3 }}
-      - run: REDIS_URI=localhost:6379 vendor/bin/phpunit --coverage-text --coverage-clover=clover.xml -c phpunit.xml.legacy
+      - run: REDIS_URI=localhost:6379 REDIS_URIS=localhost:26379 REDIS_SENTINEL_MASTER=mymaster vendor/bin/phpunit --coverage-text --coverage-clover=clover.xml -c phpunit.xml.legacy
         if: ${{ matrix.php < 7.3 }}
       - name: Check 100% code coverage
         shell: php {0}

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 2.7.0 (TBA)
+
+*   Feature: Support Redis Sentinel auto master discovery (alpha).
+    (@sartor)
+
 ## 2.6.0 (2022-05-09)
 
 *   Feature: Support PHP 8.1 release.

src/SentinelClient.php

@@ -0,0 +1,93 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Clue\React\Redis;
+
+use Clue\React\Redis\Io\Factory;
+use Clue\React\Redis\Io\StreamingClient;
+use React\EventLoop\Loop;
+use React\EventLoop\LoopInterface;
+use React\Promise\PromiseInterface;
+use React\Socket\ConnectorInterface;
+use function React\Promise\reject;
+use function React\Promise\resolve;
+
+/**
+ * Client for receiving Sentinel master url
+ */
+class SentinelClient
+{
+    /** @var array<string> */
+    private $urls;
+
+    /** @var string */
+    private $masterName;
+
+    /** @var Factory */
+    private $factory;
+
+    /** @var StreamingClient */
+    private $masterClient;
+
+    /**
+     * @param array $urls list of sentinel addresses
+     * @param string $masterName sentinel master name
+     * @param ?ConnectorInterface $connector
+     * @param ?LoopInterface $loop
+     */
+    public function __construct(array $urls, string $masterName, ConnectorInterface $connector = null, LoopInterface $loop = null)
+    {
+        $this->urls = $urls;
+        $this->masterName = $masterName;
+        $this->factory = new Factory($loop ?: Loop::get(), $connector);
+    }
+
+    public function masterAddress(): PromiseInterface
+    {
+        $chain = reject(new \RuntimeException('Initial reject promise'));
+        foreach ($this->urls as $url) {
+            $chain = $chain->then(function ($masterUrl) {
+                return $masterUrl;
+            }, function () use ($url) {
+                return $this->onError($url);
+            });
+        }
+
+        return $chain;
+    }
+
+    public function masterConnection(string $masterUriPath = '', array $masterUriParams = []): PromiseInterface
+    {
+        if (isset($this->masterClient)) {
+            return resolve($this->masterClient);
+        }
+
+        return $this
+            ->masterAddress()
+            ->then(function (string $masterUrl) use ($masterUriPath, $masterUriParams) {
+                $query = $masterUriParams ? '?' . http_build_query($masterUriParams) : '';
+                return $this->factory->createClient($masterUrl . $masterUriPath . $query);
+            })
+            ->then(function (StreamingClient $client) {
+                $this->masterClient = $client;
+                return $client->role();
+            })
+            ->then(function (array $role) {
+                $isRealMaster = ($role[0] ?? '') === 'master';
+                return $isRealMaster ? $this->masterClient : reject(new \RuntimeException("Invalid master role: {$role[0]}"));
+            });
+    }
+
+    private function onError(string $nextUrl): PromiseInterface
+    {
+        return $this->factory
+            ->createClient($nextUrl)
+            ->then(function (StreamingClient $client) {
+                return $client->sentinel('get-master-addr-by-name', $this->masterName);
+            })
+            ->then(function (array $response) {
+                return $response[0] . ':' . $response[1]; // ip:port
+            });
+    }
+}

tests/SentinelClientTest.php

@@ -0,0 +1,83 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Clue\Tests\React\Redis;
+
+use Clue\React\Redis\Io\StreamingClient;
+use Clue\React\Redis\SentinelClient;
+use React\EventLoop\StreamSelectLoop;
+use function Clue\React\Block\await;
+
+class SentinelClientTest extends TestCase
+{
+    /** @var StreamSelectLoop */
+    private $loop;
+
+    /** @var string */
+    private $masterUri;
+
+    /** @var array */
+    private $uris;
+
+    /** @var string */
+    private $masterName;
+
+    public function setUp(): void
+    {
+        $this->masterUri = getenv('REDIS_URI') ?: '';
+        if ($this->masterUri === '') {
+            $this->markTestSkipped('No REDIS_URI environment variable given for Sentinel tests');
+        }
+
+        $uris = getenv('REDIS_URIS') ?: '';
+        if ($uris === '') {
+            $this->markTestSkipped('No REDIS_URIS environment variable given for Sentinel tests');
+        }
+        $this->uris = array_map('trim', explode(',', $uris));
+
+        $this->masterName = getenv('REDIS_SENTINEL_MASTER') ?: '';
+        if ($this->masterName === '') {
+            $this->markTestSkipped('No REDIS_SENTINEL_MASTER environment variable given for Sentinel tests');
+        }
+
+        $this->loop = new StreamSelectLoop();
+    }
+
+    public function testMasterAddress()
+    {
+        $redis = new SentinelClient($this->uris, $this->masterName, null, $this->loop);
+        $masterAddressPromise = $redis->masterAddress();
+        $masterAddress = await($masterAddressPromise, $this->loop);
+        $this->assertEquals(str_replace('localhost', '127.0.0.1', $this->masterUri), $masterAddress);
+    }
+
+    public function testMasterConnectionWithParams()
+    {
+        $redis = new SentinelClient($this->uris, $this->masterName, null, $this->loop);
+        $masterConnectionPromise = $redis->masterConnection('/1', ['timeout' => 0.5]);
+        $masterConnection = await($masterConnectionPromise, $this->loop);
+        $this->assertInstanceOf(StreamingClient::class, $masterConnection);
+
+        $pong = await($masterConnection->ping(), $this->loop);
+        $this->assertEquals('PONG', $pong);
+    }
+
+    public function testConnectionFail()
+    {
+        $redis = new SentinelClient(['128.128.0.1:26379?timeout=0.1'], $this->masterName, null, $this->loop);
+        $masterConnectionPromise = $redis->masterConnection();
+
+        $this->expectException(\RuntimeException::class);
+        $this->expectExceptionMessage('Connection to redis://128.128.0.1:26379?timeout=0.1 timed out after 0.1 seconds');
+        await($masterConnectionPromise, $this->loop);
+    }
+
+    public function testConnectionSkipInvalid()
+    {
+        $redis = new SentinelClient(array_merge(['128.128.0.1:26379?timeout=0.1'], $this->uris), $this->masterName, null, $this->loop);
+        $masterConnectionPromise = $redis->masterConnection('/1', ['timeout' => 5]);
+        $masterConnection = await($masterConnectionPromise, $this->loop);
+        $this->assertInstanceOf(StreamingClient::class, $masterConnection);
+    }
+}