This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 90e38b10f feat(node): support connection string syntax for nodejs
client creation (#3917)
90e38b10f is described below
commit 90e38b10f036ea174f8c049a586e5baadac8c0d5
Author: T1B0 <[email protected]>
AuthorDate: Fri Aug 28 17:58:02 2026 +0200
feat(node): support connection string syntax for nodejs client creation
(#3917)
---
foreign/node/README.md | 39 ++
foreign/node/src/client/client.config.test.ts | 44 +++
foreign/node/src/client/client.config.ts | 33 +-
.../src/client/client.connection-string.test.ts | 310 ++++++++++++++++
.../node/src/client/client.connection-string.ts | 226 ++++++++++++
foreign/node/src/client/client.connection.test.ts | 117 +++++-
foreign/node/src/client/client.connection.ts | 39 +-
foreign/node/src/client/client.socket.ts | 5 +-
foreign/node/src/client/client.ts | 57 ++-
foreign/node/src/client/client.type.ts | 25 +-
foreign/node/src/client/index.ts | 1 +
foreign/node/src/{client/index.ts => constant.ts} | 7 +-
foreign/node/src/duration.utils.test.ts | 180 ++++++++++
foreign/node/src/duration.utils.ts | 393 +++++++++++++++++++++
foreign/node/src/e2e/tcp.connection-string.e2e.ts | 146 ++++++++
foreign/node/src/e2e/tls.system.e2e.ts | 64 ++++
foreign/node/src/stream/consumer-stream.ts | 6 +-
17 files changed, 1629 insertions(+), 63 deletions(-)
diff --git a/foreign/node/README.md b/foreign/node/README.md
index 169500437..5c0f9052f 100644
--- a/foreign/node/README.md
+++ b/foreign/node/README.md
@@ -67,6 +67,8 @@ new session.
The client pings every `heartbeatInterval` milliseconds, 5000 by default, which
keeps an idle session alive when the server's `[heartbeat]` eviction is
enabled.
+`heartbeatInterval` also accepts a duration expression such as `"10s"` or
+`"1h 30m"`, like the Rust SDK.
The server evicts a connection silent for 36 s, which is 1.2 x its 30 s
heartbeat interval. Raising the client interval past that window, or setting it
to 0 to disable client heartbeats, exposes an idle consumer-group member to
@@ -105,6 +107,43 @@ const client = new Client({
const stats = await client.system.getStats();
```
+### Connection strings
+
+Every client constructor (except `SimpleClient` see note) also accepts a
+connection string instead of a config object:
+
+```ts
+import { Client } from "apache-iggy";
+
+const client = new Client("iggy://iggy:[email protected]:8090");
+const stats = await client.system.getStats();
+```
+
+Supported schemes are `iggy://` (TCP, default) and `iggy+tcp://`. Credentials
+are `username:password` or a single personal access token. Options mirror the
+other SDKs: `tls`, `tls_domain`, `tls_ca_file`, `reconnection_retries`,
+`reconnection_interval`, `heartbeat_interval` and `nodelay`.
`reestablish_after`
+is accepted for format compatibility but has no Node equivalent.
+
+note: `SimpleClient` does not accept a connection string: it wraps an existing
+`RawClient` instance rather than building one from configuration. Pass the
+connection string to `Client`, `SingleClient` or `getRawClient` and hand the
+resulting raw client to `SimpleClient` if needed.
+
+### option limits
+
+| option | limit |
+| --- | --- |
+| `reconnection_retries` | integer up to `4294967295` (u32 max); larger values
are rejected like Rust's u32 overflow, and `unlimited` maps to this ceiling.
Defaults to unlimited |
+| `heartbeat_interval` | duration up to `2147483647ms` (Node's largest timer
delay); `0` disables heartbeats |
+| `reconnection_interval` | positive duration (`ms`, `s`, `m`, `h`) up to
`2147483647ms` (Node's largest timer delay); zero spellings are rejected.
Defaults to `1s` |
+| port in the authority | decimal up to `65535` |
+
+Durations accept the same expressions as the Rust SDK, for example `500ms`,
+`10s`, `1h 30m`, `5d`, `2w`, `1y`; matching is case-insensitive and
+`0`, `unlimited`, `disabled` and `none` map to zero. Unit-less numbers such as
+`5` are rejected.
+
## use sources
### Install
diff --git a/foreign/node/src/client/client.config.test.ts
b/foreign/node/src/client/client.config.test.ts
index 9022ab080..f2404c703 100644
--- a/foreign/node/src/client/client.config.test.ts
+++ b/foreign/node/src/client/client.config.test.ts
@@ -17,6 +17,7 @@
import assert from 'node:assert/strict';
import { describe, it } from 'node:test';
+import { MAX_U32 } from '../constant.js';
import type { ClientConfig } from './client.type.js';
import {
DEFAULT_HEARTBEAT_INTERVAL,
@@ -74,6 +75,49 @@ describe('normalizeClientConfig', () => {
);
});
+ it('accepts a usable reconnect interval', () => {
+ const reconnect = { enabled: true, interval: 1000, maxRetries: 3 };
+ assert.deepEqual(
+ normalizeClientConfig({ ...config(), reconnect }).reconnect,
+ reconnect
+ );
+ });
+
+ it('rejects unusable reconnect intervals', () => {
+ for (const interval of [
+ 0, -1000, Number.NaN, 1.5, 2_147_483_648, Number.MAX_VALUE
+ ])
+ assert.throws(
+ () => normalizeClientConfig({
+ ...config(),
+ reconnect: { enabled: true, interval, maxRetries: 1 }
+ }),
+ /reconnect\.interval/
+ );
+ });
+
+ it('skips the interval check when reconnect is disabled', () => {
+ assert.doesNotThrow(() =>
+ normalizeClientConfig({
+ ...config(),
+ reconnect: { enabled: false, interval: 0, maxRetries: 0 }
+ })
+ );
+ });
+
+ it('rejects unusable reconnect maxRetries', () => {
+ for (const maxRetries of [
+ -1, Number.NaN, 1.5, MAX_U32 + 1, Number.MAX_VALUE
+ ])
+ assert.throws(
+ () => normalizeClientConfig({
+ ...config(),
+ reconnect: { enabled: true, interval: 1000, maxRetries }
+ }),
+ /reconnect\.maxRetries/
+ );
+ });
+
it('restricts the client to one pooled connection', () => {
const normalized = normalizeClientConfig(config());
assert.deepEqual(normalized.poolSize, { min: 1, max: 1 });
diff --git a/foreign/node/src/client/client.config.ts
b/foreign/node/src/client/client.config.ts
index 7c94bdbe4..8fe48e658 100644
--- a/foreign/node/src/client/client.config.ts
+++ b/foreign/node/src/client/client.config.ts
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-import type { ClientConfig } from './client.type.js';
+import { MAX_U32 } from '../constant.js';
+import type { ClientConfig, ClientConfigOrString } from './client.type.js';
+import { parseConnectionString } from './client.connection-string.js';
export const DEFAULT_MAX_RESPONSE_FRAME_SIZE = 64 * 1024 * 1024;
@@ -29,8 +31,11 @@ export const DEFAULT_HEARTBEAT_INTERVAL = 5 * 1000;
export const MAX_HEARTBEAT_INTERVAL = 2_147_483_647;
export const normalizeClientConfig = (
- config: ClientConfig
+ config: ClientConfigOrString
): ClientConfig => {
+ if (typeof config === 'string')
+ config = parseConnectionString(config);
+
const maxResponseFrameSize =
config.maxResponseFrameSize ?? DEFAULT_MAX_RESPONSE_FRAME_SIZE;
if (!Number.isSafeInteger(maxResponseFrameSize) ||
@@ -57,6 +62,30 @@ export const normalizeClientConfig = (
`heartbeatInterval must be a safe integer of milliseconds between 0 and
${MAX_HEARTBEAT_INTERVAL} (0 disables heartbeats)`
);
+ // Unlike the heartbeat, 0 is not a disable here: an immediate retry delay
+ // turns reconnection into a hot loop. The ceiling guards the same
+ // setInterval clamp, which would turn a long backoff into a 1 ms spin.
+ // A disabled reconnect never schedules a retry, so the interval check
+ // applies only when enabled; callers commonly pass a zero interval with
+ // enabled: false. maxRetries stays bounded either way, matching the
+ // connection-string path.
+ if (config.reconnect !== undefined) {
+ const { enabled, interval, maxRetries } = config.reconnect;
+ if (!Number.isSafeInteger(maxRetries) ||
+ maxRetries < 0 ||
+ maxRetries > MAX_U32)
+ throw new TypeError(
+ `reconnect.maxRetries must be a non-negative integer of at most
${MAX_U32}`
+ );
+ if (enabled &&
+ (!Number.isSafeInteger(interval) ||
+ interval < 1 ||
+ interval > MAX_HEARTBEAT_INTERVAL))
+ throw new TypeError(
+ `reconnect.interval must be a safe integer of milliseconds between 1
and ${MAX_HEARTBEAT_INTERVAL}`
+ );
+ }
+
return {
...config,
options: { ...config.options },
diff --git a/foreign/node/src/client/client.connection-string.test.ts
b/foreign/node/src/client/client.connection-string.test.ts
new file mode 100644
index 000000000..09198e781
--- /dev/null
+++ b/foreign/node/src/client/client.connection-string.test.ts
@@ -0,0 +1,310 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import { MAX_U32 } from '../constant.js';
+import {
+ parseConnectionString,
+ parseDuration
+} from './client.connection-string.js';
+import {
+ DEFAULT_HEARTBEAT_INTERVAL,
+ normalizeClientConfig
+} from './client.config.js';
+
+describe('parseConnectionString', () => {
+ it('parses the default scheme with password credentials', () => {
+ assert.deepEqual(
+ parseConnectionString('iggy://iggy:[email protected]:8090'),
+ {
+ transport: 'TCP',
+ options: { host: '127.0.0.1', port: 8090 },
+ credentials: { username: 'iggy', password: 'secret' },
+ reconnect: { enabled: true, interval: 1000, maxRetries: MAX_U32 }
+ }
+ );
+ });
+
+ it('parses the explicit tcp scheme with a personal access token', () => {
+ assert.deepEqual(
+
parseConnectionString('iggy+tcp://iggypat-1234567890abcdef@localhost:8090'),
+ {
+ transport: 'TCP',
+ options: { host: 'localhost', port: 8090 },
+ credentials: { token: 'iggypat-1234567890abcdef' },
+ reconnect: { enabled: true, interval: 1000, maxRetries: MAX_U32 }
+ }
+ );
+ });
+
+ it('maps tls options to the TLS transport', () => {
+ assert.deepEqual(
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090?tls=true&tls_domain=iggy.apache.org'
+ ),
+ {
+ transport: 'TLS',
+ options: {
+ host: 'localhost',
+ port: 8090,
+ servername: 'iggy.apache.org'
+ },
+ credentials: { username: 'iggy', password: 'secret' },
+ reconnect: { enabled: true, interval: 1000, maxRetries: MAX_U32 }
+ }
+ );
+ });
+
+ it('maps reconnection and heartbeat options', () => {
+ assert.deepEqual(
+ parseConnectionString(
+ 'iggy+tcp://iggy:secret@localhost:8090' +
+
'?reconnection_retries=3&reconnection_interval=5s&heartbeat_interval=10s'
+ ),
+ {
+ transport: 'TCP',
+ options: { host: 'localhost', port: 8090 },
+ credentials: { username: 'iggy', password: 'secret' },
+ reconnect: {
+ enabled: true,
+ maxRetries: 3,
+ interval: 5000
+ },
+ heartbeatInterval: 10000
+ }
+ );
+ });
+
+ it('applies unlimited/1s reconnection defaults to partial options', () => {
+ // retries alone keep the 1s interval; interval alone keeps unlimited.
+ assert.deepEqual(
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090?reconnection_retries=3'
+ ).reconnect,
+ { enabled: true, interval: 1000, maxRetries: 3 }
+ );
+ assert.deepEqual(
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090?reconnection_interval=5s'
+ ).reconnect,
+ { enabled: true, interval: 5000, maxRetries: MAX_U32 }
+ );
+ });
+
+ it('maps nodelay to the socket option', () => {
+ assert.equal(
+ parseConnectionString('iggy://iggy:secret@localhost:8090?nodelay=true')
+ .options.noDelay,
+ true
+ );
+ });
+
+ it('maps unlimited retries to the u32 ceiling', () => {
+ assert.equal(
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090?reconnection_retries=unlimited'
+ ).reconnect?.maxRetries,
+ MAX_U32
+ );
+ });
+
+ it('accepts retry counts up to u32::MAX and rejects overflow', () => {
+ assert.equal(
+ parseConnectionString(
+ `iggy://iggy:secret@localhost:8090?reconnection_retries=${MAX_U32}`
+ ).reconnect?.maxRetries,
+ MAX_U32
+ );
+ for (const value of [
+ 'iggy://iggy:secret@localhost:8090?reconnection_retries=4294967296',
+ 'iggy://iggy:secret@localhost:8090?reconnection_retries=99999999999999'
+ ])
+ assert.throws(() => parseConnectionString(value), TypeError);
+ });
+
+ it('rejects a non-positive reconnection interval', () => {
+ // Zero spellings parse but are rejected by the positivity bound.
+ for (const value of ['0', '0ms', 'none'])
+ assert.throws(
+ () =>
+ parseConnectionString(
+ `iggy://iggy:secret@localhost:8090?reconnection_interval=${value}`
+ ),
+ /must be positive/
+ );
+ // Negative durations shall not parse
+ assert.throws(
+ () =>
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090?reconnection_interval=-1s'
+ ),
+ TypeError
+ );
+ });
+
+ it('ignores reestablish_after for format compatibility', () => {
+ assert.deepEqual(
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090?reestablish_after=10s'
+ ),
+ {
+ transport: 'TCP',
+ options: { host: 'localhost', port: 8090 },
+ credentials: { username: 'iggy', password: 'secret' },
+ reconnect: { enabled: true, interval: 1000, maxRetries: MAX_U32 }
+ }
+ );
+ });
+
+ it('rejects unsupported transports', () => {
+ for (const value of [
+ 'iggy+quic://iggy:secret@localhost:8090',
+ 'iggy+ws://iggy:secret@localhost:8090'
+ ])
+ assert.throws(
+ () => parseConnectionString(value),
+ /unsupported transport/
+ );
+ });
+
+ it('rejects malformed connection strings', () => {
+ for (const value of [
+ '',
+ 'iggy',
+ 'iggy://',
+ 'iggy://:secret@localhost:8090',
+ 'iggy://iggy:@localhost:8090',
+ 'iggy://iggy:secret@localhost',
+ 'iggy://iggy:secret@:8090',
+ 'iggy://iggy:secret@localhost:port',
+ 'iggy://iggy:secret@localhost:70000',
+ 'iggy://iggy:secret@localhost:8090?unknown=value',
+ 'iggy://iggy:secret@localhost:8090?tls=maybe',
+ 'iggy://iggy:secret@localhost:8090?reconnection_retries=three',
+ 'iggy://iggy:secret@[::1:8090',
+ 'iggy://iggy:secret@[]:8090',
+ 'iggy://iggy:secret@[::1]x:8090',
+ 'iggy://iggy:secret@2001:db8::1:8090',
+ 'iggy://iggy:secret@host:8090:9090',
+ 'iggy://iggy:secret@localhost:8090?',
+ 'iggy://iggy:secret@localhost:8090?&',
+ 'iggy://iggy:secret@localhost:8090?reestablish_after=garbage'
+ ])
+ assert.throws(() => parseConnectionString(value), TypeError);
+ });
+
+ it('never includes the connection string in error messages', () => {
+ const secrets = ['hunter2', 'iggypat-1234567890abcdef'];
+ for (const value of [
+ 'iggy://iggy:hunter2@localhost',
+ `iggy+tcp://iggypat-1234567890abcdef@localhost`,
+ 'iggy://iggy:hunter2@localhost:8090?unknown=value',
+ 'iggy://iggy:hunter2@localhost:8090?tls=maybe',
+ 'iggy://iggy:hunter2@localhost:8090?reconnection_retries=three',
+ 'iggy://iggy:hunter2@localhost:70000'
+ ]) {
+ try {
+ parseConnectionString(value);
+ assert.fail(`expected "${value}" to be rejected`);
+ } catch (error) {
+ assert.ok(error instanceof TypeError);
+ for (const secret of secrets)
+ assert.ok(
+ !error.message.includes(secret),
+ `error message leaked a secret: ${error.message}`
+ );
+ }
+ }
+ });
+
+ it('parses IPv6 host addresses without their brackets', () => {
+ assert.deepEqual(
+ parseConnectionString('iggy://iggy:secret@[::1]:8090').options,
+ { host: '::1', port: 8090 }
+ );
+ });
+
+ it('stores tls_ca_file as a path without reading it at parse time', () => {
+ assert.deepEqual(
+ parseConnectionString(
+ 'iggy://iggy:secret@localhost:8090' +
+ '?tls=true&tls_ca_file=/does/not/exist.pem'
+ ).options,
+ {
+ host: 'localhost',
+ port: 8090,
+ caFile: '/does/not/exist.pem'
+ }
+ );
+ });
+});
+
+describe('parseDuration', () => {
+ it('converts supported units to milliseconds', () => {
+ assert.equal(parseDuration('500ms'), 500);
+ assert.equal(parseDuration('5s'), 5000);
+ assert.equal(parseDuration('2m'), 120000);
+ assert.equal(parseDuration('1h'), 3600000);
+ assert.equal(parseDuration('0.5s'), 500);
+ assert.equal(parseDuration('1h 1m 1s'), 3661000);
+ assert.equal(parseDuration('1h30m'), 5400000);
+ assert.equal(parseDuration('5d'), 432000000);
+ assert.equal(parseDuration('2w'), 1209600000);
+ assert.equal(parseDuration('1y'), 31557600000);
+ assert.equal(parseDuration('5sec'), 5000);
+ assert.equal(parseDuration('5msec'), 5);
+ // Fractional results are rounded to whole milliseconds.
+ assert.equal(parseDuration('1.005s'), 1005);
+ assert.equal(parseDuration('5usec'), 0);
+ assert.equal(parseDuration('500nsec'), 0);
+ for (const zero of ['0', 'unlimited', 'disabled', 'none', 'UNLIMITED'])
+ assert.equal(parseDuration(zero), 0);
+ });
+
+ it('rejects unsupported durations', () => {
+ for (const value of ['5', '-1s', 'ms', '', 'abc', 's'])
+ assert.throws(() => parseDuration(value), /invalid duration/);
+ });
+});
+
+describe('normalizeClientConfig with connection strings', () => {
+ it('applies client defaults to the parsed config', () => {
+ const normalized =
normalizeClientConfig('iggy://iggy:secret@localhost:8090');
+
+ assert.equal(normalized.transport, 'TCP');
+ assert.equal(normalized.options.host, 'localhost');
+ assert.equal(normalized.options.port, 8090);
+ assert.deepEqual(normalized.credentials, {
+ username: 'iggy',
+ password: 'secret'
+ });
+ assert.equal(normalized.heartbeatInterval, DEFAULT_HEARTBEAT_INTERVAL);
+ assert.deepEqual(normalized.poolSize, { min: 1, max: 1 });
+ });
+
+ it('rejects reconnect intervals beyond the node timer ceiling', () => {
+ // Parses to 3_600_000_000 ms; setInterval would clamp it back to 1 ms.
+ assert.throws(
+ () =>
+ normalizeClientConfig(
+ 'iggy://iggy:secret@localhost:8090?reconnection_interval=1000h'
+ ),
+ /reconnect\.interval/
+ );
+ });
+});
diff --git a/foreign/node/src/client/client.connection-string.ts
b/foreign/node/src/client/client.connection-string.ts
new file mode 100644
index 000000000..75d0b67e2
--- /dev/null
+++ b/foreign/node/src/client/client.connection-string.ts
@@ -0,0 +1,226 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import { MAX_U32 } from '../constant.js';
+import {
+ nanosecondsToMilliseconds,
+ parseIggyDurationNanoseconds
+} from '../duration.utils.js';
+import type { ClientConfig, ReconnectOption } from './client.type.js';
+
+const DEFAULT_PROTOCOL = 'iggy';
+const SCHEME_PREFIX = 'iggy+';
+const SUPPORTED_PROTOCOLS = ['tcp'] as const;
+
+/** Reconnection defaults carried by every connection string.
+ *
+ * Mirrors TcpConnectionStringOptions: retries default to unlimited and the
+ * interval to 1s. Deliberately not DefaultReconnectOption, whose 12/5s pair
+ * only applies to object configs.
+ */
+const CONNECTION_STRING_RECONNECT: ReconnectOption = {
+ enabled: true,
+ interval: 1000,
+ maxRetries: MAX_U32
+};
+
+/** Parses a duration with the exact grammar of the Rust SDK's
+ * `IggyDuration::from_str`, returning whole milliseconds.
+ */
+export const parseDuration = (value: string): number => {
+ try {
+ return nanosecondsToMilliseconds(parseIggyDurationNanoseconds(value));
+ } catch (error) {
+ throw new TypeError(`invalid duration "${value}"`, { cause: error });
+ }
+};
+
+/**
+ * Parses an Iggy connection string into a client configuration.
+ *
+ * Supports `iggy://` and `iggy+tcp://`; the Node SDK implements TCP/TLS only.
+ * Credentials are either `username:password` or a single personal access
+ * token before the `@`. TLS is enabled with `tls=true`.
+ */
+export const parseConnectionString = (connectionString: string): ClientConfig
=> {
+ if (typeof connectionString !== 'string' || connectionString.length === 0)
+ throw new TypeError('connection string must be a non-empty string');
+
+ const protocolParts = connectionString.split('://');
+ if (protocolParts.length !== 2)
+ throw new TypeError('invalid connection string');
+
+ const scheme = protocolParts[0];
+ const protocol = scheme === DEFAULT_PROTOCOL
+ ? 'tcp'
+ : scheme.startsWith(SCHEME_PREFIX)
+ ? scheme.slice(SCHEME_PREFIX.length)
+ : undefined;
+ if (protocol === undefined)
+ throw new TypeError('invalid connection string');
+ if (!SUPPORTED_PROTOCOLS.includes(protocol as (typeof
SUPPORTED_PROTOCOLS)[number]))
+ throw new TypeError(
+ `unsupported transport "${protocol}", Node SDK supports tcp only`
+ );
+
+ const parts = protocolParts[1].split('@');
+ if (parts.length !== 2)
+ throw new TypeError('invalid connection string');
+
+ const credentials = parts[0].split(':');
+ const tokenCredentials = credentials.length === 1;
+ if (!tokenCredentials && credentials.length !== 2)
+ throw new TypeError('invalid connection string');
+
+ const username = credentials[0];
+ const password = credentials[1] ?? '';
+ if (!tokenCredentials && (username.length === 0 || password.length === 0))
+ throw new TypeError('invalid connection string');
+
+ const serverAndOptions = parts[1].split('?');
+ if (serverAndOptions.length > 2)
+ throw new TypeError('invalid connection string');
+
+ const serverAddress = serverAndOptions[0];
+ // One match covers both `[ipv6]:port` and `host:port`, where the
+ // unbracketed host may not contain a colon. Multi-colon authorities such
+ // as `2001:db8::1:8090` or `host:8090:9090` are rejected
+ const addressMatch =
/^(?:\[([^\]]+)\]|([^:\[\]]+)):(\d+)$/.exec(serverAddress);
+ if (!addressMatch)
+ throw new TypeError('invalid connection string');
+
+ const host = addressMatch[1] ?? addressMatch[2];
+ const port = Number(addressMatch[3]);
+ if (port > 65535)
+ throw new TypeError('invalid connection string');
+
+ const options: ParsedConnectionOptions = serverAndOptions.length === 2
+ ? parseConnectionOptions(serverAndOptions[1])
+ : {
+ tls: false,
+ reconnect: { ...CONNECTION_STRING_RECONNECT }
+ };
+ const { tls, reconnect, heartbeatInterval, ...transportOptions } = options;
+
+ const config: ClientConfig = {
+ transport: tls ? 'TLS' : 'TCP',
+ options: {
+ host,
+ port: Number(port),
+ ...transportOptions
+ },
+ credentials: tokenCredentials
+ ? { token: username }
+ : { username, password },
+ // Always present on connection strings: Rust applies its unlimited/1s
+ // reconnection defaults even without query options.
+ reconnect
+ };
+ if (heartbeatInterval !== undefined)
+ config.heartbeatInterval = heartbeatInterval;
+
+ return config;
+};
+
+type ParsedConnectionOptions = {
+ tls: boolean,
+ noDelay?: boolean,
+ servername?: string,
+ /** Path stored for connect-time reading, match Rust SDK. */
+ caFile?: string,
+ reconnect: ReconnectOption,
+ heartbeatInterval?: number
+};
+
+const parseConnectionOptions = (
+ optionsString: string
+): ParsedConnectionOptions => {
+ const parsed: ParsedConnectionOptions = {
+ tls: false,
+ reconnect: { ...CONNECTION_STRING_RECONNECT }
+ };
+ for (const option of optionsString.split('&')) {
+ const optionParts = option.split('=');
+ if (optionParts.length !== 2)
+ throw new TypeError('invalid connection string');
+ const [name, value] = optionParts;
+ switch (name) {
+ case 'tls':
+ parsed.tls = parseBoolean(name, value);
+ break;
+ case 'nodelay':
+ parsed.noDelay = parseBoolean(name, value);
+ break;
+ case 'tls_domain':
+ parsed.servername = value;
+ break;
+ case 'tls_ca_file':
+ // The path is stored unread; the certificate is loaded when the
+ // TLS socket is created, so plain-TCP configs never touch the
+ // filesystem (matches the Rust SDK).
+ parsed.caFile = value;
+ break;
+
+ case 'reconnection_retries': {
+ // Values above u32::MAX are rejected like the Rust SDK's u32
+ // overflow; otherwise they would act as a second, undocumented
+ // spelling of "unlimited".
+ const maxRetries = value === 'unlimited'
+ ? MAX_U32
+ : parseNumber(name, value);
+ if (maxRetries > MAX_U32)
+ throw new TypeError(
+ `option "${name}" must be at most ${MAX_U32}`
+ );
+ parsed.reconnect.maxRetries = maxRetries;
+ break;
+ }
+ case 'reconnection_interval': {
+ const interval = parseDuration(value);
+ // With retries defaulting to unlimited, a zero interval would turn
+ // the reconnect loop into an unbounded hot loop.
+ if (interval <= 0)
+ throw new TypeError(`option "${name}" must be positive`);
+ parsed.reconnect.interval = interval;
+ break;
+ }
+ case 'reestablish_after':
+ // No Node equivalent: validated as a duration like the Rust SDK
+ // does, then discarded.
+ parseDuration(value);
+ break;
+ case 'heartbeat_interval':
+ parsed.heartbeatInterval = parseDuration(value);
+ break;
+ default:
+ throw new TypeError(`unknown option "${name}"`);
+ }
+ }
+ return parsed;
+};
+
+const parseBoolean = (name: string, value: string): boolean => {
+ if (value !== 'true' && value !== 'false')
+ throw new TypeError(`option "${name}" must be true or false`);
+ return value === 'true';
+};
+
+const parseNumber = (name: string, value: string): number => {
+ if (!/^\d+$/.test(value))
+ throw new TypeError(`option "${name}" must be a non-negative integer`);
+ return Number(value);
+};
diff --git a/foreign/node/src/client/client.connection.test.ts
b/foreign/node/src/client/client.connection.test.ts
index b1269d19d..5e320b0ac 100644
--- a/foreign/node/src/client/client.connection.test.ts
+++ b/foreign/node/src/client/client.connection.test.ts
@@ -17,13 +17,18 @@
import assert from 'node:assert/strict';
import { once } from 'node:events';
+import { readFileSync } from 'node:fs';
import {
createServer,
type AddressInfo,
type Server,
- type Socket,
+ type Socket
} from 'node:net';
-import { describe, it } from 'node:test';
+import {
+ createServer as createTlsServer,
+ type TLSSocket
+} from 'node:tls';
+import { describe, it, before, after } from 'node:test';
import { ProtocolFrameError } from './client.frame.js';
import { IggyConnection } from './client.connection.js';
import type { ClientConfig } from './client.type.js';
@@ -31,6 +36,26 @@ import { Command, HEADER_SIZE, REPLY_OFFSET } from
'../wire/vsr/header.js';
const FRAME_LIMIT = 2 * HEADER_SIZE;
+const TLS_CERTIFICATE = readFileSync(
+ new URL('../../../../core/certs/iggy_cert.pem', import.meta.url)
+);
+const TLS_KEY = readFileSync(
+ new URL('../../../../core/certs/iggy_key.pem', import.meta.url)
+);
+const TLS_CA_CERTIFICATE = readFileSync(
+ new URL('../../../../core/certs/iggy_ca_cert.pem', import.meta.url)
+);
+
+const startTlsServer = async (): Promise<Server> => {
+ const server = createTlsServer({
+ cert: TLS_CERTIFICATE,
+ key: TLS_KEY
+ });
+ server.listen(0, '127.0.0.1');
+ await once(server, 'listening');
+ return server;
+};
+
const startServer = async (): Promise<Server> => {
const server = createServer();
server.listen(0, '127.0.0.1');
@@ -67,7 +92,16 @@ const closeConnection = async (
await new Promise<void>((resolve) => server.close(() => resolve()));
};
+let keepAlive: NodeJS.Timeout;
+
describe('IggyConnection', () => {
+
+ // Note:
+ // before node v24 Timeout.unref() would let eventloop exit before test end
+ // (tested against 22.x 23.x -> fail vs 24.x 26.x -> pass)
+ // this timeout prevent eventloop exit before this test end
+ before(() => { keepAlive = setInterval(() => {}, 10000) });
+
it('recognizes a connection established before connect is called',
async () => {
const server = await startServer();
@@ -342,8 +376,15 @@ describe('IggyConnection', () => {
(resolve) => server.close(() => resolve())
);
serverSocket?.destroy();
+ const retryStartedAt = Date.now();
await closed;
const error = await exhausted;
+ // Three retries at a 10 ms interval must spend at least 30 ms in
+ // backoff; a broken wait would redial back to back.
+ assert.ok(
+ Date.now() - retryStartedAt >= 30,
+ 'reconnect backoff did not elapse between retries'
+ );
assert.match(error.message, /reconnect maxRetries exceeded/);
await new Promise<void>((resolve) => setTimeout(resolve, 20));
assert.deepEqual(rejections, []);
@@ -813,4 +854,76 @@ describe('IggyConnection', () => {
}
}
);
+
+ it('rejects an unreadable tls_ca_file with a TypeError at socket creation',
+ () => {
+ assert.throws(
+ () =>
+ new IggyConnection({
+ transport: 'TLS',
+ options: {
+ host: '127.0.0.1',
+ port: 8090,
+ caFile: '/does/not/exist.pem'
+ },
+ credentials: { username: 'iggy', password: 'iggy' },
+ reconnect: { enabled: false, interval: 0, maxRetries: 0 }
+ }),
+ /cannot read tls_ca_file/
+ );
+ }
+ );
+
+ it('sends a DNS host as the SNI server name when none is set',
+ async () => {
+ const server = await startTlsServer();
+ const secureConnection =
+ once(server, 'secureConnection') as Promise<[TLSSocket]>;
+ const connection = new IggyConnection({
+ transport: 'TLS',
+ options: {
+ host: 'localhost',
+ port: (server.address() as AddressInfo).port,
+ ca: TLS_CA_CERTIFICATE
+ },
+ credentials: { username: 'iggy', password: 'iggy' },
+ reconnect: { enabled: false, interval: 0, maxRetries: 0 }
+ });
+ try {
+ await connection.connect();
+ assert.equal(connection.connected, true);
+ const [tlsSocket] = await secureConnection;
+ assert.equal(tlsSocket.servername, 'localhost');
+ } finally {
+ await closeConnection(connection, server);
+ }
+ }
+ );
+
+ it('omits SNI for IP literal hosts', async () => {
+ const server = await startTlsServer();
+ const secureConnection =
+ once(server, 'secureConnection') as Promise<[TLSSocket]>;
+ const connection = new IggyConnection({
+ transport: 'TLS',
+ options: {
+ host: '127.0.0.1',
+ port: (server.address() as AddressInfo).port,
+ rejectUnauthorized: false
+ },
+ credentials: { username: 'iggy', password: 'iggy' },
+ reconnect: { enabled: false, interval: 0, maxRetries: 0 }
+ });
+ try {
+ await connection.connect();
+ assert.equal(connection.connected, true);
+ const [tlsSocket] = await secureConnection;
+ // Node reports a missing SNI name as false on the server side.
+ assert.ok(!tlsSocket.servername);
+ } finally {
+ await closeConnection(connection, server);
+ }
+ });
+
+ after(() => clearInterval(keepAlive));
});
diff --git a/foreign/node/src/client/client.connection.ts
b/foreign/node/src/client/client.connection.ts
index 49f7752f8..936c6f158 100644
--- a/foreign/node/src/client/client.connection.ts
+++ b/foreign/node/src/client/client.connection.ts
@@ -17,8 +17,9 @@
import { EventEmitter } from 'node:events';
import type { Socket } from 'node:net';
-import { createConnection } from 'node:net';
+import { createConnection, isIP } from 'node:net';
import { connect as TLSConnect } from 'node:tls';
+import { readFileSync } from 'node:fs';
import type { ClientConfig, TlsOption, TcpOption, ReconnectOption } from
"./client.type.js"
import { debug } from './client.debug.js';
import { DEFAULT_MAX_RESPONSE_FRAME_SIZE } from './client.config.js';
@@ -47,8 +48,26 @@ const createTcpSocket = (options: TcpOption): Socket => {
* @returns TLS socket
*/
const createTlsSocket = ({ port, ...options }: TlsOption): Socket => {
- const socket = TLSConnect(port, options);
- return socket;
+ const { caFile, ...tlsOptions } = options;
+ if (caFile !== undefined)
+ tlsOptions.ca = readCaCertificate(caFile);
+ // An SNI-routing terminator needs a server name in the ClientHello;
+ // IP literals are never sent as SNI, matching the Rust SDK's fallback.
+ if (typeof tlsOptions.host === 'string' &&
+ tlsOptions.servername === undefined &&
+ isIP(tlsOptions.host) === 0)
+ tlsOptions.servername = tlsOptions.host;
+ return TLSConnect(port, tlsOptions);
+};
+
+// A missing or unreadable CA file is a configuration error rather than a
+// transient I/O failure, hence the TypeError instead of the raw ENOENT.
+const readCaCertificate = (caFile: string): Buffer => {
+ try {
+ return readFileSync(caFile);
+ } catch {
+ throw new TypeError(`cannot read tls_ca_file "${caFile}"`);
+ }
};
/**
@@ -96,14 +115,16 @@ const DefaultReconnectOption: ReconnectOption = {
/**
* Waits before a reconnection attempt.
*
- * @param timer - Delay in milliseconds before recreating
- * @returns Promise resolving after the delay
+ * The timer is unref'd: a queued command fails fast on 'disconnected', so
+ * the retry loop is background work that must not keep the process alive.
+ *
+ * @param interval - Delay in milliseconds before dialing again
*/
-function waitForReconnect(timer = 1000): Promise<void> {
- return new Promise((resolve) => {
- setTimeout(resolve, timer);
+const waitForReconnect = (interval: number): Promise<void> =>
+ new Promise<void>((resolve) => {
+ const timeout = setTimeout(resolve, interval);
+ (timeout as NodeJS.Timeout).unref();
});
-}
/** Socket error with optional error code */
type SocketError = Error & { code?: string };
diff --git a/foreign/node/src/client/client.socket.ts
b/foreign/node/src/client/client.socket.ts
index 3924ba14a..06670e5c1 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -18,6 +18,7 @@
import { EventEmitter } from 'node:events';
import type {
ClientConfig,
+ ClientConfigOrString,
ClientCredentials, CommandResponse,
PasswordCredentials, RawClient, SendCommandOptions,
TokenCredentials
@@ -165,7 +166,7 @@ export class CommandResponseStream extends EventEmitter {
*
* @param options - Client configuration
*/
- constructor(options: ClientConfig) {
+ constructor(options: ClientConfigOrString) {
super();
const normalizedConfig = normalizeClientConfig(options);
this.options = normalizedConfig;
@@ -950,7 +951,7 @@ export class CommandResponseStream extends EventEmitter {
* @param options - Client configuration
* @returns RawClient instance
*/
-export function getRawClient(options: ClientConfig): RawClient {
+export function getRawClient(options: ClientConfigOrString): RawClient {
return new CommandResponseStream(options);
}
diff --git a/foreign/node/src/client/client.ts
b/foreign/node/src/client/client.ts
index fc642f5ef..25f877b13 100644
--- a/foreign/node/src/client/client.ts
+++ b/foreign/node/src/client/client.ts
@@ -16,28 +16,13 @@
// under the License.
import { createPool, type Pool } from 'generic-pool';
-import type { RawClient, ClientConfig } from "./client.type.js"
+import type { RawClient, ClientConfig, ClientConfigOrString } from
"./client.type.js"
import { getRawClient } from '../client/client.socket.js';
import { CommandAPI } from '../wire/command-set.js';
import { debug } from './client.debug.js';
import { normalizeClientConfig } from './client.config.js';
-/**
- * Creates a pool factory for managing RawClient instances.
- *
- * @param config - Client configuration
- * @returns Pool factory with create and destroy methods
- */
-const createPoolFactory = (config: ClientConfig) => ({
- create: async function () {
- return getRawClient(config);
- },
- destroy: async function (client: RawClient) {
- return client.destroy();
- }
-});
-
/**
* Creates a client provider that uses connection pooling.
* Automatically acquires and releases clients from the pool.
@@ -46,20 +31,33 @@ const createPoolFactory = (config: ClientConfig) => ({
* @returns Client provider and its connection pool
*/
const createPooledClientProvider = (config: ClientConfig) => {
+ // Constructed eagerly rather than inside the factory: generic-pool never
+ // rejects waiting acquires when a create fails, and keeps re-dispatching
+ // the doomed factory, which hangs every command. Normalization already
+ // pins this pool to exactly one pooled connection, so there is nothing
+ // lazy to preserve.
+ const client = getRawClient(config);
const minPoolSize = config.poolSize?.min || 1;
const maxPoolSize = config.poolSize?.max || 4;
- const pool = createPool(createPoolFactory(config), {
+ const pool = createPool({
+ create: async function () {
+ return client;
+ },
+ destroy: async function () {
+ return client.destroy();
+ }
+ }, {
min: minPoolSize,
max: maxPoolSize
});
const clientProvider = async () => {
- const client = await pool.acquire();
+ const pooled = await pool.acquire();
debug('client acquired from pool. pool size is', pool.size);
- client.once('finishQueue', () => {
- pool.release(client)
+ pooled.once('finishQueue', () => {
+ pool.release(pooled)
debug('client released to pool. pool size is', pool.size);
});
- return client;
+ return pooled;
}
return { clientProvider, pool };
};
@@ -78,9 +76,9 @@ export class Client extends CommandAPI {
/**
* Creates a new pooled client.
*
- * @param config - Client configuration
+ * @param config - Client configuration or connection string
*/
- constructor(config: ClientConfig) {
+ constructor(config: ClientConfigOrString) {
const normalizedConfig = normalizeClientConfig(config);
const { clientProvider, pool } =
createPooledClientProvider(normalizedConfig);
@@ -124,11 +122,12 @@ export class SingleClient extends CommandAPI {
/**
* Creates a new single-connection client.
*
- * @param config - Client configuration
+ * @param config - Client configuration or connection string
*/
- constructor(config: ClientConfig) {
- super(createSingleClientProvider(config));
- this._config = config;
+ constructor(config: ClientConfigOrString) {
+ const normalizedConfig = normalizeClientConfig(config);
+ super(createSingleClientProvider(normalizedConfig));
+ this._config = normalizedConfig;
}
/**
@@ -169,10 +168,10 @@ export class SimpleClient extends CommandAPI {
* Creates a SimpleClient with the given configuration.
* Convenience function for quickly creating a client.
*
- * @param config - Client configuration
+ * @param config - Client configuration or connection string
* @returns SimpleClient instance
*/
-export const getClient = async (config: ClientConfig) => {
+export const getClient = async (config: ClientConfigOrString) => {
const client = getRawClient(config);
return new SimpleClient(client);
};
diff --git a/foreign/node/src/client/client.type.ts
b/foreign/node/src/client/client.type.ts
index 42ae39239..ad3932f42 100644
--- a/foreign/node/src/client/client.type.ts
+++ b/foreign/node/src/client/client.type.ts
@@ -16,20 +16,22 @@
// under the License.
import type { Readable } from 'stream';
-import { type TcpSocketConnectOpts } from 'node:net';
+import { type TcpNetConnectOpts } from 'node:net';
import { type ConnectionOptions } from 'node:tls';
/**
* TCP socket connection options.
- * Alias for Node.js TcpSocketConnectOpts.
+ * Alias for Node.js TcpNetConnectOpts, what net.createConnection accepts.
*/
-export type TcpOption = TcpSocketConnectOpts;
+export type TcpOption = TcpNetConnectOpts;
/**
* TLS socket connection options.
- * Combines port number with Node.js TLS ConnectionOptions.
+ * Combines port number with Node.js TLS ConnectionOptions and the
+ * net.connect options tls.connect forwards at runtime. `caFile` is an SDK
+ * extension: a CA certificate path read when the TLS socket is created.
*/
-export type TlsOption = { port: number } & ConnectionOptions;
+export type TlsOption = { port: number } & ConnectionOptions &
Partial<TcpNetConnectOpts> & { caFile?: string };
/**
* Response from a command sent to the Iggy server.
@@ -165,6 +167,12 @@ export type PoolSizeOption = {
max?: number
}
+/**
+ * Client configuration or a connection string such as
+ * `iggy://username:password@host:port`.
+ */
+export type ClientConfigOrString = ClientConfig | string;
+
/**
* Complete client configuration for connecting to the Iggy server.
*/
@@ -179,12 +187,7 @@ export type ClientConfig = {
poolSize?: PoolSizeOption,
/** Automatic reconnection configuration */
reconnect?: ReconnectOption,
- /**
- * Interval for sending heartbeat pings in milliseconds, as an integer
- * between 0 and Node's timer ceiling. Defaults to 5000. Set to 0 to disable
- * client heartbeats; any other unusable value is rejected rather than
- * silently disabling them.
- */
+ /** Interval for sending heartbeat pings, in milliseconds */
heartbeatInterval?: number,
/** Maximum accepted response frame size in bytes */
maxResponseFrameSize?: number
diff --git a/foreign/node/src/client/index.ts b/foreign/node/src/client/index.ts
index d968e5ca0..e3430c4a9 100644
--- a/foreign/node/src/client/index.ts
+++ b/foreign/node/src/client/index.ts
@@ -17,6 +17,7 @@
export { Client, SimpleClient, SingleClient } from './client.js'
export * from './client.config.js';
+export { parseConnectionString } from './client.connection-string.js';
export * from './client.utils.js';
export * from './client.socket.js';
export * from './client.type.js';
diff --git a/foreign/node/src/client/index.ts b/foreign/node/src/constant.ts
similarity index 79%
copy from foreign/node/src/client/index.ts
copy to foreign/node/src/constant.ts
index d968e5ca0..bfc196335 100644
--- a/foreign/node/src/client/index.ts
+++ b/foreign/node/src/constant.ts
@@ -15,8 +15,5 @@
// specific language governing permissions and limitations
// under the License.
-export { Client, SimpleClient, SingleClient } from './client.js'
-export * from './client.config.js';
-export * from './client.utils.js';
-export * from './client.socket.js';
-export * from './client.type.js';
+/** Largest value representable by the Rust u32 type. */
+export const MAX_U32 = 4294967295;
diff --git a/foreign/node/src/duration.utils.test.ts
b/foreign/node/src/duration.utils.test.ts
new file mode 100644
index 000000000..992ff86a4
--- /dev/null
+++ b/foreign/node/src/duration.utils.test.ts
@@ -0,0 +1,180 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import {
+ DurationParseError,
+ nanosecondsToMilliseconds,
+ parseHumantimeDuration,
+ parseIggyDurationNanoseconds
+} from './duration.utils.js';
+
+const ns = (
+ seconds: number | bigint,
+ nanos = 0n
+): bigint => BigInt(seconds) * 1_000_000_000n + nanos;
+
+const parseNs = (input: string): bigint => {
+ const { seconds, nanoseconds } = parseHumantimeDuration(input);
+ return seconds * 1_000_000_000n + nanoseconds;
+};
+
+describe('parseHumantimeDuration', () => {
+ it('parses every unit spelling', () => {
+ const cases: [string, bigint][] = [
+ ['17nsec', ns(0n, 17n)],
+ ['17nanos', ns(0n, 17n)],
+ ['33ns', ns(0n, 33n)],
+ ['3usec', ns(0n, 3000n)],
+ ['78us', ns(0n, 78_000n)],
+ ['163µs', ns(0n, 163_000n)],
+ ['31msec', ns(0n, 31_000_000n)],
+ ['31millis', ns(0n, 31_000_000n)],
+ ['6ms', ns(0n, 6_000_000n)],
+ ['3000s', ns(3000n)],
+ ['300secs', ns(300n)],
+ ['50seconds', ns(50n)],
+ ['100m', ns(6000n)],
+ ['12mins', ns(720n)],
+ ['7minutes', ns(420n)],
+ ['2h', ns(7200n)],
+ ['7hrs', ns(25_200n)],
+ ['24hours', ns(86_400n)],
+ ['2days', ns(172_800n)],
+ ['365d', ns(31_536_000n)],
+ ['7weeks', ns(4_233_600n)],
+ ['104wks', ns(62_899_200n)],
+ ['52w', ns(31_449_600n)],
+ ['3months', ns(3n * 2_630_016n)],
+ ['15yrs', ns(15n * 31_557_600n)],
+ ['10yr', ns(10n * 31_557_600n)],
+ ['17y', ns(536_479_200n)]
+ ];
+ for (const [input, expected] of cases)
+ assert.equal(parseNs(input), expected, input);
+ });
+
+ it('combines tokens with and without whitespace', () => {
+ assert.equal(parseNs('2h 37min'), ns(9420n));
+ assert.equal(parseNs('2h 15m'), ns(8100n));
+ assert.equal(parseNs('20 min 17 nsec '), ns(1200n, 17n));
+ assert.equal(parseNs('1.234s0.345ms0.678us0ns'), ns(1n, 234_345_678n));
+ assert.equal(
+ parseNs('1.234s 1.345ms 1.678us 1ns'),
+ ns(1n, 235_346_679n)
+ );
+ });
+
+ it('supports fractional values with exact division only', () => {
+ assert.equal(parseNs('4.2s'), 4_200_000_000n);
+ assert.equal(parseNs('1.5minute'), ns(90n));
+ assert.equal(parseNs('0.5h'), ns(1800n));
+ assert.equal(parseNs('1.123456789s'), ns(1n, 123_456_789n));
+ assert.equal(parseNs('31.000001ms'), ns(0n, 31_000_001n));
+ // Precision losses are rejected rather than truncated.
+ for (const input of [
+ '0.000123456789s',
+ '31.0000001ms',
+ '1.0000000002s',
+ '0.0000000002s'
+ ])
+ assert.throws(() => parseHumantimeDuration(input), (error: unknown) =>
+ error instanceof DurationParseError &&
+ error.kind === 'number-overflow'
+ );
+ });
+
+ it('rejects malformed fractional input', () => {
+ for (const input of ['1.s', '1..s'])
+ assert.throws(() => parseHumantimeDuration(input), (error: unknown) =>
+ error instanceof DurationParseError &&
+ error.kind === 'invalid-character'
+ );
+ for (const input of ['.1s', '.'])
+ assert.throws(() => parseHumantimeDuration(input), (error: unknown) =>
+ error instanceof DurationParseError &&
+ error.kind === 'number-expected'
+ );
+ });
+
+ it('reports overflow like u64 arithmetic', () => {
+ for (const input of [
+ '100000000000000000000ns',
+ '100000000000000ms',
+ '10000000000000000000m',
+ '100000000000000000d',
+ '10000000000000y'
+ ])
+ assert.throws(() => parseHumantimeDuration(input), (error: unknown) =>
+ error instanceof DurationParseError &&
+ error.kind === 'number-overflow'
+ );
+ });
+
+ it('produces the Rust error messages verbatim', () => {
+ const messageOf = (input: string): string => {
+ try {
+ parseHumantimeDuration(input);
+ } catch (error) {
+ return (error as Error).message;
+ }
+ return '';
+ };
+ assert.equal(
+ messageOf('123'),
+ 'time unit needed, for example 123sec or 123ms'
+ );
+ assert.equal(
+ messageOf('10 months 1'),
+ 'time unit needed, for example 1sec or 1ms'
+ );
+ assert.equal(
+ messageOf('10nights'),
+ 'unknown time unit "nights", supported units: ns, us/µs, ms, sec, ' +
+ 'min, hours, days, weeks, months, years (and few variations)'
+ );
+ assert.equal(messageOf('\0'), 'expected number at 0');
+ assert.equal(messageOf('\r'), 'value was empty');
+ assert.equal(messageOf('1~'), 'invalid character at 1');
+ assert.equal(messageOf('1Nå'), 'invalid character at 2');
+ assert.equal(
+ messageOf('222nsec221nanosmsec7s5msec572s'),
+ 'unknown time unit "nanosmsec", supported units: ns, us/µs, ms, ' +
+ 'sec, min, hours, days, weeks, months, years (and few variations)'
+ );
+ });
+});
+
+describe('parseIggyDurationNanoseconds', () => {
+ it('maps the zero spellings to zero case-insensitively', () => {
+ for (const value of [
+ '0', 'unlimited', 'disabled', 'none', 'UNLIMITED', 'None', 'Disabled'
+ ]) {
+ assert.equal(parseIggyDurationNanoseconds(value), 0n);
+ // The literal "0" is also short-circuited by humantime itself.
+ assert.equal(nanosecondsToMilliseconds(0n), 0);
+ }
+ });
+
+ it('rounds sub-millisecond results half up', () => {
+
assert.equal(nanosecondsToMilliseconds(parseIggyDurationNanoseconds('500ms')),
500);
+
assert.equal(nanosecondsToMilliseconds(parseIggyDurationNanoseconds('17ns')),
0);
+
assert.equal(nanosecondsToMilliseconds(parseIggyDurationNanoseconds('1.5ms')),
2);
+
assert.equal(nanosecondsToMilliseconds(parseIggyDurationNanoseconds('1.005s')),
1005);
+ });
+});
diff --git a/foreign/node/src/duration.utils.ts
b/foreign/node/src/duration.utils.ts
new file mode 100644
index 000000000..35d473919
--- /dev/null
+++ b/foreign/node/src/duration.utils.ts
@@ -0,0 +1,393 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/**
+ * TypeScript port of the duration grammar accepted by the Rust SDK.
+ *
+ * `parseHumantimeDuration` mirrors humantime 2.4.0's `parse_duration`,
+ * which `IggyDuration::from_str` delegates to after lowercasing and
+ * mapping its zero spellings. Control flow, the unit table, u64 overflow
+ * checks, exact fraction division, and error messages are kept identical
+ * so both SDKs accept and reject the same strings.
+ */
+
+const MAX_U64 = 18_446_744_073_709_551_615n;
+
+export type DurationErrorKind =
+ | 'invalid-character'
+ | 'number-expected'
+ | 'unknown-unit'
+ | 'number-overflow'
+ | 'empty';
+
+/** Carries the message the corresponding Rust error's Display produces. */
+export class DurationParseError extends TypeError {
+ kind: DurationErrorKind;
+
+ constructor(kind: DurationErrorKind, message: string) {
+ super(message);
+ this.kind = kind;
+ }
+}
+
+const overflowError = () =>
+ new DurationParseError(
+ 'number-overflow',
+ 'number is too large or cannot be represented without a lack of ' +
+ 'precision (values below 1ns are not supported)'
+ );
+
+const checkedMul = (a: bigint, b: bigint): bigint => {
+ const product = a * b;
+ if (product > MAX_U64)
+ throw overflowError();
+ return product;
+};
+
+const checkedAdd = (a: bigint, b: bigint): bigint => {
+ const sum = a + b;
+ if (sum > MAX_U64)
+ throw overflowError();
+ return sum;
+};
+
+// humantime's OverflowOp::div: a non-exact division loses precision,
+// reported as overflow rather than truncated.
+const checkedExactDiv = (a: bigint, b: bigint): bigint => {
+ if (a % b !== 0n)
+ throw overflowError();
+ return a / b;
+};
+
+type UnitKind =
+ | 'ns' | 'us' | 'ms' | 's' | 'm' | 'h' | 'd' | 'w' | 'M' | 'y';
+
+const UNITS: Record<string, UnitKind> = {
+ nanos: 'ns', nsec: 'ns', ns: 'ns',
+ usec: 'us', us: 'us', 'µs': 'us',
+ millis: 'ms', msec: 'ms', ms: 'ms',
+ seconds: 's', second: 's', secs: 's', sec: 's', s: 's',
+ minutes: 'm', minute: 'm', min: 'm', mins: 'm', m: 'm',
+ hours: 'h', hour: 'h', hr: 'h', hrs: 'h', h: 'h',
+ days: 'd', day: 'd', d: 'd',
+ weeks: 'w', week: 'w', wk: 'w', wks: 'w', w: 'w',
+ months: 'M', month: 'M',
+ years: 'y', year: 'y', yr: 'y', yrs: 'y', y: 'y'
+};
+
+// Month is 30.44 days and year is 365.25 days, matching the Rust table.
+const SECONDS_PER_UNIT: Record<UnitKind, bigint> = {
+ ns: 0n, us: 0n, ms: 0n, s: 1n, m: 60n,
+ h: 3600n, d: 86_400n, w: 604_800n, M: 2_630_016n, y: 31_557_600n
+};
+
+const NANOS_PER_UNIT: Record<UnitKind, bigint> = {
+ ns: 1n, us: 1000n, ms: 1_000_000n, s: 0n, m: 0n,
+ h: 0n, d: 0n, w: 0n, M: 0n, y: 0n
+};
+
+type Fraction = {
+ numerator: bigint,
+ denominator: bigint
+};
+
+type ParsedDuration = {
+ seconds: bigint,
+ nanoseconds: bigint
+};
+
+const utf8ByteLength = (char: string): number => {
+ const code = char.codePointAt(0)!;
+ return code <= 0x7f ? 1 : code <= 0x7ff ? 2 : code <= 0xffff ? 3 : 4;
+};
+
+const isDigit = (c: string): boolean => c >= '0' && c <= '9';
+const isUnitChar = (c: string): boolean => /^[a-zA-Zµ]$/.test(c);
+
+class DurationParser {
+ private readonly chars: string[];
+ private readonly byteLengths: number[];
+ /** Index of the next unconsumed char. */
+ private index = 0;
+ /** Byte offset of the next unconsumed char; error offsets are byte-based. */
+ private consumedBytes = 0;
+
+ constructor(src: string) {
+ this.chars = Array.from(src);
+ this.byteLengths = this.chars.map(utf8ByteLength);
+ }
+
+ private off(): number {
+ return this.consumedBytes;
+ }
+
+ private next(): string | undefined {
+ const c = this.chars[this.index];
+ if (c === undefined)
+ return undefined;
+ this.index += 1;
+ this.consumedBytes += this.byteLengths[this.index - 1];
+ return c;
+ }
+
+ /** Byte-exact source slice; unit spellings are ASCII-only. */
+ private sliceBytes(startByte: number, endByte: number): string {
+ let collected = '';
+ let position = 0;
+ for (let i = 0; i < this.chars.length && position < endByte; i += 1) {
+ if (position >= startByte)
+ collected += this.chars[i];
+ position += this.byteLengths[i];
+ }
+ return collected;
+ }
+
+ parse(): ParsedDuration {
+ // The error offset is fixed where scanning began, matching the Rust
+ // implementation's single capture outside the loop.
+ let n = this.parseFirstChar(this.off());
+ // Rust: `.ok_or(Error::Empty)` on the first-char scan.
+ if (n === undefined)
+ throw new DurationParseError('empty', 'value was empty');
+ let seconds = 0n;
+ let nanoseconds = 0n;
+
+ outer:
+ while (true) {
+ let fraction: Fraction | undefined;
+ // Offsets refresh at iteration end only, so a break leaves the
+ // offset pointing at the char that caused it.
+ let off = this.off();
+ while (true) {
+ const c = this.next();
+ if (c === undefined)
+ break;
+ if (isDigit(c)) {
+ n = checkedAdd(checkedMul(n, 10n), BigInt(c));
+ } else if (isUnitChar(c)) {
+ break;
+ } else if (c === '.') {
+ // The scanner's final offset becomes the unit start.
+ const scanned = this.parseFractionalPart(off);
+ fraction = scanned.fraction;
+ off = scanned.offset;
+ break;
+ } else if (!/\s/.test(c)) {
+ throw new DurationParseError(
+ 'invalid-character',
+ `invalid character at ${off}`
+ );
+ }
+ off = this.off();
+ }
+ const start = off;
+ let unitEnd = this.off();
+ while (true) {
+ const c = this.next();
+ if (c === undefined)
+ break;
+ if (isDigit(c)) {
+ ({ seconds, nanoseconds } = this.addUnit(
+ n, fraction, start, unitEnd, seconds, nanoseconds
+ ));
+ n = BigInt(c);
+ continue outer;
+ }
+ if (/\s/.test(c))
+ break;
+ if (!isUnitChar(c))
+ throw new DurationParseError(
+ 'invalid-character',
+ `invalid character at ${unitEnd}`
+ );
+ unitEnd = this.off();
+ }
+ ({ seconds, nanoseconds } = this.addUnit(
+ n, fraction, start, unitEnd, seconds, nanoseconds
+ ));
+ const next = this.parseFirstChar(this.off());
+ if (next === undefined)
+ return { seconds, nanoseconds };
+ n = next;
+ }
+ }
+
+ private parseFirstChar(scanStart: number): bigint | undefined {
+ for (;;) {
+ const c = this.next();
+ if (c === undefined)
+ return undefined;
+ if (isDigit(c))
+ return BigInt(c);
+ if (/\s/.test(c))
+ continue;
+ throw new DurationParseError(
+ 'number-expected',
+ `expected number at ${scanStart}`
+ );
+ }
+ }
+
+ /**
+ * Consumes fraction digits after the decimal separator. Whitespace
+ * between digits is tolerated; the returned offset is the position the
+ * scanner stopped at, which becomes the unit start.
+ */
+ private parseFractionalPart(
+ startOffset: number
+ ): { fraction: Fraction, offset: number } {
+ let numerator = 0n;
+ let denominator = 1n;
+ // Leading zeros grow the denominator only.
+ let zeros = true;
+ let off = startOffset;
+ for (;;) {
+ const c = this.next();
+ if (c === undefined) {
+ off = this.off();
+ break;
+ }
+ if (c === '0') {
+ denominator = checkedMul(denominator, 10n);
+ if (!zeros)
+ numerator = checkedMul(numerator, 10n);
+ } else if (isDigit(c)) {
+ zeros = false;
+ denominator = checkedMul(denominator, 10n);
+ numerator = checkedAdd(checkedMul(numerator, 10n), BigInt(c));
+ } else if (isUnitChar(c)) {
+ break;
+ } else if (!/\s/.test(c)) {
+ throw new DurationParseError(
+ 'invalid-character',
+ `invalid character at ${off}`
+ );
+ }
+ off = this.off();
+ }
+ if (denominator === 1n)
+ throw new DurationParseError(
+ 'invalid-character',
+ `invalid character at ${off}`
+ );
+ return { fraction: { numerator, denominator }, offset: off };
+ }
+
+ private addUnit(
+ n: bigint,
+ fraction: Fraction | undefined,
+ start: number,
+ end: number,
+ seconds: bigint,
+ nanoseconds: bigint
+ ): ParsedDuration {
+ const unitSlice = this.sliceBytes(start, end);
+ const kind = UNITS[unitSlice];
+ if (kind === undefined) {
+ const detail = unitSlice.length === 0
+ ? `time unit needed, for example ${n}sec or ${n}ms`
+ : 'unknown time unit "' + unitSlice + '", supported units: ns, ' +
+ 'us/µs, ms, sec, min, hours, days, weeks, months, years ' +
+ '(and few variations)';
+ throw new DurationParseError('unknown-unit', detail);
+ }
+
+ const intSeconds = checkedMul(n, SECONDS_PER_UNIT[kind]);
+ const intNanos = checkedMul(n, NANOS_PER_UNIT[kind]);
+ ({ seconds, nanoseconds } = addCurrent(
+ seconds, nanoseconds, intSeconds, intNanos
+ ));
+
+ if (fraction === undefined)
+ return { seconds, nanoseconds };
+
+ // Fractional part: sub-nanosecond results are a precision loss.
+ const { numerator, denominator } = fraction;
+ let fracSeconds = 0n;
+ let fracNanos = 0n;
+ switch (kind) {
+ case 'ns':
+ throw overflowError();
+ case 'us':
+ fracNanos =
+ checkedExactDiv(checkedMul(numerator, 1000n), denominator);
+ break;
+ case 'ms':
+ fracNanos = checkedExactDiv(
+ checkedMul(numerator, 1_000_000n), denominator
+ );
+ break;
+ case 's':
+ fracNanos = checkedExactDiv(
+ checkedMul(numerator, 1_000_000_000n), denominator
+ );
+ break;
+ default:
+ fracSeconds = checkedExactDiv(
+ checkedMul(numerator, SECONDS_PER_UNIT[kind]), denominator
+ );
+ }
+ return addCurrent(seconds, nanoseconds, fracSeconds, fracNanos);
+ }
+}
+
+const addCurrent = (
+ seconds: bigint,
+ nanoseconds: bigint,
+ addSeconds: bigint,
+ addNanos: bigint
+): ParsedDuration => {
+ // Strictly-greater carry preserved from the Rust implementation.
+ let totalNanos = checkedAdd(nanoseconds, addNanos);
+ let totalSeconds = addSeconds;
+ if (totalNanos > 1_000_000_000n) {
+ totalSeconds = checkedAdd(totalSeconds, totalNanos / 1_000_000_000n);
+ totalNanos %= 1_000_000_000n;
+ }
+ return {
+ seconds: checkedAdd(seconds, totalSeconds),
+ nanoseconds: totalNanos
+ };
+};
+
+/** Parses a humantime expression into seconds plus subsecond nanos. */
+export const parseHumantimeDuration = (input: string): ParsedDuration => {
+ // Rust short-circuits the literal "0" before the parser runs.
+ if (input === '0')
+ return { seconds: 0n, nanoseconds: 0n };
+ return new DurationParser(input).parse();
+};
+
+const ZERO_SPELLINGS = new Set(['0', 'unlimited', 'disabled', 'none']);
+
+/**
+ * Mirrors `IggyDuration::from_str`: case-insensitive, with the zero
+ * spellings mapped to a zero duration before humantime parsing.
+ *
+ * @returns Total nanoseconds as BigInt.
+ */
+export const parseIggyDurationNanoseconds = (value: string): bigint => {
+ const lowered = value.toLowerCase();
+ if (ZERO_SPELLINGS.has(lowered))
+ return 0n;
+ const { seconds, nanoseconds } = parseHumantimeDuration(lowered);
+ return seconds * 1_000_000_000n + nanoseconds;
+};
+
+/** Rounds nanoseconds to whole milliseconds, half up. */
+export const nanosecondsToMilliseconds = (totalNanoseconds: bigint): number =>
+ Number((totalNanoseconds + 500_000n) / 1_000_000n);
diff --git a/foreign/node/src/e2e/tcp.connection-string.e2e.ts
b/foreign/node/src/e2e/tcp.connection-string.e2e.ts
new file mode 100644
index 000000000..c42eac44a
--- /dev/null
+++ b/foreign/node/src/e2e/tcp.connection-string.e2e.ts
@@ -0,0 +1,146 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import { after, describe, it } from 'node:test';
+import assert from 'node:assert/strict';
+import { Client } from '../client/client.js';
+import type { TransportType } from '../client/client.type.js';
+import { MAX_U32 } from '../constant.js';
+import { getIggyAddress } from '../tcp.sm.utils.js';
+
+const dummyOpt = 'nodelay=true' +
+ '&reconnection_retries=1' +
+ '&reconnection_interval=1s' +
+ '&heartbeat_interval=10s' +
+ '&tls=false';
+
+/** Option-value variations exercised against a live server. */
+const optionCases: {
+ name: string,
+ query: string,
+ expect: {
+ transport?: TransportType,
+ reconnect?: Record<string, unknown>,
+ heartbeatInterval?: number,
+ options?: Record<string, unknown>
+ }
+}[] = [
+ {
+ name: 'unlimited retries at the default interval',
+ query: 'reconnection_retries=unlimited',
+ expect: {
+ reconnect: { enabled: true, interval: 1000, maxRetries: MAX_U32 }
+ }
+ },
+ {
+ name: 'bounded retries with a sub-second interval',
+ query: 'reconnection_retries=10&reconnection_interval=250ms',
+ expect: {
+ reconnect: { enabled: true, interval: 250, maxRetries: 10 }
+ }
+ },
+ {
+ name: 'compound duration interval',
+ query: 'reconnection_interval=1m30s',
+ expect: {
+ reconnect: { enabled: true, interval: 90000, maxRetries: MAX_U32 }
+ }
+ },
+ {
+ name: 'disabled heartbeats and nodelay off',
+ query: 'heartbeat_interval=0ms&nodelay=false',
+ expect: {
+ heartbeatInterval: 0,
+ options: { noDelay: false }
+ }
+ },
+ {
+ name: 'reestablish_after validated then ignored',
+ query: 'reestablish_after=7s',
+ expect: {
+ reconnect: { enabled: true, interval: 1000, maxRetries: MAX_U32 }
+ }
+ },
+ {
+ name: 'timer-ceiling heartbeat interval',
+ query: 'heartbeat_interval=2147483647ms',
+ expect: {
+ heartbeatInterval: 2147483647
+ }
+ }
+];
+
+describe('e2e -> connection string', async () => {
+ const [host, port] = getIggyAddress();
+ const client = new Client(`iggy://iggy:iggy@${host}:${port}?${dummyOpt}`);
+
+ it('e2e -> connection string::parses every option exactly once',
+ () => {
+ // A repeated key would silently keep only the last value, so each
+ // option appears once above and must all land in the config.
+ assert.equal(client._config.transport, 'TCP');
+ assert.equal(client._config.options.noDelay, true);
+ assert.deepEqual(client._config.reconnect, {
+ enabled: true,
+ interval: 1000,
+ maxRetries: 1
+ });
+ assert.equal(client._config.heartbeatInterval, 10000);
+ });
+
+ it('e2e -> connection string::ping', async () => {
+ assert.ok(await client.system.ping());
+ });
+
+ describe('option values', async () => {
+ for (const { name, query, expect } of optionCases) {
+ it(name, async () => {
+ const caseClient =
+ new Client(`iggy://iggy:iggy@${host}:${port}?${query}`);
+ try {
+ if (expect.transport !== undefined)
+ assert.equal(caseClient._config.transport, expect.transport);
+ if (expect.reconnect !== undefined)
+ assert.deepEqual(
+ caseClient._config.reconnect,
+ expect.reconnect
+ );
+ if (expect.heartbeatInterval !== undefined)
+ assert.equal(
+ caseClient._config.heartbeatInterval,
+ expect.heartbeatInterval
+ );
+ for (const [key, value] of Object.entries(expect.options ?? {}))
+ assert.deepEqual(
+ (caseClient._config.options as unknown as
+ Record<string, unknown>)[key],
+ value
+ );
+
+ // Every accepted value set must still reach a live server.
+ assert.ok(await caseClient.system.ping());
+ } finally {
+ await caseClient.destroy();
+ }
+ });
+ }
+ });
+
+ after(async () => {
+ await client.destroy();
+ });
+});
diff --git a/foreign/node/src/e2e/tls.system.e2e.ts
b/foreign/node/src/e2e/tls.system.e2e.ts
index eeb8c3d64..95caa7232 100644
--- a/foreign/node/src/e2e/tls.system.e2e.ts
+++ b/foreign/node/src/e2e/tls.system.e2e.ts
@@ -37,6 +37,7 @@ import { resolve } from 'node:path';
import { after, describe, it } from 'node:test';
import assert from 'node:assert/strict';
import { Client } from '../client/client.js';
+import type { TlsOption } from '../client/client.type.js';
import { Partitioning, Consumer, PollingStrategy } from '../wire/index.js';
import { getIggyAddress } from '../tcp.sm.utils.js';
@@ -127,6 +128,69 @@ describe('e2e -> tls', { skip: !tlsEnabled &&
'IGGY_TCP_TLS_ENABLED is not set'
await c.stream.delete({ streamId: streamName });
});
+ it('e2e -> tls::connect via connection string', async () => {
+ const [, port] = getIggyAddress();
+
+ // The tls_ca_file path is resolved by the connection string parser;
+ // the certificate itself is read when the TLS socket connects.
+ const csClient = new Client(
+ 'iggy://iggy:iggy@localhost:' +
+ `${port}?tls=true&tls_ca_file=${caCertPath}`
+ );
+ try {
+ // tls=true alone: servername defaults to the DNS host.
+ assert.equal(csClient._config.transport, 'TLS');
+ const options = csClient._config.options as TlsOption;
+ assert.equal(options.servername, undefined);
+ assert.ok(await csClient.system.ping());
+ assert.deepEqual(
+ await csClient.session.login(credentials),
+ { userId: 0 }
+ );
+ } finally {
+ await csClient.destroy();
+ }
+ });
+
+ it('e2e -> tls::connect via connection string with explicit tls_domain',
+ async () => {
+ const [, port] = getIggyAddress();
+
+ const csClient = new Client(
+ 'iggy://iggy:iggy@localhost:' +
+ `${port}?tls=true&tls_domain=localhost&tls_ca_file=${caCertPath}` +
+ '&heartbeat_interval=15s'
+ );
+ try {
+ assert.equal(csClient._config.transport, 'TLS');
+ assert.equal(
+ (csClient._config.options as TlsOption).servername,
+ 'localhost'
+ );
+ assert.equal(csClient._config.heartbeatInterval, 15000);
+ assert.ok(await csClient.system.ping());
+ } finally {
+ await csClient.destroy();
+ }
+ }
+ );
+
+ it('e2e -> tls::rejects a wrong ca path with a TypeError', () => {
+ // The client is constructed eagerly, so an unreadable CA file surfaces
+ // as a TypeError from the constructor instead of a raw ENOENT Error
+ // leaking from the socket layer.
+ assert.throws(
+ () =>
+ new Client(
+ 'iggy://iggy:iggy@localhost:8090' +
+ '?tls=true&tls_ca_file=/does/not/exist.pem'
+ ),
+ (error: unknown) =>
+ error instanceof TypeError &&
+ /cannot read tls_ca_file/.test(error.message)
+ );
+ });
+
it('e2e -> tls::logout', async () => {
assert.ok(await c.session.logout());
});
diff --git a/foreign/node/src/stream/consumer-stream.ts
b/foreign/node/src/stream/consumer-stream.ts
index f69873421..80109e4f1 100644
--- a/foreign/node/src/stream/consumer-stream.ts
+++ b/foreign/node/src/stream/consumer-stream.ts
@@ -16,7 +16,7 @@
// under the License.
import { Readable } from "node:stream";
-import type { ClientConfig } from "../client/client.type.js";
+import type { ClientConfigOrString } from "../client/client.type.js";
import type { Id } from '../wire/identifier.utils.js';
import { getClient } from "../client/client.js";
import {
@@ -78,7 +78,7 @@ export type SingleConsumerStreamRequest =
ConsumerStreamRequest & {
export type GroupConsumerStreamRequest = ConsumerStreamRequest & { groupName:
string };
-export const singleConsumerStream = (config: ClientConfig) => async (
+export const singleConsumerStream = (config: ClientConfigOrString) => async (
{
streamId,
topicId,
@@ -110,7 +110,7 @@ export const singleConsumerStream = (config: ClientConfig)
=> async (
};
-export const groupConsumerStream = (config: ClientConfig) =>
+export const groupConsumerStream = (config: ClientConfigOrString) =>
async function groupConsumerStream({
groupName,
streamId,
