RedisClientTest.php

<?php

namespace Clue\Tests\React\Redis;

use Clue\React\Redis\RedisClient;
use Clue\React\Redis\Io\Factory;
use Clue\React\Redis\Io\StreamingClient;
use PHPUnit\Framework\MockObject\MockObject;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\EventLoop\TimerInterface;
use React\Promise\Deferred;
use React\Promise\Promise;

class RedisClientTest extends TestCase
{
    /** @var MockObject */
    private $factory;

    /** @var RedisClient */
    private $redis;

    /** @var Loopinterface */
    public static $loop;

    public static function setUpBeforeClass(): void
    {
        self::$loop = Loop::get();
    }

    public static function tearDownAfterClass(): void
    {
        Loop::set(self::$loop);
    }

    public function setUp(): void
    {
        $this->factory = $this->createMock(Factory::class);

        $this->redis = new RedisClient('localhost');

        $ref = new \ReflectionProperty($this->redis, 'factory');
        if (PHP_VERSION_ID < 80100) {
            $ref->setAccessible(true);
        }
        $ref->setValue($this->redis, $this->factory);
    }

    public static function provideInvalidUris(): \Generator
    {
        yield [
            '',
            'redis://'
        ];
        yield [
            'localhost:100000',
            'redis://localhost:100000'
        ];
        yield [
            'tcp://localhost',
            'tcp://localhost'
        ];
        yield [
            'redis://',
            'redis://'
        ];
        yield [
            'redis+unix://',
            'redis+unix://'
        ];
        yield [
            'user@localhost:100000',
            'redis://user@localhost:100000'
        ];
        yield [
            ':pass@localhost:100000',
            'redis://:***@localhost:100000'
        ];
        yield [
            'user:@localhost:100000',
            'redis://user:***@localhost:100000'
        ];
        yield [
            'user:pass@localhost:100000',
            'redis://user:***@localhost:100000'
        ];
        yield [
            'localhost:100000?password=secret',
            'redis://localhost:100000?password=***'
        ];
        yield [
            'user@',
            'redis://user@'
        ];
        yield [
            'user:pass@',
            'redis://user:***@'
        ];
    }

    /** @dataProvider provideInvalidUris() */
    public function testCtorWithInvalidUriThrows(string $uri, string $message): void
    {
        $this->expectException(\InvalidArgumentException::class);
        $this->expectExceptionMessage('Invalid Redis URI "' . $message . '" (EINVAL)');
        $this->expectExceptionCode(defined('SOCKET_EINVAL') ? SOCKET_EINVAL : 22);
        new RedisClient($uri);
    }

    public function testPingWillCreateUnderlyingClientAndReturnPendingPromise(): void
    {
        $promise = new Promise(function () { });
        $this->factory->expects($this->once())->method('createClient')->with('redis://localhost')->willReturn($promise);

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();

        $promise->then($this->expectCallableNever());
    }

    public function testPingWithAuthWillCreateUnderlyingClientWithAuthAndReturnPendingPromise(): void
    {
        $this->redis = new RedisClient('user:pass@localhost');
        $ref = new \ReflectionProperty($this->redis, 'factory');
        if (PHP_VERSION_ID < 80100) {
            $ref->setAccessible(true);
        }
        $ref->setValue($this->redis, $this->factory);

        $promise = new Promise(function () { });
        $this->factory->expects($this->once())->method('createClient')->with('redis://user:pass@localhost')->willReturn($promise);

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();

        $promise->then($this->expectCallableNever());
    }

    public function testPingWithUnixUriWillCreateUnderlyingClientAndReturnPendingPromise(): void
    {
        $this->redis = new RedisClient('redis+unix:///tmp/redis.sock');
        $ref = new \ReflectionProperty($this->redis, 'factory');
        if (PHP_VERSION_ID < 80100) {
            $ref->setAccessible(true);
        }
        $ref->setValue($this->redis, $this->factory);

        $promise = new Promise(function () { });
        $this->factory->expects($this->once())->method('createClient')->with('redis+unix:///tmp/redis.sock')->willReturn($promise);

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();

        $promise->then($this->expectCallableNever());
    }

    public function testPingTwiceWillCreateOnceUnderlyingClient(): void
    {
        $promise = new Promise(function () { });
        $this->factory->expects($this->once())->method('createClient')->willReturn($promise);

        $this->redis->ping();
        $this->redis->ping();
    }

    public function testPingWillResolveWhenUnderlyingClientResolvesPingAndStartIdleTimer(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\resolve('PONG'));

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->once())->method('addTimer')->with(0.001, $this->anything());
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();
        $deferred->resolve($client);

        $promise->then($this->expectCallableOnceWith('PONG'));
    }

    public function testPingWillResolveWhenUnderlyingClientResolvesPingAndStartIdleTimerWithIdleTimeFromQueryParam(): void
    {
        $this->redis = new RedisClient('localhost?idle=10');

        $ref = new \ReflectionProperty($this->redis, 'factory');
        if (PHP_VERSION_ID < 80100) {
            $ref->setAccessible(true);
        }
        $ref->setValue($this->redis, $this->factory);

        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\resolve('PONG'));

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->once())->method('addTimer')->with(10.0, $this->anything());
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();
        $deferred->resolve($client);

        $promise->then($this->expectCallableOnceWith('PONG'));
    }

    public function testPingWillResolveWhenUnderlyingClientResolvesPingAndNotStartIdleTimerWhenIdleParamIsNegative(): void
    {
        $this->redis = new RedisClient('localhost?idle=-1');

        $ref = new \ReflectionProperty($this->redis, 'factory');
        if (PHP_VERSION_ID < 80100) {
            $ref->setAccessible(true);
        }
        $ref->setValue($this->redis, $this->factory);

        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\resolve('PONG'));

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();
        $deferred->resolve($client);

        $promise->then($this->expectCallableOnceWith('PONG'));
    }

    public function testPingWillRejectWhenUnderlyingClientRejectsPingAndStartIdleTimer(): void
    {
        $error = new \RuntimeException();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\reject($error));

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->once())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->ping();
        $deferred->resolve($client);

        $promise->then(null, $this->expectCallableOnceWith($error));
    }

    public function testPingWillRejectAndNotEmitErrorOrCloseWhenFactoryRejectsUnderlyingClient(): void
    {
        $error = new \RuntimeException();

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->on('error', $this->expectCallableNever());
        $this->redis->on('close', $this->expectCallableNever());

        $promise = $this->redis->ping();
        $deferred->reject($error);

        $promise->then(null, $this->expectCallableOnceWith($error));
    }

    public function testPingAfterPreviousFactoryRejectsUnderlyingClientWillCreateNewUnderlyingConnection(): void
    {
        $error = new \RuntimeException();

        $deferred = new Deferred();
        $this->factory->expects($this->exactly(2))->method('createClient')->willReturnOnConsecutiveCalls(
            $deferred->promise(),
            new Promise(function () { })
        );

        $promise = $this->redis->ping();

        $promise->then(null, $this->expectCallableOnce()); // avoid reporting unhandled rejection

        $deferred->reject($error);

        $this->redis->ping();
    }

    public function testPingAfterPreviousUnderlyingClientAlreadyClosedWillCreateNewUnderlyingConnection(): void
    {
        $closeHandler = null;
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\resolve('PONG'));
        $client->expects($this->any())->method('on')->withConsecutive(
            ['close', $this->callback(function ($arg) use (&$closeHandler) {
                $closeHandler = $arg;
                return true;
            })]
        );

        $this->factory->expects($this->exactly(2))->method('createClient')->willReturnOnConsecutiveCalls(
            \React\Promise\resolve($client),
            new Promise(function () { })
        );

        $this->redis->ping();
        $this->assertTrue(is_callable($closeHandler));
        $closeHandler();

        $this->redis->ping();
    }

    public function testPingAfterCloseWillRejectWithoutCreatingUnderlyingConnection(): void
    {
        $this->factory->expects($this->never())->method('createClient');

        $this->redis->close();
        $promise = $this->redis->ping();

        $promise->then(null, $this->expectCallableOnceWith(
            $this->logicalAnd(
                $this->isInstanceOf(\RuntimeException::class),
                $this->callback(function (\RuntimeException $e) {
                    return $e->getMessage() === 'Connection closed (ENOTCONN)';
                }),
                $this->callback(function (\RuntimeException $e) {
                    return $e->getCode() === (defined('SOCKET_ENOTCONN') ? SOCKET_ENOTCONN : 107);
                })
            )
        ));
    }

    public function testPingAfterPingWillNotStartIdleTimerWhenFirstPingResolves(): void
    {
        $deferred = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->exactly(2))->method('callAsync')->willReturnOnConsecutiveCalls(
            $deferred->promise(),
            new Promise(function () { })
        );

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->ping();
        $this->redis->ping();
        $deferred->resolve(null);
    }

    public function testPingAfterPingWillStartAndCancelIdleTimerWhenSecondPingStartsAfterFirstResolves(): void
    {
        $deferred = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->exactly(2))->method('callAsync')->willReturnOnConsecutiveCalls(
            $deferred->promise(),
            new Promise(function () { })
        );

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $timer = $this->createMock(TimerInterface::class);
        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
        $loop->expects($this->once())->method('cancelTimer')->with($timer);
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->ping();
        $deferred->resolve(null);
        $this->redis->ping();
    }

    public function testPingFollowedByIdleTimerWillCloseUnderlyingConnectionWithoutCloseEvent(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn(\React\Promise\resolve(null));
        $client->expects($this->once())->method('close');

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $timeout = null;
        $timer = $this->createMock(TimerInterface::class);
        $loop->expects($this->once())->method('addTimer')->with($this->anything(), $this->callback(function ($cb) use (&$timeout) {
            $timeout = $cb;
            return true;
        }))->willReturn($timer);
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->on('close', $this->expectCallableNever());

        $this->redis->ping();

        $this->assertNotNull($timeout);
        $timeout();
    }

    public function testBlpopWillForwardArgumentsAsStringToUnderlyingClient(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('BLPOP', 'foo', 'bar', '10.0')->willReturn(new Promise(function () { }));

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->callAsync('BLPOP', 'foo', 'bar', 10.0);
    }

    public function testCallAsyncWillForwardArgumentsAsStringToUnderlyingClient(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ZCOUNT', 'foo', '-INF', 'INF')->willReturn(new Promise(function () { }));

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->callAsync('ZCOUNT', 'foo', -INF, INF);
    }

    public function testSetWithInvalidBoolArgumentThrows(): void
    {
        $this->expectException(\TypeError::class);
        $this->expectExceptionMessage('Argument must be of type string|int|float, boolean given');
        $this->redis->set('foo', true);
    }

    public function testSetWithInvalidObjectArgumentThrows(): void
    {
        $this->expectException(\TypeError::class);
        $this->expectExceptionMessage('Argument must be of type string|int|float, stdClass given');
        $this->redis->set('foo', (object) []);
    }

    public function testCallAsyncWithInvalidBoolArgumentThrows(): void
    {
        $this->expectException(\TypeError::class);
        $this->expectExceptionMessage('Argument must be of type string|int|float, boolean given');
        $this->redis->callAsync('SET', 'foo', true); // @phpstan-ignore-line
    }

    public function testCallAsyncWithInvalidObjectArgumentThrows(): void
    {
        $this->expectException(\TypeError::class);
        $this->expectExceptionMessage('Argument must be of type string|int|float, stdClass given');
        $this->redis->callAsync('SET', 'foo', (object) []); // @phpstan-ignore-line
    }

    public function testCloseWillEmitCloseEventWithoutCreatingUnderlyingClient(): void
    {
        $this->factory->expects($this->never())->method('createClient');

        $this->redis->on('close', $this->expectCallableOnce());

        $this->redis->close();
    }

    public function testCloseTwiceWillEmitCloseEventOnce(): void
    {
        $this->redis->on('close', $this->expectCallableOnce());

        $this->redis->close();
        $this->redis->close();
    }

    public function testCloseAfterPingWillCancelUnderlyingClientConnectionWhenStillPending(): void
    {
        $promise = new Promise(function () { }, $this->expectCallableOnce());
        $this->factory->expects($this->once())->method('createClient')->willReturn($promise);

        $this->redis->ping();
        $this->redis->close();
    }

    public function testCloseAfterPingWillEmitCloseWithoutErrorWhenUnderlyingClientConnectionThrowsDueToCancellation(): void
    {
        $promise = new Promise(function () { }, function () {
            throw new \RuntimeException('Discarded');
        });
        $this->factory->expects($this->once())->method('createClient')->willReturn($promise);

        $this->redis->on('error', $this->expectCallableNever());
        $this->redis->on('close', $this->expectCallableOnce());

        $promise = $this->redis->ping();

        $promise->then(null, $this->expectCallableOnce()); // avoid reporting unhandled rejection

        $this->redis->close();
    }

    public function testCloseAfterPingWillCloseUnderlyingClientConnectionWhenAlreadyResolved(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn(\React\Promise\resolve(null));
        $client->expects($this->once())->method('close');

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->ping();
        $deferred->resolve($client);
        $this->redis->close();
    }

    public function testCloseAfterPingWillCancelIdleTimerWhenPingIsAlreadyResolved(): void
    {
        $deferred = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn($deferred->promise());
        $client->expects($this->once())->method('close');

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $timer = $this->createMock(TimerInterface::class);
        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
        $loop->expects($this->once())->method('cancelTimer')->with($timer);
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->ping();
        $deferred->resolve(null);
        $this->redis->close();
    }

    public function testCloseAfterPingRejectsWillEmitClose(): void
    {
        $deferred = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn($deferred->promise());
        $client->expects($this->once())->method('close')->willReturnCallback(function () use ($client) {
            assert($client instanceof StreamingClient);
            $client->emit('close');
        });

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $timer = $this->createMock(TimerInterface::class);
        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
        $loop->expects($this->once())->method('cancelTimer')->with($timer);
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->ping()->then(null, function () {
            $this->redis->close();
        });
        $this->redis->on('close', $this->expectCallableOnce());
        $deferred->reject(new \RuntimeException());
    }

    public function testEndWillCloseClientIfUnderlyingConnectionIsNotPending(): void
    {
        $this->redis->on('close', $this->expectCallableOnce());
        $this->redis->end();
    }

    public function testEndAfterPingWillEndUnderlyingClient(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\resolve('PONG'));
        $client->expects($this->once())->method('end');

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->ping();
        $deferred->resolve($client);
        $this->redis->end();
    }

    public function testEndAfterPingWillCloseClientWhenUnderlyingClientEmitsClose(): void
    {
        $closeHandler = null;
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('ping')->willReturn(\React\Promise\resolve('PONG'));
        $client->expects($this->once())->method('end');
        $client->expects($this->any())->method('on')->willReturnCallback(function ($event, $callback) use (&$closeHandler) {
            if ($event === 'close') {
                $closeHandler = $callback;
            }
        });

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->ping();
        $deferred->resolve($client);

        $this->redis->on('close', $this->expectCallableOnce());
        $this->redis->end();

        $this->assertTrue(is_callable($closeHandler));
        $closeHandler();
    }

    public function testEmitsNoErrorEventWhenUnderlyingClientEmitsError(): void
    {
        $error = new \RuntimeException();

        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn(\React\Promise\resolve(null));

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->ping();
        $deferred->resolve($client);

        $this->redis->on('error', $this->expectCallableNever());
        assert($client instanceof StreamingClient);
        $client->emit('error', [$error]);
    }

    public function testEmitsNoCloseEventWhenUnderlyingClientEmitsClose(): void
    {
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn(\React\Promise\resolve(null));

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->ping();
        $deferred->resolve($client);

        $this->redis->on('close', $this->expectCallableNever());
        assert($client instanceof StreamingClient);
        $client->emit('close');
    }

    public function testEmitsNoCloseEventButWillCancelIdleTimerWhenUnderlyingConnectionEmitsCloseAfterPingIsAlreadyResolved(): void
    {
        $closeHandler = null;
        $client = $this->createMock(StreamingClient::class);
        $deferred = new Deferred();
        $client->expects($this->once())->method('callAsync')->willReturn($deferred->promise());
        $client->expects($this->any())->method('on')->withConsecutive(
            ['close', $this->callback(function ($arg) use (&$closeHandler) {
                $closeHandler = $arg;
                return true;
            })]
        );

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $timer = $this->createMock(TimerInterface::class);
        $loop->expects($this->once())->method('addTimer')->willReturn($timer);
        $loop->expects($this->once())->method('cancelTimer')->with($timer);
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $this->redis->on('close', $this->expectCallableNever());

        $this->redis->ping();
        $deferred->resolve(null);

        $this->assertTrue(is_callable($closeHandler));
        $closeHandler();
    }

    public function testEmitsMessageEventWhenUnderlyingClientEmitsMessageForPubSubChannel(): void
    {
        $messageHandler = null;
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->willReturn(\React\Promise\resolve(null));
        $client->expects($this->any())->method('on')->willReturnCallback(function ($event, $callback) use (&$messageHandler) {
            if ($event === 'message') {
                $messageHandler = $callback;
            }
        });

        $deferred = new Deferred();
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $this->redis->subscribe('foo');
        $deferred->resolve($client);

        $this->redis->on('message', $this->expectCallableOnce());
        $this->assertTrue(is_callable($messageHandler));
        $messageHandler('foo', 'bar');
    }

    public function testEmitsUnsubscribeAndPunsubscribeEventsWhenUnderlyingClientClosesWhileUsingPubSubChannel(): void
    {
        $allHandler = null;
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->exactly(6))->method('callAsync')->willReturn(\React\Promise\resolve(null));
        $client->expects($this->any())->method('on')->willReturnCallback(function ($event, $callback) use (&$allHandler) {
            if (!isset($allHandler[$event])) {
                $allHandler[$event] = $callback;
            }
        });

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $this->redis->subscribe('foo');
        assert(isset($allHandler['subscribe']) && is_callable($allHandler['subscribe']));
        $allHandler['subscribe']('foo', 1);

        $this->redis->subscribe('bar');
        $allHandler['subscribe']('bar', 2);

        $this->redis->unsubscribe('bar');
        assert(isset($allHandler['unsubscribe']) && is_callable($allHandler['unsubscribe']));
        $allHandler['unsubscribe']('bar', 1);

        $this->redis->psubscribe('foo*');
        assert(isset($allHandler['psubscribe']) && is_callable($allHandler['psubscribe']));
        $allHandler['psubscribe']('foo*', 1);

        $this->redis->psubscribe('bar*');
        $allHandler['psubscribe']('bar*', 2);

        $this->redis->punsubscribe('bar*');
        assert(isset($allHandler['punsubscribe']) && is_callable($allHandler['punsubscribe']));
        $allHandler['punsubscribe']('bar*', 1);

        $this->redis->on('unsubscribe', $this->expectCallableOnce());
        $this->redis->on('punsubscribe', $this->expectCallableOnce());

        $this->assertTrue(is_callable($allHandler['close']));
        $allHandler['close']();
    }

    public function testSubscribeWillResolveWhenUnderlyingClientResolvesSubscribeAndNotStartIdleTimerWithIdleDueToSubscription(): void
    {
        $subscribeHandler = null;
        $deferred = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('subscribe')->willReturn($deferred->promise());
        $client->expects($this->any())->method('on')->willReturnCallback(function ($event, $callback) use (&$subscribeHandler) {
            if ($event === 'subscribe' && $subscribeHandler === null) {
                $subscribeHandler = $callback;
            }
        });

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->subscribe('foo');
        $this->assertTrue(is_callable($subscribeHandler));
        $subscribeHandler('foo', 1);
        $deferred->resolve(['subscribe', 'foo', 1]);

        $promise->then($this->expectCallableOnceWith(['subscribe', 'foo', 1]));
    }

    public function testUnsubscribeAfterSubscribeWillResolveWhenUnderlyingClientResolvesUnsubscribeAndStartIdleTimerWhenSubscriptionStopped(): void
    {
        $subscribeHandler = null;
        $unsubscribeHandler = null;
        $deferredSubscribe = new Deferred();
        $deferredUnsubscribe = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->exactly(2))->method('callAsync')->willReturnOnConsecutiveCalls($deferredSubscribe->promise(), $deferredUnsubscribe->promise());
        $client->expects($this->any())->method('on')->willReturnCallback(function ($event, $callback) use (&$subscribeHandler, &$unsubscribeHandler) {
            if ($event === 'subscribe' && $subscribeHandler === null) {
                $subscribeHandler = $callback;
            }
            if ($event === 'unsubscribe' && $unsubscribeHandler === null) {
                $unsubscribeHandler = $callback;
            }
        });

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->once())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->subscribe('foo');
        $this->assertTrue(is_callable($subscribeHandler));
        $subscribeHandler('foo', 1);
        $deferredSubscribe->resolve(['subscribe', 'foo', 1]);
        $promise->then($this->expectCallableOnceWith(['subscribe', 'foo', 1]));

        $promise = $this->redis->unsubscribe('foo');
        $this->assertTrue(is_callable($unsubscribeHandler));
        $unsubscribeHandler('foo', 0);
        $deferredUnsubscribe->resolve(['unsubscribe', 'foo', 0]);
        $promise->then($this->expectCallableOnceWith(['unsubscribe', 'foo', 0]));
    }

    public function testBlpopWillRejectWhenUnderlyingClientClosesWhileWaitingForResponse(): void
    {
        $closeHandler = null;
        $deferred = new Deferred();
        $client = $this->createMock(StreamingClient::class);
        $client->expects($this->once())->method('callAsync')->with('blpop')->willReturn($deferred->promise());
        $client->expects($this->any())->method('on')->withConsecutive(
            ['close', $this->callback(function ($arg) use (&$closeHandler) {
                $closeHandler = $arg;
                return true;
            })]
        );

        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));

        $loop = $this->createMock(LoopInterface::class);
        $loop->expects($this->never())->method('addTimer');
        assert($loop instanceof LoopInterface);
        Loop::set($loop);

        $promise = $this->redis->blpop('list');

        $this->assertTrue(is_callable($closeHandler));
        $closeHandler();

        $deferred->reject($e = new \RuntimeException());

        $promise->then(null, $this->expectCallableOnceWith($e));
    }

    public function testCloneClosedClientReturnsClientThatWillCreateNewConnectionForFirstCommand(): void
    {
        $this->redis->close();

        $redis = clone $this->redis;

        $deferred = new Deferred($this->expectCallableNever());
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $promise = $redis->callAsync('PING');

        $promise->then($this->expectCallableNever(), $this->expectCallableNever());
    }

    public function testCloneClientReturnsClientThatWillNotBeAffectedByOldClientClosing(): void
    {
        $this->redis->on('close', $this->expectCallableOnce());

        $redis = clone $this->redis;

        $this->assertEquals([], $redis->listeners());

        $deferred = new Deferred($this->expectCallableNever());
        $this->factory->expects($this->once())->method('createClient')->willReturn($deferred->promise());

        $promise = $redis->callAsync('PING');

        $this->redis->close();

        $promise->then($this->expectCallableNever(), $this->expectCallableNever());
    }
}