From 8c104236fa74cf3205ff2d56cda0c74d6ab819a9 Mon Sep 17 00:00:00 2001 From: T1B0 Date: Tue, 18 Aug 2026 18:07:11 +0200 Subject: [PATCH 1/2] feat(node): support connection string syntax for nodejs client creation --- foreign/node/README.md | 39 +++ foreign/node/package-lock.json | 7 + foreign/node/package.json | 1 + foreign/node/src/client/client.config.test.ts | 48 +++ foreign/node/src/client/client.config.ts | 30 +- .../client/client.connection-string.test.ts | 310 ++++++++++++++++++ .../src/client/client.connection-string.ts | 238 ++++++++++++++ .../node/src/client/client.connection.test.ts | 97 +++++- foreign/node/src/client/client.connection.ts | 31 +- foreign/node/src/client/client.socket.ts | 8 +- foreign/node/src/client/client.ts | 57 ++-- foreign/node/src/client/client.type.ts | 30 +- foreign/node/src/client/index.ts | 1 + foreign/node/src/constant.ts | 19 ++ .../node/src/e2e/tcp.connection-string.e2e.ts | 146 +++++++++ foreign/node/src/e2e/tls.system.e2e.ts | 78 +++++ foreign/node/src/stream/consumer-stream.ts | 11 +- 17 files changed, 1095 insertions(+), 56 deletions(-) create mode 100644 foreign/node/src/client/client.connection-string.test.ts create mode 100644 foreign/node/src/client/client.connection-string.ts create mode 100644 foreign/node/src/constant.ts create mode 100644 foreign/node/src/e2e/tcp.connection-string.e2e.ts diff --git a/foreign/node/README.md b/foreign/node/README.md index 169500437f..c95a258268 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 also accepts a connection string instead of a config +object: + +```ts +import { Client } from "apache-iggy"; + +const client = new Client("iggy://iggy:iggy@127.0.0.1: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. + +`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/package-lock.json b/foreign/node/package-lock.json index 72ab426d58..2b7b34cc2f 100644 --- a/foreign/node/package-lock.json +++ b/foreign/node/package-lock.json @@ -12,6 +12,7 @@ "@node-rs/xxhash": "1.7.7", "debug": "4.4.3", "generic-pool": "3.9.0", + "parse-duration": "^2.1.8", "uuidv7": "1.2.1" }, "devDependencies": { @@ -3658,6 +3659,12 @@ "node": ">=6" } }, + "node_modules/parse-duration": { + "version": "2.1.8", + "resolved": "https://registry.npmjs.org/parse-duration/-/parse-duration-2.1.8.tgz", + "integrity": "sha512-hM72vQ2w/HebbXx2pyUaR+EBjbkPx3Xi2aWAF9SNvJnRbP0p46KLcI8WP/z5ZruCUsJr0L2HWexVtB3PlBf/LA==", + "license": "MIT" + }, "node_modules/parse-json": { "version": "5.2.0", "resolved": "https://registry.npmjs.org/parse-json/-/parse-json-5.2.0.tgz", diff --git a/foreign/node/package.json b/foreign/node/package.json index 755019ae08..3f1c0997ce 100644 --- a/foreign/node/package.json +++ b/foreign/node/package.json @@ -53,6 +53,7 @@ "@node-rs/xxhash": "1.7.7", "debug": "4.4.3", "generic-pool": "3.9.0", + "parse-duration": "^2.1.8", "uuidv7": "1.2.1" }, "devDependencies": { diff --git a/foreign/node/src/client/client.config.test.ts b/foreign/node/src/client/client.config.test.ts index 9022ab0804..a96205d451 100644 --- a/foreign/node/src/client/client.config.test.ts +++ b/foreign/node/src/client/client.config.test.ts @@ -61,6 +61,33 @@ describe('normalizeClientConfig', () => { ); }); + it('accepts duration expressions for the heartbeat interval', () => { + assert.equal( + normalizeClientConfig({ ...config(), heartbeatInterval: '10s' }) + .heartbeatInterval, + 10000 + ); + assert.equal( + normalizeClientConfig({ ...config(), heartbeatInterval: '1h 30m' }) + .heartbeatInterval, + 5400000 + ); + for (const disabled of ['0', 'unlimited', 'disabled', 'none']) + assert.equal( + normalizeClientConfig({ ...config(), heartbeatInterval: disabled }) + .heartbeatInterval, + 0 + ); + assert.throws( + () => normalizeClientConfig({ ...config(), heartbeatInterval: '1000h' }), + /between 0 and/ + ); + assert.throws( + () => normalizeClientConfig({ ...config(), heartbeatInterval: 'garbage' }), + /invalid duration/ + ); + }); + it('rejects unusable heartbeat intervals', () => { for (const heartbeatInterval of [ -1, -5000, Number.NaN, 1.5, 2_147_483_648, 2 ** 32, Number.MAX_VALUE @@ -74,6 +101,27 @@ 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('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 7c94bdbe40..d91743faad 100644 --- a/foreign/node/src/client/client.config.ts +++ b/foreign/node/src/client/client.config.ts @@ -15,7 +15,11 @@ // specific language governing permissions and limitations // under the License. -import type { ClientConfig } from './client.type.js'; +import type { ClientConfig, ClientConfigOrString } from './client.type.js'; +import { + parseConnectionString, + parseDuration +} from './client.connection-string.js'; export const DEFAULT_MAX_RESPONSE_FRAME_SIZE = 64 * 1024 * 1024; @@ -29,8 +33,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) || @@ -48,8 +55,12 @@ export const normalizeClientConfig = ( // the default below is nullish-only, so a value that is negative, fractional // or above MAX_HEARTBEAT_INTERVAL reaches setInterval, which clamps it to // 1 ms and floods the server. - const heartbeatInterval = + // A duration expression is accepted like the Rust SDK's IggyDuration. + const rawHeartbeatInterval = config.heartbeatInterval ?? DEFAULT_HEARTBEAT_INTERVAL; + const heartbeatInterval = typeof rawHeartbeatInterval === 'string' + ? parseDuration(rawHeartbeatInterval) + : rawHeartbeatInterval; if (!Number.isSafeInteger(heartbeatInterval) || heartbeatInterval < 0 || heartbeatInterval > MAX_HEARTBEAT_INTERVAL) @@ -57,6 +68,19 @@ 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. + if (config.reconnect !== undefined) { + const { interval } = config.reconnect; + if (!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 0000000000..09198e7813 --- /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:secret@127.0.0.1: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 0000000000..ce7d359fad --- /dev/null +++ b/foreign/node/src/client/client.connection-string.ts @@ -0,0 +1,238 @@ +// 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 type { ClientConfig, ReconnectOption } from './client.type.js'; +import parseDurationLib from 'parse-duration'; + +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 such as "500ms", "1h 30m" or "2w" into milliseconds. + * + * Mirrors `IggyDuration::from_str`: matching is case-insensitive, the zero + * spellings map to 0, and unit-less or negative values are rejected like + * humantime does. + */ +export const parseDuration = (value: string): number => { + // humantime's long unit spellings that the library lacks. + const expression = value.toLowerCase() + .replaceAll('nsec', 'ns') + .replaceAll('usec', 'us') + .replaceAll('msec', 'ms'); + if (expression === '0' || expression === 'unlimited' || + expression === 'disabled' || expression === 'none') + return 0; + // The library alone would accept a bare number as milliseconds. + if (/^-?\d*\.?\d+$/.test(expression)) + throw new TypeError(`invalid duration "${value}"`); + const parsed = parseDurationLib(expression); + if (parsed === null || parsed < 0) + throw new TypeError(`invalid duration "${value}"`); + // The library accumulates floats: "1.005s" yields 1004.9999999999999, + // which downstream integer validation would reject. + return Math.round(parsed); +}; + +/** + * 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 ed1f6428fa..7a58e0c7bd 100644 --- a/foreign/node/src/client/client.connection.test.ts +++ b/foreign/node/src/client/client.connection.test.ts @@ -17,12 +17,17 @@ 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 { + createServer as createTlsServer, + type TLSSocket +} from 'node:tls'; import { describe, it } from 'node:test'; import { ProtocolFrameError } from './client.frame.js'; import { IggyConnection } from './client.connection.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 => { + 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 => { const server = createServer(); server.listen(0, '127.0.0.1'); @@ -454,4 +479,74 @@ 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); + } + }); }); diff --git a/foreign/node/src/client/client.connection.ts b/foreign/node/src/client/client.connection.ts index 767da571d8..a83d7663fc 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}"`); + } }; /** @@ -84,8 +103,10 @@ const DefaultReconnectOption: ReconnectOption = { * @returns Promise resolving after the delay */ function waitForReconnect(timer = 1000): Promise { - return new Promise((resolve) => { - setTimeout(resolve, timer); + return new Promise((resolve) => { + const timeout = setTimeout(resolve, timer); + (timeout as NodeJS.Timeout).unref(); + resolve(); }); } diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts index 2f02764c6c..3f101e91a7 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 @@ -121,7 +122,7 @@ export class CommandResponseStream extends EventEmitter { * * @param options - Client configuration */ - constructor(options: ClientConfig) { + constructor(options: ClientConfigOrString) { super(); const normalizedConfig = normalizeClientConfig(options); this.options = normalizedConfig; @@ -143,7 +144,8 @@ export class CommandResponseStream extends EventEmitter { * Initializes the stream by setting up heartbeat and connection event handlers. */ _init() { - this.heartbeat(this.options.heartbeatInterval); + // Normalization has already resolved duration expressions to milliseconds. + this.heartbeat(this.options.heartbeatInterval as number | undefined); this.connection.on('error', (error: Error) => { this._failQueue(error); }); @@ -656,7 +658,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 fc642f5ef8..25f877b131 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 40c29bc2cc..b3ac10e8fe 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 & { caFile?: string }; /** * Response from a command sent to the Iggy server. @@ -145,6 +147,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. */ @@ -160,12 +168,14 @@ export type ClientConfig = { /** 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, either in milliseconds or as a + * duration expression such as "5s" or "1h 30m" (same grammar as the Rust + * SDK's IggyDuration). Normalizes to an integer between 0 and Node's timer + * ceiling. Defaults to 5000. Set to 0 (or "unlimited"/"disabled"/"none") + * to disable client heartbeats; any other unusable value is rejected + * rather than silently disabling them. */ - heartbeatInterval?: number, + heartbeatInterval?: number | string, /** 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 d968e5ca0b..e3430c4a9b 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/constant.ts b/foreign/node/src/constant.ts new file mode 100644 index 0000000000..bfc196335b --- /dev/null +++ b/foreign/node/src/constant.ts @@ -0,0 +1,19 @@ +// 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. + +/** Largest value representable by the Rust u32 type. */ +export const MAX_U32 = 4294967295; 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 0000000000..c42eac44a6 --- /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, + heartbeatInterval?: number, + options?: Record + } +}[] = [ + { + 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)[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 eeb8c3d640..8b19164e75 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,83 @@ 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::connection string without tls stays plain TCP', async () => { + const [, port] = getIggyAddress(); + + const plainClient = new Client( + `iggy://iggy:iggy@localhost:${port}?tls=false` + ); + try { + assert.equal(plainClient._config.transport, 'TCP'); + assert.ok(await plainClient.system.ping()); + } finally { + await plainClient.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 f69873421b..2f330cad65 100644 --- a/foreign/node/src/stream/consumer-stream.ts +++ b/foreign/node/src/stream/consumer-stream.ts @@ -16,9 +16,10 @@ // 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 { normalizeClientConfig } from "../client/client.config.js"; import { NO_ASSIGNED_PARTITION, type PollMessages @@ -78,7 +79,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, @@ -90,7 +91,7 @@ export const singleConsumerStream = (config: ClientConfig) => async ( endOnLastOffset = false }: SingleConsumerStreamRequest ): Promise => { - const c = await getClient(config); + const c = await getClient(normalizeClientConfig(config)); const poll = { streamId, @@ -110,7 +111,7 @@ export const singleConsumerStream = (config: ClientConfig) => async ( }; -export const groupConsumerStream = (config: ClientConfig) => +export const groupConsumerStream = (config: ClientConfigOrString) => async function groupConsumerStream({ groupName, streamId, @@ -121,7 +122,7 @@ export const groupConsumerStream = (config: ClientConfig) => autocommit = true, endOnLastOffset = false }: GroupConsumerStreamRequest): Promise { - const s = await getClient(config); + const s = await getClient(normalizeClientConfig(config)); await s.group.ensureAndJoin(streamId, topicId, groupName); const poll = { From a09ab2e3c98f7b16a86fba31d4e8f615a01dd85f Mon Sep 17 00:00:00 2001 From: T1B0 Date: Tue, 18 Aug 2026 18:07:11 +0200 Subject: [PATCH 2/2] feat(node): support connection string syntax for nodejs client creation --- foreign/node/src/e2e/tls.system.e2e.ts | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/foreign/node/src/e2e/tls.system.e2e.ts b/foreign/node/src/e2e/tls.system.e2e.ts index 8b19164e75..95caa7232d 100644 --- a/foreign/node/src/e2e/tls.system.e2e.ts +++ b/foreign/node/src/e2e/tls.system.e2e.ts @@ -175,20 +175,6 @@ describe('e2e -> tls', { skip: !tlsEnabled && 'IGGY_TCP_TLS_ENABLED is not set' } ); - it('e2e -> tls::connection string without tls stays plain TCP', async () => { - const [, port] = getIggyAddress(); - - const plainClient = new Client( - `iggy://iggy:iggy@localhost:${port}?tls=false` - ); - try { - assert.equal(plainClient._config.transport, 'TCP'); - assert.ok(await plainClient.system.ping()); - } finally { - await plainClient.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