From 139ae50a3bfc69292250a60671667ec6cb1f8814 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Fri, 16 Oct 2026 11:04:17 +0300 Subject: [PATCH] Parse the connection URI without parse_url --- README.md | 23 ++++ src/Config.php | 142 ++++----------------- src/Internal/Connection/Query.php | 123 ++++++++++++++++++ src/Internal/Connection/Uri.php | 203 ++++++++++++++++++++++++++++++ tests/ClientTest.php | 19 +++ tests/ConfigTest.php | 127 +++++++++++++++++++ 6 files changed, 521 insertions(+), 116 deletions(-) create mode 100644 src/Internal/Connection/Query.php create mode 100644 src/Internal/Connection/Uri.php diff --git a/README.md b/README.md index 794e756..4e8a85d 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,7 @@ Pure non-blocking (fiber based) strictly typed full-featured PHP driver for NATS ## Features - [NATS Core](https://docs.nats.io/nats-concepts/core-nats) + - [Connection](#connection) - [Publish-Subscribe](#pub-sub) - [Queues](#queues) - [Request-Reply](#request-reply) @@ -58,6 +59,28 @@ composer require thesis/nats The library implements the full functionality of NATS Core, including pub-sub, queues and request–reply. +#### Connection + +`Nats\Config::fromURI()` takes `[scheme://][user[:password]@]host[:port][,host[:port]...][?options]`. The scheme is `nats://` (`tcp://` is an alias, and it may be omitted), +`tls://`, `ws://` and `wss://` are rejected as not supported yet. A host without a port connects to `4222`, an IPv6 host goes in brackets (`[::1]:4222`), +and the credentials are percent-encoded like in any url (`p%40ss` for `p@ss`). The servers of a list are tried in order until one accepts the connection; +each one may repeat the scheme and the credentials, as in nats.go (`nats://u:p@h1:4222,nats://u:p@h2:4222`): one user is sent to every server, so different credentials are rejected. The options are the snake_case names of the `Nats\Config` parameters, +e.g. `connection_timeout` (seconds, fractions allowed), `request_timeout`, `verbose`, `no_responders`, `ping`, `jwt`, `nkey`, `client_name`: + +```php +jetStream()->accountInfo()); +``` + #### Pub-Sub ```php diff --git a/src/Config.php b/src/Config.php index 11f8ba9..5750c70 100644 --- a/src/Config.php +++ b/src/Config.php @@ -4,6 +4,7 @@ namespace Thesis\Nats; +use Thesis\Nats\Internal\Connection\Uri; use Thesis\Nats\Internal\Validation\Validate; /** @@ -12,9 +13,7 @@ final readonly class Config { private const string DEFAULT_URL = '127.0.0.1:4222'; - private const string DEFAULT_HOST = '127.0.0.1'; - private const int DEFAULT_PORT = 4222; - private const int DEFAULT_CONNECTION_TIMEOUT = 10; + private const float DEFAULT_CONNECTION_TIMEOUT = 10.0; private const int DEFAULT_PING_INTERVAL = 10000; private const int DEFAULT_MAX_PINGS = 5; private const string DEFAULT_CLIENT_NAME = 'thesis/nats'; @@ -60,6 +59,8 @@ public function __construct( public string $clientName = self::DEFAULT_CLIENT_NAME, public ?float $requestTimeout = null, ) { + Validate::duration($connectionTimeout, 'connection timeout'); + if ($requestTimeout !== null) { Validate::duration($requestTimeout, 'request timeout'); } @@ -78,120 +79,29 @@ public static function default(): self */ public static function fromURI(#[\SensitiveParameter] string $uri): self { - $components = parse_url($uri); - - if ($components === false) { - throw new \InvalidArgumentException("The uri '{$uri}' is invalid."); - } - - $query = []; - if (isset($components['query']) && $components['query'] !== '') { - parse_str($components['query'], $query); - } - - $connectionTimeout = self::DEFAULT_CONNECTION_TIMEOUT; - if (isset($query['connection_timeout']) && is_numeric($query['connection_timeout']) && (int) $query['connection_timeout'] > 0) { - /** @var positive-int $connectionTimeout */ - $connectionTimeout = (int) $query['connection_timeout']; - } - - $tcpNoDelay = true; - if (isset($query['tcp_nodelay'])) { - $tcpNoDelay = filter_var($query['tcp_nodelay'], FILTER_VALIDATE_BOOL); - } - - $verbose = false; - if (isset($query['verbose'])) { - $verbose = filter_var($query['verbose'], FILTER_VALIDATE_BOOL); - } - - $pedantic = false; - if (isset($query['pedantic'])) { - $pedantic = filter_var($query['pedantic'], FILTER_VALIDATE_BOOL); - } - - $noResponders = false; - if (isset($query['no_responders'])) { - $noResponders = filter_var($query['no_responders'], FILTER_VALIDATE_BOOL); - } - - $ping = self::DEFAULT_PING_INTERVAL; - if (isset($query['ping']) && is_numeric($query['ping'])) { - /** @var ?positive-int $ping */ - $ping = match ($interval = (int) $query['ping']) { - -1 => null, - default => $interval < 0 ? self::DEFAULT_PING_INTERVAL : $interval, - }; - } - - $maxPings = self::DEFAULT_MAX_PINGS; - if (isset($query['max_pings']) && is_numeric($query['max_pings']) && (int) $query['max_pings'] > 0) { - /** @var positive-int $maxPings */ - $maxPings = (int) $query['max_pings']; - } - - $jetStreamDomain = null; - if (isset($query['jetstream_domain']) && \is_string($query['jetstream_domain']) && $query['jetstream_domain'] !== '') { - $jetStreamDomain = $query['jetstream_domain']; - } - - $port = self::DEFAULT_PORT; - if (isset($components['port']) && $components['port'] > 0) { - $port = $components['port']; - } - - $urls = []; - foreach (explode(',', $components['host'] ?? '') as $host) { - $hostport = explode(':', $host); - $urls[] = \sprintf('%s:%d', $hostport[0] === '' ? self::DEFAULT_HOST : $hostport[0], (int) ($hostport[1] ?? $port)); - } - - $user = null; - if (isset($components['user']) && $components['user'] !== '') { - $user = $components['user']; - } - - $password = null; - if (isset($components['pass']) && $components['pass'] !== '') { - $password = $components['pass']; - } - - $jwt = null; - if (isset($query['jwt']) && \is_string($query['jwt']) && $query['jwt'] !== '') { - $jwt = $query['jwt']; - } - - $nkey = null; - if (isset($query['nkey']) && \is_string($query['nkey']) && $query['nkey'] !== '') { - $nkey = $query['nkey']; - } - - $clientName = self::DEFAULT_CLIENT_NAME; - if (isset($query['client_name']) && \is_string($query['client_name']) && $query['client_name'] !== '') { - $clientName = $query['client_name']; - } - - $requestTimeout = null; - if (isset($query['request_timeout']) && is_numeric($query['request_timeout']) && (float) $query['request_timeout'] > 0) { - $requestTimeout = (float) $query['request_timeout']; - } + $parsed = Uri::parse($uri); + $query = $parsed->query; return new self( - urls: $urls, - verbose: $verbose, - pedantic: $pedantic, - connectionTimeout: $connectionTimeout, - user: $user, - password: $password, - jwt: $jwt, - nkey: $nkey, - tcpNoDelay: $tcpNoDelay, - noResponders: $noResponders, - ping: $ping, - maxPings: $maxPings, - jetStreamDomain: $jetStreamDomain, - clientName: $clientName, - requestTimeout: $requestTimeout, + urls: $parsed->urls, + verbose: $query->verbose ?? false, + pedantic: $query->pedantic ?? false, + connectionTimeout: $query->connectionTimeout ?? self::DEFAULT_CONNECTION_TIMEOUT, + user: $parsed->user, + password: $parsed->password, + jwt: $query->jwt, + nkey: $query->nkey, + tcpNoDelay: $query->tcpNoDelay ?? true, + noResponders: $query->noResponders ?? false, + ping: match (true) { + $query->ping === -1 => null, + $query->ping !== null && $query->ping > 0 => $query->ping, + default => self::DEFAULT_PING_INTERVAL, + }, + maxPings: $query->maxPings ?? self::DEFAULT_MAX_PINGS, + jetStreamDomain: $query->jetStreamDomain, + clientName: $query->clientName ?? self::DEFAULT_CLIENT_NAME, + requestTimeout: $query->requestTimeout, ); } @@ -204,7 +114,7 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self * nkey?: non-empty-string, * verbose?: bool, * pedantic?: bool, - * connection_timeout?: positive-int, + * connection_timeout?: float, * tcp_nodelay?: bool, * no_responders?: bool, * ping?: positive-int, diff --git a/src/Internal/Connection/Query.php b/src/Internal/Connection/Query.php new file mode 100644 index 0000000..82da1d4 --- /dev/null +++ b/src/Internal/Connection/Query.php @@ -0,0 +1,123 @@ + $options + */ + private static function bool(array $options, string $name): ?bool + { + if (!isset($options[$name])) { + return null; + } + + return filter_var($options[$name], FILTER_VALIDATE_BOOL, FILTER_NULL_ON_FAILURE); + } + + /** + * @param array $options + */ + private static function int(array $options, string $name): ?int + { + if (!isset($options[$name])) { + return null; + } + + $value = filter_var($options[$name], FILTER_VALIDATE_INT); + + return $value !== false ? $value : null; + } + + /** + * @param array $options + * @return ?positive-int + */ + private static function positiveInt(array $options, string $name): ?int + { + $value = self::int($options, $name); + + return $value !== null && $value > 0 ? $value : null; + } + + /** + * @param array $options + */ + private static function duration(array $options, string $name): ?float + { + if (!isset($options[$name])) { + return null; + } + + $value = filter_var($options[$name], FILTER_VALIDATE_FLOAT); + + return $value !== false && is_finite($value) && $value > 0 ? $value : null; + } + + /** + * @param array $options + * @return ?non-empty-string + */ + private static function string(#[\SensitiveParameter] array $options, string $name): ?string + { + $value = $options[$name] ?? null; + + return \is_string($value) && $value !== '' ? $value : null; + } +} diff --git a/src/Internal/Connection/Uri.php b/src/Internal/Connection/Uri.php new file mode 100644 index 0000000..b5b7040 --- /dev/null +++ b/src/Internal/Connection/Uri.php @@ -0,0 +1,203 @@ + */ + private const array SCHEMES = ['nats', 'tcp']; + + /** @var list */ + private const array UNSUPPORTED_SCHEMES = ['tls', 'ws', 'wss']; + + /** + * @param non-empty-list $urls in the `host:port` form, an IPv6 host in brackets + * @param ?non-empty-string $user + * @param ?non-empty-string $password + */ + private function __construct( + public array $urls, + public ?string $user, + public ?string $password, + public Query $query, + ) {} + + /** + * @throws \InvalidArgumentException + */ + public static function parse(#[\SensitiveParameter] string $uri): self + { + $rest = trim($uri); + + $fragment = strpos($rest, '#'); + if ($fragment !== false) { + $rest = substr($rest, 0, $fragment); + } + + [$rest, $options] = self::split($rest, '?'); + $query = Query::parse($options); + + $scheme = null; + $credentials = null; + $urls = []; + + // The list is split first: every server is parsed on its own, so its scheme and credentials + // cannot be taken for a part of a neighbour (a password with a comma must be percent-encoded). + $servers = explode(',', $rest); + + foreach ($servers as $server) { + $server = trim($server); + + [$serverScheme, $server] = self::scheme($server) ?? [$scheme ?? self::DEFAULT_SCHEME, $server]; + if ($scheme !== null && $serverScheme !== $scheme) { + throw new \InvalidArgumentException("The uri mixes the schemes '{$scheme}' and '{$serverScheme}'."); + } + + $scheme = $serverScheme; + + $at = strrpos($server, '@'); + if ($at !== false) { + $serverCredentials = self::credentials(substr($server, 0, $at)); + $server = substr($server, $at + 1); + + // A single user and password are sent to every server: the nats.go form repeats them, any other is refused. + if ($credentials !== null && $serverCredentials !== $credentials) { + throw new \InvalidArgumentException('The servers of the uri must share the same credentials.'); + } + + $credentials = $serverCredentials; + } + + if (str_ends_with($server, '/')) { + $server = substr($server, 0, -1); + } + + // A lone empty server is the default one, as in "nats://", an empty one in a list is likely a typo. + if ($server === '' && \count($servers) > 1) { + throw new \InvalidArgumentException('The uri has an empty server in the list.'); + } + + $urls[] = self::server($server); + } + + [$user, $password] = $credentials ?? [null, null]; + + return new self($urls, $user, $password, $query); + } + + /** + * @return ?array{non-empty-string, string} + * @throws \InvalidArgumentException + */ + private static function scheme(string $uri): ?array + { + if (preg_match('/^([a-z][a-z0-9+.-]*):\/\//i', $uri, $matches) !== 1) { + return null; + } + + $scheme = strtolower($matches[1]); + + if (\in_array($scheme, self::UNSUPPORTED_SCHEMES, true)) { + throw new \InvalidArgumentException("The scheme '{$scheme}' is not supported yet."); + } + + if (!\in_array($scheme, self::SCHEMES, true)) { + throw new \InvalidArgumentException(\sprintf("The scheme '%s' is unknown, expected one of: %s.", $scheme, implode(', ', self::SCHEMES))); + } + + // tcp:// is kept for compatibility, both mean a plain connection. + return [$scheme === 'tcp' ? self::DEFAULT_SCHEME : $scheme, substr($uri, \strlen($matches[0]))]; + } + + /** + * @return array{?non-empty-string, ?non-empty-string} + */ + private static function credentials(#[\SensitiveParameter] string $userinfo): array + { + [$user, $password] = self::split($userinfo, ':'); + $user = rawurldecode($user); + $password = rawurldecode($password); + + return [ + $user !== '' ? $user : null, + $password !== '' ? $password : null, + ]; + } + + /** + * @return non-empty-string + * @throws \InvalidArgumentException + */ + private static function server(string $server): string + { + if (str_starts_with($server, '[')) { + $end = strpos($server, ']'); + if ($end === false) { + throw new \InvalidArgumentException("The server '{$server}' has an unterminated IPv6 address."); + } + + $host = substr($server, 0, $end + 1); + $port = substr($server, $end + 1); + + if ($port !== '' && !str_starts_with($port, ':')) { + throw new \InvalidArgumentException("The server '{$server}' is invalid."); + } + + $port = substr($port, 1); + } else { + [$host, $port] = self::split($server, ':'); + + if (str_contains($port, ':')) { + throw new \InvalidArgumentException("The server '{$server}' is invalid: an IPv6 address must be enclosed in brackets."); + } + } + + if ($host !== '' && preg_match('/[\/?#@\s]/', $host) === 1) { + throw new \InvalidArgumentException("The server '{$server}' has an invalid host."); + } + + return \sprintf('%s:%d', $host === '' ? self::DEFAULT_HOST : $host, self::port($server, $port)); + } + + /** + * @return int<1, 65535> + * @throws \InvalidArgumentException + */ + private static function port(string $server, string $port): int + { + if ($port === '') { + return self::DEFAULT_PORT; + } + + if (preg_match('/^\d+$/', $port) !== 1 || (int) $port < 1 || (int) $port > 65535) { + throw new \InvalidArgumentException("The server '{$server}' has an invalid port '{$port}'."); + } + + return (int) $port; + } + + /** + * @param non-empty-string $separator + * @return array{string, string} the parts before and after the first separator, the second is empty without one + */ + private static function split(#[\SensitiveParameter] string $value, string $separator): array + { + $position = strpos($value, $separator); + + return $position === false + ? [$value, ''] + : [substr($value, 0, $position), substr($value, $position + \strlen($separator))]; + } +} diff --git a/tests/ClientTest.php b/tests/ClientTest.php index b67be12..6b0e49f 100644 --- a/tests/ClientTest.php +++ b/tests/ClientTest.php @@ -390,6 +390,25 @@ public function testConnectFailsWhenServerDoesNotAnswerHandshake(): void } } + public function testConnectsWithServerListInNatsGoForm(): void + { + // Every server repeats its scheme and credentials, the first one is down and the second has no port, as nats.go accepts it. + // The first password of user:Pswd1 is percent-encoded ("P" is %50), as in any url, so both decode to the same one. + $refused = self::reserveFreePort(); + $host = explode(':', $this->natsAddress())[0]; + + $client = $this->clientWithConfig(Config::fromURI("nats://user:%50swd1@{$refused}, nats://user:Pswd1@{$host}/?connection_timeout=1")); + $id = Id\generateUniqueId(); + + /** @var DeferredFuture $deferred */ + $deferred = new DeferredFuture(); + + $client->subscribe($id, $deferred->complete(...)); + $client->publish($id, new Message('ok')); + + self::assertSame('ok', $deferred->getFuture()->await(new TimeoutCancellation(5))->message->payload); + } + public function testConnectsToServerWithMinimalInfo(): void { // Old servers (before 2.2) and some proxies send only a part of the INFO fields. diff --git a/tests/ConfigTest.php b/tests/ConfigTest.php index ef91f7a..dc65a59 100644 --- a/tests/ConfigTest.php +++ b/tests/ConfigTest.php @@ -7,8 +7,12 @@ use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\Attributes\TestWith; use PHPUnit\Framework\TestCase; +use Thesis\Nats\Internal\Connection\Query; +use Thesis\Nats\Internal\Connection\Uri; #[CoversClass(Config::class)] +#[CoversClass(Uri::class)] +#[CoversClass(Query::class)] final class ConfigTest extends TestCase { /** @@ -195,11 +199,134 @@ final class ConfigTest extends TestCase new Config(), ], )] + #[TestWith( + [ + 'tcp://127.0.0.1:4222?connection_timeout=0.5', + new Config(connectionTimeout: 0.5), + ], + )] + #[TestWith( + [ + 'nats://user:p%40ss%3Aw%2Frd@127.0.0.1:4222', + new Config(user: 'user', password: 'p@ss:w/rd'), + ], + )] + #[TestWith( + [ + 'NATS://h1,h2:4223', + new Config(urls: ['h1:4222', 'h2:4223']), + ], + )] + #[TestWith( + [ + 'nats://admin:secret@h1:4222, nats://h2:4223,tcp://h3/?verbose=true', + new Config(urls: ['h1:4222', 'h2:4223', 'h3:4222'], verbose: true, user: 'admin', password: 'secret'), + ], + )] + #[TestWith( + [ + 'h1:4223', + new Config(urls: ['h1:4223']), + ], + )] + #[TestWith( + [ + 'nats://[::1]:4223,[fe80::1]', + new Config(urls: ['[::1]:4223', '[fe80::1]:4222']), + ], + )] + #[TestWith( + [ + 'nats://', + new Config(), + ], + )] + #[TestWith( + [ + 'nats://:4223', + new Config(urls: ['127.0.0.1:4223']), + ], + )] + #[TestWith( + [ + 'nats://127.0.0.1:4222?ping=-1&max_pings=3&jetstream_domain=hub', + new Config(ping: null, maxPings: 3, jetStreamDomain: 'hub'), + ], + )] + #[TestWith( + [ + 'nats://127.0.0.1:4222?ping=0&max_pings=0&connection_timeout=INF&request_timeout=NAN&client_name=', + new Config(), + ], + )] + #[TestWith( + [ + 'nats://127.0.0.1:4222?ping=1.5&max_pings=x&verbose=maybe&tcp_nodelay=maybe&jwt[]=x', + new Config(), + ], + )] + #[TestWith( + [ + 'nats://127.0.0.1:4222?verbose=1&pedantic=yes&tcp_nodelay=off&no_responders=on', + new Config(verbose: true, pedantic: true, tcpNoDelay: false, noResponders: true), + ], + )] + #[TestWith( + [ + 'nats://user:pass@h1:4222,nats://user:pass@h2:4223', + new Config(urls: ['h1:4222', 'h2:4223'], user: 'user', password: 'pass'), + ], + )] + #[TestWith( + [ + 'nats://h1,nats://user:p%2Cass@h2', + new Config(urls: ['h1:4222', 'h2:4222'], user: 'user', password: 'p,ass'), + ], + )] public function testFromURI(string $uri, Config $config): void { self::assertEquals($config, Config::fromURI($uri)); } + /** + * @param non-empty-string $uri + */ + #[TestWith(['tls://127.0.0.1:4222'])] + #[TestWith(['ws://127.0.0.1:4222'])] + #[TestWith(['wss://127.0.0.1:4222'])] + #[TestWith(['http://127.0.0.1:4222'])] + #[TestWith(['nats://h1:4222,ws://h2:4222'])] + #[TestWith(['nats://user:pass@h1:4222,tls://user:pass@h2:4222'])] + #[TestWith(['nats://h1:4222,tls://user:pass@h2:4222'])] + #[TestWith(['nats://user:pass@h1:4222,nats://user:other@h2:4222'])] + #[TestWith(['nats://user:pass@h1:4222,nats://h2:4222@h3'])] + #[TestWith(['nats://,h2:4222'])] + #[TestWith(['nats://h1:4222,,h2:4222'])] + #[TestWith(['nats://h1:4222,'])] + #[TestWith(['nats://h1:port'])] + #[TestWith(['nats://h1:0'])] + #[TestWith(['nats://h1:65536'])] + #[TestWith(['nats://h1:-1'])] + #[TestWith(['nats://::1'])] + #[TestWith(['nats://[::1'])] + #[TestWith(['nats://[::1]4222'])] + #[TestWith(['nats://h1:4222/path'])] + public function testFromURIRejectsInvalidUri(string $uri): void + { + $this->expectException(\InvalidArgumentException::class); + Config::fromURI($uri); + } + + #[TestWith([0.0])] + #[TestWith([-1.0])] + #[TestWith([INF])] + #[TestWith([NAN])] + public function testConnectionTimeoutMustBePositive(float $timeout): void + { + $this->expectException(\InvalidArgumentException::class); + new Config(connectionTimeout: $timeout); + } + public function testFromArrayWithJwtAndNkey(): void { $config = Config::fromArray([