Shared channel

When rapidly creating and discarding object proxies the channel for
communicating to the main thread stays behind. By using a shared
channel that problem is solved.

Author: WyriHaximus

src/AbstractGeneratedProxy.php

@@ -10,10 +10,12 @@
 abstract class AbstractGeneratedProxy
 {
     private Channel $out;
+    private string $hash;
 
-    final public function __construct(Channel $out)
+    final public function __construct(Channel $out, string $hash)
     {
-        $this->out = $out;
+        $this->out  = $out;
+        $this->hash = $hash;
     }
 
     /**
@@ -26,6 +28,7 @@ final protected function proxyCallToMainThread(string $method, array $args)
         $input = new Channel(1);
         $this->out->send(new Call(
             $input,
+            $this->hash,
             $method,
             $args,
         ));

src/Message/Call.php

@@ -9,6 +9,7 @@
 final class Call
 {
     private Channel $channel;
+    private string $hash;
     private string $method;
 
     /** @var mixed[] */
@@ -17,9 +18,10 @@ final class Call
     /**
      * @param mixed[] $args
      */
-    public function __construct(Channel $channel, string $method, array $args)
+    public function __construct(Channel $channel, string $hash, string $method, array $args)
     {
         $this->channel = $channel;
+        $this->hash    = $hash;
         $this->method  = $method;
         $this->args    = $args;
     }
@@ -29,6 +31,11 @@ public function channel(): Channel
         return $this->channel;
     }
 
+    public function hash(): string
+    {
+        return $this->hash;
+    }
+
     public function method(): string
     {
         return $this->method;

src/Proxy.php

@@ -7,21 +7,26 @@
 use parallel\Channel;
 use ReactParallel\Factory;
 use ReactParallel\ObjectProxy\Generated\ProxyList;
+use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Proxy\CallHandler;
+use Rx\Observable;
 
 use function array_key_exists;
+use function spl_object_hash;
 
 final class Proxy extends ProxyList
 {
     private const HASNT_PROXYABLE_INTERFACE = false;
 
-    private Factory $factory;
+    private Channel $output;
+    private Observable $outputStream;
     private CallHandler $callHandler;
 
     public function __construct(Factory $factory)
     {
-        $this->factory     = $factory;
-        $this->callHandler = new CallHandler($this);
+        $this->output       = new Channel(Channel::Infinite);
+        $this->outputStream = $factory->streams()->channel($this->output)->share();
+        $this->callHandler  = new CallHandler($this);
     }
 
     public function has(string $interface): bool
@@ -35,15 +40,14 @@ public function create(object $object, string $interface): object
             throw NonExistentInterface::create($interface);
         }
 
-        $output = new Channel(Channel::Infinite);
-
-        $this->factory->streams()->channel($output)->subscribe(
+        $hash = spl_object_hash($object);
+        $this->outputStream->filter(static fn (Call $call): bool => $call->hash() === $hash)->subscribe(
             ($this->callHandler)($object, $interface)
         );
 
         $class = self::KNOWN_INTERFACE[$interface];
 
         /** @psalm-suppress InvalidStringClass */
-        return new $class($output);
+        return new $class($this->output, $hash);
     }
 }

tests/Message/CallTest.php

@@ -18,12 +18,14 @@ final class CallTest extends AsyncTestCase
     public function getters(): void
     {
         $channel = new Channel(1);
+        $hash    = 'haajshdkjlsajkl';
         $method  = 'hammer';
         $args    = [time()];
 
-        $call = new Call($channel, $method, $args);
+        $call = new Call($channel, $hash, $method, $args);
 
         self::assertSame($channel, $call->channel());
+        self::assertSame($hash, $call->hash());
         self::assertSame($method, $call->method());
         self::assertSame($args, $call->args());
     }

tests/Proxy/CallHandlerTest.php

@@ -40,8 +40,9 @@ public function logger(): void
         });
         $callHandler = new Proxy\CallHandler($proxy);
         $time        = time();
+        $hash        = 'asljdjaslkdhklasdslkadskl';
         $channel     = new Channel(1);
-        $call        = new Call($channel, 'get', [LoggerInterface::class]);
+        $call        = new Call($channel, $hash, 'get', [LoggerInterface::class]);
 
         $loop->futureTick(static function () use ($container, $callHandler, $call): void {
             ($callHandler)($container, ContainerInterface::class)($call);