Accept INFO without optional fields from old servers

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);

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,

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<Delivery> $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);

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);