Merge pull request #1 from reactphp-parallel/add-recv-observable

Add RecvObservable

Author: WyriHaximus

.phpcs.cache

@@ -14,23 +14,24 @@
         "ext-parallel": "*",
         "react/event-loop": "^1.1",
         "react/promise": "^2.7",
+        "reactivex/rxphp": "^2.0",
         "wyrihaximus/constants": "^1.4.3"
     },
     "require-dev": {
         "wyrihaximus/async-test-utilities": "^2.0",
         "wyrihaximus/ticking-promise": "^1.6"
     },
-    "extra": {
-        "unused": [
-            "ext-parallel"
-        ]
-    },
     "config": {
         "platform": {
             "php": "7.4"
         },
         "sort-packages": true
     },
+    "extra": {
+        "unused": [
+            "ext-parallel"
+        ]
+    },
     "autoload": {
         "psr-4": {
             "ReactParallel\\Streams\\": "src/"

composer.json

@@ -1,6 +1,8 @@
 parameters:
 	checkMissingIterableValueType: false
-	ignoreErrors: []
+	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:
 	- vendor/wyrihaximus/async-test-utilities/rules.neon

composer.lock

@@ -0,0 +1,64 @@
+<?php declare(strict_types=1);
+
+namespace ReactParallel\Streams;
+
+use parallel\Events;
+use React\EventLoop\LoopInterface;
+use React\EventLoop\TimerInterface;
+use Rx\Observable;
+use Rx\Subject\Subject;
+use Throwable;
+
+final class RecvObservable
+{
+    private LoopInterface $loop;
+
+    private Events $events;
+
+    public function __construct(LoopInterface $loop, Events $events)
+    {
+        $this->loop   = $loop;
+        $this->events = $events;
+    }
+
+    public function recv(): Observable
+    {
+        $subject = new Subject();
+
+        // Call 1K times per second
+        $timer = $this->loop->addPeriodicTimer(0.05, function () use (&$timer, $subject): void {
+            try {
+                while ($event = $this->events->poll()) {
+                    if ($event->type === Events\Event\Type::Read) {
+                        $subject->onNext($event->value);
+                        $this->events->addChannel($event->object);
+
+                        break;
+                    }
+
+                    if ($event->type !== Events\Event\Type::Close) {
+                        break;
+                    }
+
+                    if ($timer instanceof TimerInterface) {
+                        $this->loop->cancelTimer($timer);
+                    }
+
+                    $subject->onCompleted();
+
+                    return;
+                }
+            } catch (Events\Error\Timeout $timeout) {
+                return;
+            } catch (Throwable $throwable) {
+                if ($timer instanceof TimerInterface) {
+                    $this->loop->cancelTimer($timer);
+                }
+
+                $subject->onError($throwable);
+            }
+        });
+
+        return $subject;
+    }
+}

phpstan.neon

@@ -0,0 +1,46 @@
+<?php declare(strict_types=1);
+
+namespace ReactParallel\Tests\Streams;
+
+use parallel\Channel;
+use parallel\Events;
+use React\EventLoop\Factory;
+use ReactParallel\Streams\RecvObservable;
+use WyriHaximus\AsyncTestUtilities\AsyncTestCase;
+use function parallel\run;
+use function sleep;
+
+/**
+ * @internal
+ */
+final class SingleRecvObservableTest extends AsyncTestCase
+{
+    /**
+     * @test
+     */
+    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);
+
+        $recvObservable = new RecvObservable($loop, $events);
+
+        run(function () use ($channel): void {
+            foreach (range(0, 13) as $i) {
+                usleep(100);
+                $channel->send($i);
+            }
+            sleep(1);
+            $channel->close();
+        });
+
+        $rd = $this->await($recvObservable->recv()->toArray()->toPromise(), $loop, 3.3);
+
+        self::assertSame(range(0, 13), $rd);
+    }
+}

src/RecvObservable.php

@@ -38,7 +38,6 @@ public function recv(): void
             $channel->send($d);
         });
 
-        $loop->run();
         $rd = $this->await($singleRecv->recv(), $loop, 3.3);
 
         self::assertSame($d, $rd);
@@ -63,7 +62,6 @@ public function timedOut(): void
             $channel->close();
         });
 
-        $loop->run();
         $rd = $this->await($singleRecv->recv(), $loop, 3.3);
 
         self::assertNull($rd);