Merge pull request #59 from reactphp-parallel/2.x-replace-observables-with-iterables

[2.x] Replace observables with iterables

Author: WyriHaximus

composer.json

@@ -15,11 +15,8 @@
         "react/async": "^4.2",
         "react/event-loop": "^1.5",
         "react/promise": "^2.11 || ^3.0",
-        "reactivex/rxphp": "^2.0.12",
         "wyrihaximus/constants": "^1.6.0",
-        "wyrihaximus/metrics": "^1.0 || ^2",
-        "wyrihaximus/react-awaitable-observable": "^1",
-        "wyrihaximus/react-event-loop-rx-scheduler-hook-up": "^0.1.1"
+        "wyrihaximus/metrics": "^1.0 || ^2"
     },
     "require-dev": {
         "wyrihaximus/async-test-utilities": "^7.2 || ^5"

composer.lock

@@ -1,5 +1,5 @@
 {
-    "timeout": 120,
+    "timeout": 1800,
     "source": {
         "directories": [
             "src"
@@ -9,8 +9,12 @@
         "text": "./var/infection.log",
         "summary": "./var/infection-summary.log",
         "json": "./var/infection.json",
-        "perMutator": "./var/infection-per-mutator.md"
+        "perMutator": "./var/infection-per-mutator.md",
+        "github": true
     },
+    "minMsi": 100,
+    "minCoveredMsi": 100,
+    "ignoreMsiWithNoMutations": true,
     "mutators": {
         "@default": true
     },

infection.json.dist

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\EventLoop;
+
+/**
+ * @internal
+ *
+ * Marker class for when a stream is done
+ */
+final class Done
+{
+}

src/Done.php

@@ -10,14 +10,12 @@
 use React\EventLoop\Loop;
 use React\EventLoop\TimerInterface;
 use React\Promise\Deferred;
-use Rx\Subject\Subject;
 use WyriHaximus\Metrics\Label;
 
 use function count;
 use function React\Async\await;
 use function spl_object_hash;
 use function spl_object_id;
-use function WyriHaximus\React\awaitObservable;
 
 use const WyriHaximus\Constants\Numeric\ZERO;
 
@@ -40,7 +38,7 @@ final class EventLoopBridge
 
     private TimerInterface|null $timer = null;
 
-    /** @var array<int, Subject> */
+    /** @var array<int, Stream> */
     private array $channels = [];
 
     /** @var array<int, Deferred> */
@@ -68,8 +66,7 @@ public function withMetrics(Metrics $metrics): self
     /** @return iterable<mixed> */
     public function observe(Channel $channel): iterable
     {
-        $subject                                 = new Subject();
-        $this->channels[spl_object_id($channel)] = $subject;
+        $this->channels[spl_object_id($channel)] = new Stream();
         $this->events->addChannel($channel);
 
         if ($this->metrics instanceof Metrics) {
@@ -78,7 +75,7 @@ public function observe(Channel $channel): iterable
 
         $this->startTimer();
 
-        return awaitObservable($subject);
+        yield from $this->channels[spl_object_id($channel)]->iterable();
     }
 
     public function await(Future $future): mixed
@@ -228,7 +225,7 @@ private function handleFutureReadEvent(Events\Event $event): void
 
     private function handleChannelReadEvent(Events\Event $event): void
     {
-        $this->channels[spl_object_id($event->object)]->onNext($event->value);
+        $this->channels[spl_object_id($event->object)]->value($event->value);
         $this->events->addChannel($event->object); /** @phpstan-ignore-line */
 
         if (! ($this->metrics instanceof Metrics)) {
@@ -240,7 +237,7 @@ private function handleChannelReadEvent(Events\Event $event): void
 
     private function handleCloseEvent(Events\Event $event): void
     {
-        $this->channels[spl_object_id($event->object)]->onCompleted();
+        $this->channels[spl_object_id($event->object)]->done();
         unset($this->channels[spl_object_id($event->object)]);
 
         if (! ($this->metrics instanceof Metrics)) {

src/EventLoopBridge.php

@@ -0,0 +1,54 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\EventLoop;
+
+use React\Promise\Deferred;
+use SplQueue;
+
+use function React\Async\await;
+
+final class Stream
+{
+    private SplQueue $queue;
+    private Deferred $wait;
+
+    public function __construct()
+    {
+        $this->queue = new SplQueue();
+        $this->queue->setIteratorMode(SplQueue::IT_MODE_DELETE);
+        $this->wait = new Deferred();
+    }
+
+    public function value(mixed $value): void
+    {
+        $this->queue->enqueue($value);
+        $this->wait->resolve(new Value());
+    }
+
+    public function done(): void
+    {
+        $this->wait->resolve(new Done());
+    }
+
+    /** @return iterable<mixed> */
+    public function iterable(): iterable
+    {
+        do {
+            $run  = false;
+            $type = await($this->wait->promise());
+
+            foreach ($this->queue as $value) {
+                yield $value;
+            }
+
+            if ($type instanceof Done) {
+                continue;
+            }
+
+            $this->wait = new Deferred();
+            $run        = true;
+        } while ($run);
+    }
+}

src/Stream.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\EventLoop;
+
+/**
+ * @internal
+ *
+ * Marker class for when a stream is received a value
+ */
+final class Value
+{
+}