Merge pull request #149 from clue-labs/async

Update test suite to use new reactphp/async package instead of clue/reactphp-block

Author: SimonFrings

composer.json

@@ -17,12 +17,12 @@
         "react/event-loop": "^1.2",
         "react/promise": "^3 || ^2.0 || ^1.1",
         "react/promise-timer": "^1.10",
-        "react/socket": "^1.12"
+        "react/socket": "^1.15"
     },
     "require-dev": {
-        "clue/block-react": "^1.5",
         "phpstan/phpstan": "1.10.15 || 1.4.10",
-        "phpunit/phpunit": "^9.6 || ^7.5"
+        "phpunit/phpunit": "^9.6 || ^7.5",
+        "react/async": "^4.2 || ^3.2"
     },
     "autoload": {
         "psr-4": { 

tests/FunctionalTest.php

@@ -3,15 +3,14 @@
 namespace Clue\Tests\React\Redis;
 
 use Clue\React\Redis\RedisClient;
-use React\EventLoop\StreamSelectLoop;
+use React\EventLoop\Loop;
 use React\Promise\Deferred;
 use React\Promise\PromiseInterface;
-use function Clue\React\Block\await;
+use function React\Async\await;
+use function React\Promise\Timer\timeout;
 
 class FunctionalTest extends TestCase
 {
-    /** @var StreamSelectLoop */
-    private $loop;
 
     /** @var string */
     private $uri;
@@ -22,30 +21,28 @@ public function setUp(): void
         if ($this->uri === '') {
             $this->markTestSkipped('No REDIS_URI environment variable given');
         }
-
-        $this->loop = new StreamSelectLoop();
     }
 
     public function testPing(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
+        /** @var PromiseInterface<string> */
         $promise = $redis->ping();
-        $this->assertInstanceOf(PromiseInterface::class, $promise);
 
-        $ret = await($promise, $this->loop);
+        $ret = await($promise);
 
         $this->assertEquals('PONG', $ret);
     }
 
     public function testPingLazy(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
+        /** @var PromiseInterface<string> */
         $promise = $redis->ping();
-        $this->assertInstanceOf(PromiseInterface::class, $promise);
 
-        $ret = await($promise, $this->loop);
+        $ret = await($promise);
 
         $this->assertEquals('PONG', $ret);
     }
@@ -55,89 +52,99 @@ public function testPingLazy(): void
      */
     public function testPingLazyWillNotBlockLoop(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
         $redis->ping();
 
-        $this->loop->run();
+        Loop::run();
     }
 
     /**
      * @doesNotPerformAssertions
      */
     public function testLazyClientWithoutCommandsWillNotBlockLoop(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
-        $this->loop->run();
+        Loop::run();
 
         unset($redis);
     }
 
     public function testMgetIsNotInterpretedAsSubMessage(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
         $redis->mset('message', 'message', 'channel', 'channel', 'payload', 'payload');
 
+        /** @var PromiseInterface<never> */
         $promise = $redis->mget('message', 'channel', 'payload')->then($this->expectCallableOnce());
         $redis->on('message', $this->expectCallableNever());
 
-        await($promise, $this->loop);
+        await($promise);
     }
 
     public function testPipeline(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
         $redis->set('a', 1)->then($this->expectCallableOnceWith('OK'));
         $redis->incr('a')->then($this->expectCallableOnceWith(2));
         $redis->incr('a')->then($this->expectCallableOnceWith(3));
+
+        /** @var PromiseInterface<void> */
         $promise = $redis->get('a')->then($this->expectCallableOnceWith('3'));
 
-        await($promise, $this->loop);
+        await($promise);
     }
 
     public function testInvalidCommand(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
+
+        /** @var PromiseInterface<never> */
         $promise = $redis->doesnotexist(1, 2, 3);
 
         $this->expectException(\Exception::class);
-        await($promise, $this->loop);
+        await($promise);
     }
 
     public function testMultiExecEmpty(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
         $redis->multi()->then($this->expectCallableOnceWith('OK'));
+
+        /** @var PromiseInterface<void> */
         $promise = $redis->exec()->then($this->expectCallableOnceWith([]));
 
-        await($promise, $this->loop);
+        await($promise);
     }
 
     public function testMultiExecQueuedExecHasValues(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
         $redis->multi()->then($this->expectCallableOnceWith('OK'));
         $redis->set('b', 10)->then($this->expectCallableOnceWith('QUEUED'));
         $redis->expire('b', 20)->then($this->expectCallableOnceWith('QUEUED'));
         $redis->incrBy('b', 2)->then($this->expectCallableOnceWith('QUEUED'));
         $redis->ttl('b')->then($this->expectCallableOnceWith('QUEUED'));
+
+        /** @var PromiseInterface<void> */
         $promise = $redis->exec()->then($this->expectCallableOnceWith(['OK', 1, 12, 20]));
 
-        await($promise, $this->loop);
+        await($promise);
     }
 
     public function testPubSub(): void
     {
-        $consumer = new RedisClient($this->uri, null, $this->loop);
-        $producer = new RedisClient($this->uri, null, $this->loop);
+        $consumer = new RedisClient($this->uri);
+        $producer = new RedisClient($this->uri);
 
         $channel = 'channel:test:' . mt_rand();
 
         // consumer receives a single message
+        /** @var Deferred<void> */
         $deferred = new Deferred();
         $consumer->on('message', $this->expectCallableOnce());
         $consumer->on('message', [$deferred, 'resolve']);
@@ -148,12 +155,17 @@ public function testPubSub(): void
         })->then($this->expectCallableOnce());
 
         // expect "message" event to take no longer than 0.1s
-        await($deferred->promise(), $this->loop, 0.1);
+
+        await(timeout($deferred->promise(), 0.1));
+
+        /** @var PromiseInterface<array{0:"unsubscribe",1:string,2:0}> */
+        $promise = $consumer->unsubscribe($channel);
+        await($promise);
     }
 
     public function testClose(): void
     {
-        $redis = new RedisClient($this->uri, null, $this->loop);
+        $redis = new RedisClient($this->uri);
 
         $redis->get('willBeCanceledAnyway')->then(null, $this->expectCallableOnce());