Add multi message parcel

Author: WyriHaximus

src/Message/Parcel.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\ObjectProxy\Message;
+
+use function array_merge;
+
+final class Parcel
+{
+    /** @var array<Call|Existence|Notify> */
+    private array $messages = [];
+
+    /** @var array<Destruct> */
+    private array $destructions = [];
+
+    public function call(Call $call): void
+    {
+        $this->messages[] = $call;
+    }
+
+    public function existence(Existence $existence): void
+    {
+        $this->messages[] = $existence;
+    }
+
+    public function notify(Notify $notify): void
+    {
+        $this->messages[] = $notify;
+    }
+
+    public function destruct(Destruct $destruct): void
+    {
+        $this->destructions[] = $destruct;
+    }
+
+    /**
+     * @return iterable<Call|Destruct|Existence|Notify>
+     */
+    public function messages(): iterable
+    {
+        yield from array_merge($this->messages, $this->destructions);
+    }
+}

src/Proxy.php

@@ -11,17 +11,21 @@
 use ReactParallel\ObjectProxy\Message\Destruct;
 use ReactParallel\ObjectProxy\Message\Existence;
 use ReactParallel\ObjectProxy\Message\Notify;
+use ReactParallel\ObjectProxy\Message\Parcel;
 use ReactParallel\ObjectProxy\Proxy\Instance;
+use Rx\Observable;
 use WyriHaximus\Metrics\Label;
 use WyriHaximus\Metrics\Registry;
 use WyriHaximus\Metrics\Registry\Counters;
 
+use function ApiClients\Tools\Rx\observableFromArray;
 use function array_key_exists;
 use function bin2hex;
 use function get_class;
 use function is_object;
 use function random_bytes;
 use function var_export;
+use function WyriHaximus\iteratorOrArrayToArray;
 
 final class Proxy extends ProxyList
 {
@@ -113,7 +117,7 @@ public function __destruct()
 
     private function setUpHandlers(): void
     {
-        $this->factory->streams()->channel($this->in)->subscribe(function (object $message): void {
+        $this->factory->streams()->channel($this->in)->flatMap(static fn (object $message): Observable => observableFromArray($message instanceof Parcel ? iteratorOrArrayToArray($message->messages()) : [$message]))->subscribe(function (object $message): void {
             if ($message instanceof Existence) {
                 $this->handleExistence($message);
 

tests/Message/ParcelTest.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\Tests\ObjectProxy\Message;
+
+use parallel\Channel;
+use ReactParallel\ObjectProxy\Message\Call;
+use ReactParallel\ObjectProxy\Message\Destruct;
+use ReactParallel\ObjectProxy\Message\Existence;
+use ReactParallel\ObjectProxy\Message\Parcel;
+use WyriHaximus\AsyncTestUtilities\AsyncTestCase;
+
+use function bin2hex;
+use function random_bytes;
+use function time;
+use function WyriHaximus\iteratorOrArrayToArray;
+
+final class ParcelTest extends AsyncTestCase
+{
+    /**
+     * @test
+     */
+    public function getters(): void
+    {
+        $destruct  = new Destruct(bin2hex(random_bytes(1024)), bin2hex(random_bytes(1024)));
+        $call      = new Call(new Channel(1), bin2hex(random_bytes(1024)), bin2hex(random_bytes(1024)), 'hammer', [time()]);
+        $existence = new Existence(bin2hex(random_bytes(1024)), bin2hex(random_bytes(1024)));
+
+        $parcel = new Parcel();
+        $parcel->destruct($destruct);
+        $parcel->call($call);
+        $parcel->existence($existence);
+
+        self::assertSame([$call, $existence, $destruct], iteratorOrArrayToArray($parcel->messages()));
+    }
+}