Make it possible to notify the main thread of our existence

Author: WyriHaximus

composer-require-checker.json

@@ -7,7 +7,8 @@
     "WyriHaximus\\Constants\\Numeric\\ZERO", "WyriHaximus\\getIn", "WyriHaximus\\Constants\\Numeric\\TWO",
     "Composer\\Composer", "Composer\\Config", "Composer\\IO\\IOInterface", "Composer\\Script\\Event",
     "Composer\\EventDispatcher\\EventSubscriberInterface", "Composer\\Package\\RootPackageInterface",
-    "Composer\\Plugin\\PluginInterface", "Composer\\Script\\ScriptEvents"
+    "Composer\\Plugin\\PluginInterface", "Composer\\Script\\ScriptEvents",
+    "Safe\\chmod", "Safe\\file_get_contents", "Safe\\file_put_contents", "Safe\\sprintf"
   ],
   "php-core-extensions" : [
     "Core",

composer.json

@@ -21,14 +21,14 @@
         "psr/log": "^1.1",
         "react-parallel/react-parallel": "^1.0",
         "rx/operator-extras": "^2.1",
-        "thecodingmachine/safe": "^1.2",
         "wyrihaximus/constants": "^1.5",
         "wyrihaximus/iterator-or-array-to-array": "^1.1",
         "wyrihaximus/metrics": "dev-master",
         "wyrihaximus/string-get-in": "^1.0"
     },
     "require-dev": {
         "monolog/monolog": "^2.1",
+        "thecodingmachine/safe": "^1.3",
         "wyrihaximus/async-test-utilities": "^2.3",
         "yuloh/container": "^1.0"
     },

composer.lock

@@ -7,6 +7,7 @@
 use parallel\Channel;
 use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Message\Destruct;
+use ReactParallel\ObjectProxy\Message\Existence;
 use ReactParallel\ObjectProxy\Message\Notify;
 
 use function spl_object_hash;
@@ -65,4 +66,13 @@ final protected function notifyMainThreadAboutDestruction(): void
             // @ignoreException
         }
     }
+
+    final public function notifyMainThreadAboutOurExistence(): void
+    {
+        try {
+            $this->out->send(new Existence($this->hash, spl_object_hash($this)));
+        } catch (Channel\Error\Closed $closed) {
+            // @ignoreException
+        }
+    }
 }

src/AbstractGeneratedProxy.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\ObjectProxy\Message;
+
+final class Existence
+{
+    private string $hash;
+    private string $objectHash;
+
+    public function __construct(string $hash, string $objectHash)
+    {
+        $this->hash       = $hash;
+        $this->objectHash = $objectHash;
+    }
+
+    public function hash(): string
+    {
+        return $this->hash;
+    }
+
+    public function objectHash(): string
+    {
+        return $this->objectHash;
+    }
+}

src/Message/Existence.php

@@ -9,6 +9,7 @@
 use ReactParallel\ObjectProxy\Generated\ProxyList;
 use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Message\Destruct;
+use ReactParallel\ObjectProxy\Message\Existence;
 use ReactParallel\ObjectProxy\Message\Notify;
 use ReactParallel\ObjectProxy\Proxy\Instance;
 use WyriHaximus\Metrics\Label;
@@ -112,12 +113,22 @@ public function __destruct()
     private function setUpHandlers(): void
     {
         $this->factory->streams()->channel($this->in)->subscribe(function (object $message): void {
+            if ($message instanceof Existence) {
+                $this->handleExistence($message);
+
+                return;
+            }
+
             if ($message instanceof Notify) {
                 $this->handleNotify($message);
+
+                return;
             }
 
             if ($message instanceof Call) {
                 $this->handleCall($message);
+
+                return;
             }
 
             if (! ($message instanceof Destruct)) {
@@ -130,11 +141,21 @@ private function setUpHandlers(): void
         });
     }
 
+    private function handleExistence(Existence $existence): void
+    {
+        if (! array_key_exists($existence->hash(), $this->instances)) {
+            var_export($existence);
+
+            return;
+        }
+
+        $instance = $this->instances[$existence->hash()];
+        $instance->reference($existence->objectHash());
+    }
+
     private function handleNotify(Notify $notify): void
     {
         if (! array_key_exists($notify->hash(), $this->instances)) {
-            var_export($notify);
-
             return;
         }
 
@@ -151,8 +172,6 @@ private function handleNotify(Notify $notify): void
     private function handleCall(Call $call): void
     {
         if (! array_key_exists($call->hash(), $this->instances)) {
-            var_export($call);
-
             return;
         }
 

src/Proxy.php

@@ -12,6 +12,7 @@
 use React\Promise\PromiseInterface;
 use React\Promise\Timer\TimeoutException;
 use ReactParallel\Factory;
+use ReactParallel\ObjectProxy\Generated\WyriHaximus__Metrics_RegistryProxy;
 use ReactParallel\ObjectProxy\NonExistentInterface;
 use ReactParallel\ObjectProxy\Proxy;
 use stdClass;
@@ -26,6 +27,7 @@
 use function explode;
 use function range;
 use function React\Promise\all;
+use function Safe\sleep;
 use function time;
 
 final class ProxyTest extends AsyncTestCase
@@ -179,29 +181,31 @@ public function destructionEdgeCases(): void
         $factory       = new Factory($loop);
         $registry      = new InMemmoryRegistry();
         $proxy         = (new Proxy($factory))->withMetrics($registry);
-        $limitedPool   = $factory->limitedPool(1);
+        $limitedPool   = $factory->limitedPool(13);
         $registryProxy = $proxy->create($registry, Registry::class);
-        $fn            = static function (int $int, Registry $registryProxy): int {
+        $fn            = static function (int $int, WyriHaximus__Metrics_RegistryProxy $registryProxy, int $sleep): int {
+            $registryProxy->notifyMainThreadAboutOurExistence();
+            sleep($sleep);
             $registryProxy->counter('counter', 'bla bla bla', new Label\Name('name'))->counter(new Label('name', 'value'))->incr();
 
             return $int;
         };
 
         $promises = [];
         foreach (range(0, 3) as $i) {
-            $promises[] = $limitedPool->run($fn, [$i, $registryProxy]);
+            $promises[] = $limitedPool->run($fn, [$i, $registryProxy, 0]);
         }
 
         $leet = $this->await(
             // @phpstan-ignore-next-line
             all($promises)->then(static function (array $v) use ($factory): PromiseInterface {
                 return new Promise(static function (callable $resolve) use ($v, $factory): void {
-                    $factory->loop()->addTimer(20, static function () use ($resolve, $v): void {
+                    $factory->loop()->addTimer(3, static function () use ($resolve, $v): void {
                         $resolve($v);
                     });
                 });
             })->then(static function () use ($fn, $registryProxy, $limitedPool): PromiseInterface {
-                return $limitedPool->run($fn, [1337, $registryProxy]);
+                return $limitedPool->run($fn, [1337, $registryProxy, 20]);
             })->always(static function () use ($limitedPool): void {
                 $limitedPool->close();
             }),