From 038629753a88c4a0210def8df0878636fca7d6a1 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Fri, 16 Oct 2026 08:35:20 +0300 Subject: [PATCH] Accept INFO without optional fields from old servers --- src/Internal/Connection/SocketConnection.php | 3 +- src/Internal/Protocol/ServerInfo.php | 48 +++++++++-------- tests/ClientTest.php | 56 +++++++++++++++++++- tests/Internal/Protocol/ServerInfoTest.php | 29 ++++++++++ 4 files changed, 112 insertions(+), 24 deletions(-) diff --git a/src/Internal/Connection/SocketConnection.php b/src/Internal/Connection/SocketConnection.php index 854a7ce..0df357a 100644 --- a/src/Internal/Connection/SocketConnection.php +++ b/src/Internal/Connection/SocketConnection.php @@ -170,7 +170,8 @@ private function handshake(Cancellation $cancellation): void pass: $this->config->password, sig: $this->keypair !== null && $frame->nonce !== null && $frame->nonce !== '' ? $this->keypair->sign($frame->nonce) : null, jwt: $this->config->jwt, - noResponders: $this->config->noResponders, + // The server rejects no_responders without headers: the no responders status is a header. + noResponders: $this->config->noResponders && $this->info->allowHeaders, headers: $this->info->allowHeaders, nkey: $this->keypair?->publicKey, ), $cancellation); diff --git a/src/Internal/Protocol/ServerInfo.php b/src/Internal/Protocol/ServerInfo.php index d0a8a14..a07a28a 100644 --- a/src/Internal/Protocol/ServerInfo.php +++ b/src/Internal/Protocol/ServerInfo.php @@ -35,15 +35,15 @@ public static function fromJson(string $json): self public static function fromArray(array $payload): self { return new self( - serverId: self::assert($payload, 'server_id', self::nonEmptyString(...)), - serverName: self::assert($payload, 'server_name', self::nonEmptyString(...)), version: self::assert($payload, 'version', self::nonEmptyString(...)), - go: self::assert($payload, 'go', self::nonEmptyString(...)), - host: self::assert($payload, 'host', self::nonEmptyString(...)), - port: self::assert($payload, 'port', self::positiveInt(...)), - headers: self::assert($payload, 'headers', self::bool(...)), - maxPayload: self::assert($payload, 'max_payload', self::positiveInt(...)), - proto: self::assert($payload, 'proto', self::int(...)), + serverId: self::nullable($payload, 'server_id', self::nonEmptyString(...)), + serverName: self::nullable($payload, 'server_name', self::nonEmptyString(...)), + go: self::nullable($payload, 'go', self::nonEmptyString(...)), + host: self::nullable($payload, 'host', self::nonEmptyString(...)), + port: self::nullable($payload, 'port', self::positiveInt(...)), + headers: self::nullable($payload, 'headers', self::bool(...)) ?? false, + maxPayload: self::nullable($payload, 'max_payload', self::positiveInt(...)), + proto: self::nullable($payload, 'proto', self::int(...)) ?? 0, clientId: self::nullable($payload, 'client_id', self::int(...)), authRequired: self::nullable($payload, 'auth_required', self::bool(...)), tlsRequired: self::nullable($payload, 'tls_required', self::bool(...)), @@ -63,14 +63,18 @@ public static function fromArray(array $payload): self } /** - * @param non-empty-string $serverId the unique identifier of the NATS server - * @param non-empty-string $serverName the name of the NATS server + * Only the version is required: old servers (before 2.2) and some proxies omit the rest, so, like nats.go, + * a missing field means "unknown", a missing headers flag means no header support and a missing proto means + * the original protocol. + * * @param non-empty-string $version the version of NATS - * @param non-empty-string $go the version of golang the NATS server was built with - * @param non-empty-string $host the IP address used to start the NATS server, by default this will be 0.0.0.0 and can be configured with -client_advertise host:port - * @param positive-int $port the port number the NATS server is configured to listen on + * @param ?non-empty-string $serverId the unique identifier of the NATS server + * @param ?non-empty-string $serverName the name of the NATS server + * @param ?non-empty-string $go the version of golang the NATS server was built with + * @param ?non-empty-string $host the IP address used to start the NATS server, by default this will be 0.0.0.0 and can be configured with -client_advertise host:port + * @param ?positive-int $port the port number the NATS server is configured to listen on * @param bool $headers whether the server supports headers - * @param positive-int $maxPayload maximum payload size, in bytes, that the server will accept from the client + * @param ?positive-int $maxPayload maximum payload size, in bytes, that the server will accept from the client * @param int $proto an integer indicating the protocol version of the server. The server version 1.2.0 sets this to 1 to indicate that it supports the "Echo" feature * @param ?int $clientId the internal client identifier in the server. This can be used to filter client connections in monitoring, correlate with error logs * @param ?bool $authRequired if this is true, then the client should try to authenticate upon connect @@ -89,15 +93,15 @@ public static function fromArray(array $payload): self * @param ?string $domain the configured NATS domain of the server */ public function __construct( - public string $serverId, - public string $serverName, public string $version, - public string $go, - public string $host, - public int $port, - public bool $headers, - public int $maxPayload, - public int $proto, + public ?string $serverId = null, + public ?string $serverName = null, + public ?string $go = null, + public ?string $host = null, + public ?int $port = null, + public bool $headers = false, + public ?int $maxPayload = null, + public int $proto = 0, public ?int $clientId = null, public ?bool $authRequired = null, public ?bool $tlsRequired = null, diff --git a/tests/ClientTest.php b/tests/ClientTest.php index 4bfb3da..b67be12 100644 --- a/tests/ClientTest.php +++ b/tests/ClientTest.php @@ -12,6 +12,7 @@ use PHPUnit\Framework\Attributes\DoesNotPerformAssertions; use Revolt\EventLoop; use Thesis\Nats\Exception\ConnectionWasClosed; +use Thesis\Nats\Exception\FeatureIsNotSupported; use Thesis\Nats\Exception\RequestHasNoResponders; use Thesis\Nats\Exception\ServerIsNotAvailable; use Thesis\Nats\Internal\Id; @@ -389,6 +390,46 @@ public function testConnectFailsWhenServerDoesNotAnswerHandshake(): void } } + public function testConnectsToServerWithMinimalInfo(): void + { + // Old servers (before 2.2) and some proxies send only a part of the INFO fields. + $address = self::reserveFreePort(); + $stopProxy = self::proxyToNats( + $address, + $this->natsAddress(), + static function (string $info): string { + /** @var array{version: string} $payload */ + $payload = json_decode(substr($info, \strlen('INFO ')), associative: true, flags: JSON_THROW_ON_ERROR); + + return 'INFO ' . json_encode(['version' => $payload['version']], flags: JSON_THROW_ON_ERROR); + }, + ); + + try { + $client = $this->clientWithUrls([$address]); + $id = Id\generateUniqueId(); + + /** @var DeferredFuture $deferred */ + $deferred = new DeferredFuture(); + + $client->subscribe("{$id}.*", $deferred->complete(...)); + $client->publish("{$id}.happens", new Message('ok')); + + self::assertSame('ok', $deferred->getFuture()->await(new TimeoutCancellation(5))->message->payload); + + // Without the headers flag the server is treated as one without header support. + try { + $client->publish("{$id}.happens", new Message('ok', new Headers(['X-Id' => [$id]]))); + self::fail('The server does not announce header support, publish with headers must fail.'); + } catch (FeatureIsNotSupported) { + } + + $client->stop(); + } finally { + $stopProxy(); + } + } + #[DoesNotPerformAssertions] public function testClientWithFailedConnectionIsDestroyedQuietly(): void { @@ -502,9 +543,10 @@ private static function reserveFreePort(): string * * @param non-empty-string $listen * @param non-empty-string $upstream + * @param ?\Closure(string): string $rewriteInfo replaces the INFO line (without the trailing CRLF) of the server * @return \Closure(): void stops the proxy */ - private static function proxyToNats(string $listen, string $upstream): \Closure + private static function proxyToNats(string $listen, string $upstream, ?\Closure $rewriteInfo = null): \Closure { $server = listen($listen); @@ -522,10 +564,22 @@ private static function proxyToNats(string $listen, string $upstream): \Closure $server, $upstream, $forward, + $rewriteInfo, ): void { while (($client = $server->accept()) !== null) { $nats = connect($upstream); + if ($rewriteInfo !== null) { + $buffer = ''; + + while (!str_contains($buffer, "\r\n") && ($chunk = $nats->read()) !== null) { + $buffer .= $chunk; + } + + $end = (int) strpos($buffer, "\r\n"); + $client->write($rewriteInfo(substr($buffer, 0, $end)) . substr($buffer, $end)); + } + $future = async($forward, $client, $nats); $future->ignore(); $future = async($forward, $nats, $client); diff --git a/tests/Internal/Protocol/ServerInfoTest.php b/tests/Internal/Protocol/ServerInfoTest.php index f761e87..c890bfa 100644 --- a/tests/Internal/Protocol/ServerInfoTest.php +++ b/tests/Internal/Protocol/ServerInfoTest.php @@ -63,6 +63,35 @@ public function testFromJson(string $json, ServerInfo $info): void self::assertEquals($info, ServerInfo::fromJson($json)); } + public function testOnlyVersionIsRequired(): void + { + self::assertEquals(new ServerInfo(version: '2.1.9'), ServerInfo::fromJson('{"version":"2.1.9"}')); + } + + public function testDefaultsOfMissingFields(): void + { + $info = ServerInfo::fromJson('{"server_id":"NCNITNSOJEGWKBAVPXS5M43GLPANADGR6YOMDAYSTVZDHDR6DLD64ZDA","version":"2.1.9"}'); + + self::assertFalse($info->headers); + self::assertSame(0, $info->proto); + self::assertNull($info->maxPayload); + self::assertNull($info->serverName); + } + + public function testMissingVersion(): void + { + self::expectException(\InvalidArgumentException::class); + self::expectExceptionMessage("'info' must contain the key 'version'."); + ServerInfo::fromJson('{"server_id":"NCNITNSOJEGWKBAVPXS5M43GLPANADGR6YOMDAYSTVZDHDR6DLD64ZDA"}'); + } + + public function testPresentFieldIsStillChecked(): void + { + self::expectException(\UnexpectedValueException::class); + self::expectExceptionMessage("The 'info' key 'headers' must be a boolean."); + ServerInfo::fromJson('{"version":"2.1.9","headers":"yes"}'); + } + public function testInvalidJson(): void { self::expectException(\JsonException::class);