Merge pull request #33 from reactphp-parallel/add-support-for-locking-proxies-and-never-discard-them

Add support for locking proxies and never discard them

Author: WyriHaximus

infection.json.dist

@@ -35,7 +35,7 @@
                 "ReactParallel\\ObjectProxy\\Proxy::__destruct::109",
                 "ReactParallel\\ObjectProxy\\NonExistentInterface::interface::22",
                 "ReactParallel\\ObjectProxy\\Message\\Parcel::notify::27",
-                "ReactParallel\\ObjectProxy\\Proxy\\Instance::interface::36"
+                "ReactParallel\\ObjectProxy\\Proxy\\Instance::interface::39"
             ]
         }
     }

src/Proxy.php

@@ -89,7 +89,7 @@ public function has(string $interface): bool
         return array_key_exists($interface, self::KNOWN_INTERFACE);
     }
 
-    public function create(object $object, string $interface): object
+    public function create(object $object, string $interface, bool $locked = false): object
     {
         if ($this->has($interface) === self::HASNT_PROXYABLE_INTERFACE) {
             throw NonExistentInterface::create($interface);
@@ -99,7 +99,7 @@ public function create(object $object, string $interface): object
          * @psalm-suppress EmptyArrayAccess
          */
         $class    = self::KNOWN_INTERFACE[$interface];
-        $instance = new Instance($object, $interface);
+        $instance = new Instance($object, $interface, $locked);
         $hash     = $instance->class() . '___' . spl_object_hash($object);
 
         $this->instances[$hash] = $instance;
@@ -212,10 +212,22 @@ private function handleDestruct(Destruct $destruct): void
 
         $instance = $this->instances[$destruct->hash()];
         $count    = $instance->dereference($destruct->objectHash());
-        if ($count === 0) {
-            unset($this->instances[$destruct->hash()]);
+
+        $this->countDestruct($instance);
+
+        if ($instance->isLocked()) {
+            return;
         }
 
+        if ($count !== 0) {
+            return;
+        }
+
+        unset($this->instances[$destruct->hash()]);
+    }
+
+    private function countDestruct(Instance $instance): void
+    {
         if (! ($this->counterDestruct instanceof Counters)) {
             return;
         }

src/Proxy/Instance.php

@@ -16,11 +16,14 @@ final class Instance
     /** @var array<string> */
     private array $references = [];
 
-    public function __construct(object $object, string $interface)
+    private bool $locked;
+
+    public function __construct(object $object, string $interface, bool $locked)
     {
         $this->object    = $object;
         $this->class     = get_class($object);
         $this->interface = $interface;
+        $this->locked    = $locked;
     }
 
     public function object(): object
@@ -49,4 +52,9 @@ public function dereference(string $hash): int
 
         return count($this->references);
     }
+
+    public function isLocked(): bool
+    {
+        return $this->locked;
+    }
 }

tests/ProxyTest.php

@@ -207,4 +207,47 @@ public function destructionEdgeCases(): void
         );
         self::assertSame(1337, $leet);
     }
+
+    /**
+     * @test
+     */
+    public function destructionLocked(): void
+    {
+        $loop          = EventLoopFactory::create();
+        $factory       = new Factory($loop);
+        $registry      = new InMemmoryRegistry();
+        $proxy         = (new Proxy($factory))->withMetrics($registry);
+        $limitedPool   = $factory->limitedPool(13);
+        $registryProxy = $proxy->create($registry, Registry::class, true);
+        $fn            = static function (int $int, WyriHaximus__Metrics_RegistryProxy $registryProxy, int $sleep): int {
+            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, 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(3, static function () use ($resolve, $v): void {
+                        $resolve($v);
+                    });
+                });
+            })->then(static function () use ($fn, $registryProxy, $limitedPool): PromiseInterface {
+                return $limitedPool->run($fn, [1337, $registryProxy, 20]);
+            })->always(static function () use ($limitedPool, $loop): void {
+                $limitedPool->kill();
+                $loop->stop();
+            }),
+            $loop,
+            133
+        );
+        self::assertSame(1337, $leet);
+    }
 }