From 45bc573287ce3d2cee38d9baa838f20ef6b684b0 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Thu, 15 Oct 2026 21:26:35 +0300 Subject: [PATCH] Release an unreferenced client and close its connection --- README.md | 2 ++ src/Client.php | 38 ++++++++++++++------ tests/ClientTest.php | 78 ++++++++++++++++++++++++++++++++++++++++++ tests/NatsTestCase.php | 56 +++++++++++++++++++++--------- 4 files changed, 146 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 3b2d034..c07a399 100644 --- a/README.md +++ b/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 diff --git a/src/Client.php b/src/Client.php index c3f61c4..4f20295 100644 --- a/src/Client.php +++ b/src/Client.php @@ -63,8 +63,6 @@ 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(), @@ -72,7 +70,6 @@ public function __construct( ) { $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; diff --git a/tests/ClientTest.php b/tests/ClientTest.php index 0a077e5..4bfb3da 100644 --- a/tests/ClientTest.php +++ b/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 $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 */ diff --git a/tests/NatsTestCase.php b/tests/NatsTestCase.php index b9bc1d4..465cc4e 100644 --- a/tests/NatsTestCase.php +++ b/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}} $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, + )); } /**