Merge pull request #2 from reactphp-parallel/refactor0recv-observable-to-use-one-event-loop-for-all-channels

Refactor RecvObservable and SingleRecv into Factory which is syntactic sugar around the event loop bridge

Author: WyriHaximus

composer.json

@@ -12,6 +12,7 @@
     "require": {
         "php": "^7.4",
         "ext-parallel": "*",
+        "react-parallel/event-loop": "dev-master",
         "react/event-loop": "^1.1",
         "react/promise": "^2.7",
         "reactivex/rxphp": "^2.0",

composer.lock

@@ -1,7 +1,6 @@
 parameters:
 	checkMissingIterableValueType: false
 	ignoreErrors:
-	    - '#In method \"ReactParallel\\Streams\\RecvObservable::recv\", caught \"Throwable\" must be rethrown. Either catch a more specific exception or add a \"throw\" clause in the "catch" block to propagate the exception. More info: http:\/\/bit.ly\/failloud#'
 	    - '#Function sleep is unsafe to use.#'
 
 includes:

phpstan.neon

@@ -0,0 +1,28 @@
+<?php declare(strict_types=1);
+
+namespace ReactParallel\Streams;
+
+use parallel\Channel;
+use React\Promise\PromiseInterface;
+use ReactParallel\EventLoop\EventLoopBridge;
+use Rx\Observable;
+
+final class Factory
+{
+    private EventLoopBridge $loop;
+
+    public function __construct(EventLoopBridge $loop)
+    {
+        $this->loop = $loop;
+    }
+
+    public function stream(Channel $channel): Observable
+    {
+        return $this->loop->observe($channel);
+    }
+
+    public function single(Channel $channel): PromiseInterface
+    {
+        return $this->loop->observe($channel)->take(1)->toPromise();
+    }
+}

src/Factory.php

@@ -5,9 +5,12 @@
 use parallel\Channel;
 use parallel\Events;
 use React\EventLoop\Factory;
+use ReactParallel\EventLoop\EventLoopBridge;
+use ReactParallel\Streams\Factory as StreamFactory;
 use ReactParallel\Streams\RecvObservable;
 use WyriHaximus\AsyncTestUtilities\AsyncTestCase;
 use function parallel\run;
+use function React\Promise\all;
 use function sleep;
 
 /**
@@ -23,24 +26,38 @@ public function recv(): void
         $d = bin2hex(random_bytes(13));
 
         $loop = Factory::create();
-        $channel = Channel::make($d, Channel::Infinite);
-        $events = new Events();
-        $events->setTimeout(0);
-        $events->addChannel($channel);
+        $channels = [Channel::make($d . '_a', Channel::Infinite), Channel::make($d . '_b', Channel::Infinite)];
 
-        $recvObservable = new RecvObservable($loop, $events);
+        $recvObservable = new StreamFactory(new EventLoopBridge($loop));
 
-        run(function () use ($channel): void {
+        run(function () use ($channels): void {
             foreach (range(0, 13) as $i) {
                 usleep(100);
-                $channel->send($i);
+                foreach (range(0, 130) as $j) {
+                    foreach ($channels as $channel) {
+                        $channel->send($i);
+                    }
+                }
             }
             sleep(1);
-            $channel->close();
+            foreach ($channels as $channel) {
+                $channel->close();
+            }
         });
 
-        $rd = $this->await($recvObservable->recv()->toArray()->toPromise(), $loop, 3.3);
+        $promises = [];
+        foreach ($channels as $channel) {
+            $promises[] = $recvObservable->stream($channel)->toArray()->toPromise();
+        }
+
+        $rd = $this->await(all($promises), $loop, 3.3);
 
-        self::assertSame(range(0, 13), $rd);
+        $range = [];
+        foreach (range(0, 13) as $i) {
+            foreach (range(0, 130) as $j) {
+                $range[] = $i;
+            }
+        }
+        self::assertSame([$range, $range], $rd);
     }
 }

src/RecvObservable.php

@@ -6,6 +6,8 @@
 use parallel\Events;
 use React\EventLoop\Factory;
 use React\Promise\ExtendedPromiseInterface;
+use ReactParallel\EventLoop\EventLoopBridge;
+use ReactParallel\Streams\Factory as StreamFactory;
 use ReactParallel\Streams\SingleRecv;
 use WyriHaximus\AsyncTestUtilities\AsyncTestCase;
 use ReactParallel\FutureToPromiseConverter\FutureToPromiseConverter;
@@ -28,17 +30,14 @@ public function recv(): void
 
         $loop = Factory::create();
         $channel = Channel::make($d, Channel::Infinite);
-        $events = new Events();
-        $events->setTimeout(0);
-        $events->addChannel($channel);
 
-        $singleRecv = new SingleRecv($loop, $events);
+        $singleRecv = new StreamFactory(new EventLoopBridge($loop));
 
         $loop->addTimer(2, function () use ($channel, $d): void {
             $channel->send($d);
         });
 
-        $rd = $this->await($singleRecv->recv(), $loop, 3.3);
+        $rd = $this->await($singleRecv->single($channel), $loop, 3.3);
 
         self::assertSame($d, $rd);
     }
@@ -52,17 +51,14 @@ public function timedOut(): void
 
         $loop = Factory::create();
         $channel = Channel::make($d, Channel::Infinite);
-        $events = new Events();
-        $events->setTimeout(0);
-        $events->addChannel($channel);
 
-        $singleRecv = new SingleRecv($loop, $events);
+        $singleRecv = new StreamFactory(new EventLoopBridge($loop));
 
         $loop->futureTick(static function () use ($channel): void {
             $channel->close();
         });
 
-        $rd = $this->await($singleRecv->recv(), $loop, 3.3);
+        $rd = $this->await($singleRecv->single($channel), $loop, 3.3);
 
         self::assertNull($rd);
     }