Release an unreferenced client and close its connection

README.md

@@ -97,6 +97,8 @@ trapSignal([\SIGTERM, \SIGINT]);
 $nc->drain();
 ```
 
+The connection lives as long as the client: keep a reference to the client (or to an object that uses it, such as a subscription, a JetStream context or a consumer) while you need it. A client that is no longer referenced closes its connection without draining, so prefer an explicit `drain()` or `stop()`.
+
 #### Queues
 
 ```php

src/Client.php

@@ -63,16 +63,13 @@ final class Client
 
     private readonly Id\SubscriptionIdGenerator $subscriptionIdGenerator;
 
-    private readonly Acks $acks;
-
     public function __construct(
         private readonly Config $config,
         private readonly Serializer $serializer = new ValinorSerializer(),
         private readonly Encoder $encoder = new NativeEncoder(),
     ) {
         $this->connectionFactory = Connection\SocketConnectionFactory::fromConfig($this->config);
         $this->subscriptionIdGenerator = new Id\SubscriptionIdGenerator();
-        $this->acks = Acks::fromClient($this);
     }
 
     /**
@@ -404,7 +401,8 @@ public function toJetStreamDelivery(Delivery $delivery): JetStreamDelivery
         return new JetStreamDelivery(
             message: $delivery->message,
             subject: $delivery->subject,
-            acks: $this->acks,
+            // Created per delivery: kept in a property, the acks would form a reference cycle with the client.
+            acks: Acks::fromClient($this),
             metadata: $delivery->replyTo !== null ? Metadata::parse($delivery->replyTo) : null,
             replyTo: $delivery->replyTo,
         );
@@ -520,8 +518,14 @@ private function subscribeCallback(
             throw $e;
         }
 
-        return function (?Cancellation $cancellation = null) use ($subscriptionId): void {
-            $this->unsubscribe($subscriptionId, $cancellation);
+        // The client keeps the request inbox subscription, so it is referred back to weakly.
+        $client = \WeakReference::create($this);
+
+        return static function (?Cancellation $cancellation = null) use (
+            $client,
+            $subscriptionId,
+        ): void {
+            $client->get()?->unsubscribe($subscriptionId, $cancellation);
         };
     }
 
@@ -813,18 +817,30 @@ private function connection(?Cancellation $cancellation = null): Connection\Conn
             /** @var ?\Throwable $reason the fatal server error that closed the connection */
             $reason = null;
 
-            $connection->hooks->onMessage($this->invokeSubscriber(...));
-            $connection->hooks->onError(function (Hooks\ErrorReceived $event) use (
+            // The connection's reader runs in the event loop: strong references from its hooks would keep the client
+            // alive for as long as the socket is open, so the destructor that closes the socket would never run.
+            // Once the client is gone, its hooks do nothing: e.g. the close event of the connection it closed itself.
+            $client = \WeakReference::create($this);
+
+            $connection->hooks->onMessage(static function (Hooks\MessageReceived $event) use ($client): void {
+                $client->get()?->invokeSubscriber($event);
+            });
+            $connection->hooks->onError(static function (Hooks\ErrorReceived $event) use (
+                $client,
                 $connection,
                 &$reason,
             ): void {
-                $reason ??= $this->handleError($event, $connection);
+                $self = $client->get();
+                if ($self !== null) {
+                    $reason ??= $self->handleError($event, $connection);
+                }
             });
-            $connection->hooks->onClose(function () use (
+            $connection->hooks->onClose(static function () use (
+                $client,
                 $connection,
                 &$reason,
             ): void {
-                $this->closed($connection, $reason);
+                $client->get()?->closed($connection, $reason);
             });
 
             return $connection;

tests/ClientTest.php

@@ -405,6 +405,84 @@ public function testClientWithFailedConnectionIsDestroyedQuietly(): void
         gc_collect_cycles();
     }
 
+    public function testDroppedClientClosesItsConnection(): void
+    {
+        $name = Id\generateUniqueId();
+        $client = new Client($this->namedConfig($name));
+
+        // Connects, sets up the request inbox and the JetStream acks.
+        $client->jetStream();
+        self::assertTrue($this->isConnected($name));
+
+        $reference = \WeakReference::create($client);
+        unset($client);
+
+        // Released at once, without waiting for the cycle collector.
+        self::assertNull($reference->get());
+        $this->awaitDisconnected($name);
+    }
+
+    public function testDroppedClientWithSubscriptionClosesItsConnection(): void
+    {
+        $name = Id\generateUniqueId();
+        $client = new Client($this->namedConfig($name));
+        $client->subscribe('events.*', static function (): void {});
+
+        self::assertTrue($this->isConnected($name));
+
+        unset($client);
+        // The client and its subscriptions refer to each other.
+        gc_collect_cycles();
+
+        $this->awaitDisconnected($name);
+    }
+
+    public function testHeldSubscriptionKeepsItsClientAlive(): void
+    {
+        $id = Id\generateUniqueId();
+
+        /** @var DeferredFuture<Delivery> $deferred */
+        $deferred = new DeferredFuture();
+
+        $subscription = new Client($this->config())->subscribe("{$id}.*", $deferred->complete(...));
+        gc_collect_cycles();
+
+        $this->client()->publish("{$id}.happens", new Message('ok'));
+
+        self::assertSame('ok', $deferred->getFuture()->await(new TimeoutCancellation(5))->message->payload);
+
+        $subscription->stop();
+    }
+
+    /**
+     * @param non-empty-string $name
+     */
+    private function namedConfig(string $name): Config
+    {
+        $config = $this->config();
+
+        return new Config(
+            urls: $config->urls,
+            user: $config->user,
+            password: $config->password,
+            noResponders: $config->noResponders,
+            clientName: $name,
+        );
+    }
+
+    /**
+     * @param non-empty-string $name
+     */
+    private function awaitDisconnected(string $name): void
+    {
+        // The server notices the closed socket a moment later.
+        for ($attempt = 0; $attempt < 20 && $this->isConnected($name); ++$attempt) {
+            delay(0.05);
+        }
+
+        self::assertFalse($this->isConnected($name));
+    }
+
     /**
      * @return non-empty-string an address on which nothing is listening at the moment
      */

tests/NatsTestCase.php

@@ -111,33 +111,55 @@ final protected function clientNamed(string $name): Client
      */
     final protected function kick(string $name): void
     {
-        $sys = $this->sysClient ??= new Client(new Config(
-            urls: $this->config()->urls,
-            user: 'admin',
-            password: 'Pswd1',
-            noResponders: true,
-        ));
+        $connection = $this->connectionNamed($name) ?? self::fail("The client '{$name}' is not connected.");
+
+        self::decodeServerResponse(
+            $this->sysClient()->request(
+                "\$SYS.REQ.SERVER.{$connection['server']}.KICK",
+                new Message(json_encode(['cid' => $connection['cid']], JSON_THROW_ON_ERROR)),
+                new TimeoutCancellation(5),
+            ),
+        );
+    }
+
+    /**
+     * Whether the server still has a connection of the client connected under the given name.
+     *
+     * @param non-empty-string $name
+     */
+    final protected function isConnected(string $name): bool
+    {
+        return $this->connectionNamed($name) !== null;
+    }
 
+    /**
+     * @param non-empty-string $name
+     * @return ?array{server: non-empty-string, cid: int}
+     */
+    private function connectionNamed(string $name): ?array
+    {
         /** @var array{server: array{id: non-empty-string}, data: array{connections: list<array{cid: int, name?: string}>}} $connz */
         $connz = self::decodeServerResponse(
-            $sys->request('$SYS.REQ.SERVER.PING.CONNZ', new Message('{"acc": "CLIENT"}'), new TimeoutCancellation(5)),
+            $this->sysClient()->request('$SYS.REQ.SERVER.PING.CONNZ', new Message('{"acc": "CLIENT"}'), new TimeoutCancellation(5)),
         );
 
         foreach ($connz['data']['connections'] as $connection) {
             if (($connection['name'] ?? null) === $name) {
-                self::decodeServerResponse(
-                    $sys->request(
-                        "\$SYS.REQ.SERVER.{$connz['server']['id']}.KICK",
-                        new Message(json_encode(['cid' => $connection['cid']], JSON_THROW_ON_ERROR)),
-                        new TimeoutCancellation(5),
-                    ),
-                );
-
-                return;
+                return ['server' => $connz['server']['id'], 'cid' => $connection['cid']];
             }
         }
 
-        self::fail("The client '{$name}' is not connected.");
+        return null;
+    }
+
+    private function sysClient(): Client
+    {
+        return $this->sysClient ??= new Client(new Config(
+            urls: $this->config()->urls,
+            user: 'admin',
+            password: 'Pswd1',
+            noResponders: true,
+        ));
     }
 
     /**