From 98215c2e2836a04faf4c3f75ebf381ec090dfe69 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 18 Sep 2026 16:20:57 -0400 Subject: [PATCH 1/6] feat: Migrate the node-family SDKs to @launchdarkly/eventsource with injected fetch adapters --- .../__tests__/platform/ElectronFetch.test.ts | 108 ++++++++++ packages/sdk/electron/package.json | 5 +- .../electron/src/platform/ElectronFetch.ts | 76 +++++++ .../electron/src/platform/ElectronRequests.ts | 14 +- .../__tests__/platform/NodeFetch.test.ts | 153 ++++++++++++++ packages/sdk/node-client/package.json | 2 +- .../sdk/node-client/src/platform/NodeFetch.ts | 114 +++++++++++ .../node-client/src/platform/NodeRequests.ts | 13 +- .../__tests__/platform/NodeFetch.test.ts | 188 ++++++++++++++++++ packages/sdk/server-node/package.json | 2 +- .../sdk/server-node/src/platform/NodeFetch.ts | 116 +++++++++++ .../server-node/src/platform/NodeRequests.ts | 16 +- 12 files changed, 785 insertions(+), 22 deletions(-) create mode 100644 packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts create mode 100644 packages/sdk/electron/src/platform/ElectronFetch.ts create mode 100644 packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts create mode 100644 packages/sdk/node-client/src/platform/NodeFetch.ts create mode 100644 packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts create mode 100644 packages/sdk/server-node/src/platform/NodeFetch.ts diff --git a/packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts b/packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts new file mode 100644 index 0000000000..8a138e4d66 --- /dev/null +++ b/packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts @@ -0,0 +1,108 @@ +import { + AsyncQueue, + TestHttpHandlers, + TestHttpServer, + TestHttpServers, +} from 'launchdarkly-js-test-helpers'; + +import createElectronFetch from '../../src/platform/ElectronFetch'; +import ElectronRequests from '../../src/platform/ElectronRequests'; + +describe('given a running HTTP server', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.start(); + }); + + afterEach(async () => { + await server.closeAndWait(); + }); + + it('forwards the request and exposes the status, headers, and body chunks', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const electronFetch = createElectronFetch(); + const res = await electronFetch(`${server.url}/stream`, { + method: 'REPORT', + headers: { authorization: 'sdk-key' }, + body: '{"kind":"user"}', + }); + + expect(res.status).toEqual(200); + const headers: Record = {}; + res.headers.forEach((value, key) => { + headers[key] = value; + }); + expect(headers['content-type']).toEqual('text/event-stream'); + + const reader = res.body?.getReader(); + const first = await reader?.read(); + expect(first?.done).toBe(false); + expect(Buffer.from(first?.value ?? []).toString()).toEqual('first'); + + const received = await server.nextRequest(); + expect(received.method.toUpperCase()).toEqual('REPORT'); + expect(received.headers.authorization).toEqual('sdk-key'); + expect(received.body).toEqual('{"kind":"user"}'); + }); + + it('does not follow redirects', async () => { + server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); + + const electronFetch = createElectronFetch(); + const res = await electronFetch(server.url, { method: 'GET', headers: {} }); + + expect(res.status).toEqual(301); + expect(server.requestCount()).toEqual(1); + }); + + it('streams SSE events through createEventSource', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: hello\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new ElectronRequests(); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('hello'); + } finally { + es.close(); + } + }); +}); + +describe('given a running HTTPS server with a self-signed certificate', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.startSecure(); + server.byDefault(TestHttpHandlers.respond(200)); + }); + + afterEach(async () => { + await server.closeAndWait(); + }); + + it('rejects the connection when the certificate is not trusted', async () => { + // The Electron SDK exposes no TLS options. Verification follows the platform default, so a + // self-signed certificate the machine does not trust must fail the request. + const electronFetch = createElectronFetch(); + await expect(electronFetch(server.url, { method: 'GET', headers: {} })).rejects.toThrow(); + }); +}); diff --git a/packages/sdk/electron/package.json b/packages/sdk/electron/package.json index ae38ae5d5e..8556b45738 100644 --- a/packages/sdk/electron/package.json +++ b/packages/sdk/electron/package.json @@ -53,13 +53,14 @@ "electron": ">=34.5.6" }, "dependencies": { - "@launchdarkly/js-client-sdk-common": "workspace:^", - "launchdarkly-eventsource": "2.2.0" + "@launchdarkly/eventsource": "workspace:^", + "@launchdarkly/js-client-sdk-common": "workspace:^" }, "devDependencies": { "@types/jest": "^29.4.0", "electron": "^42.4.0", "jest": "^29.5.0", + "launchdarkly-js-test-helpers": "^2.2.0", "oxfmt": "0.63.0", "oxlint": "1.78.0", "ts-jest": "^29.0.5", diff --git a/packages/sdk/electron/src/platform/ElectronFetch.ts b/packages/sdk/electron/src/platform/ElectronFetch.ts new file mode 100644 index 0000000000..1a30115dff --- /dev/null +++ b/packages/sdk/electron/src/platform/ElectronFetch.ts @@ -0,0 +1,76 @@ +import * as http from 'http'; +import * as https from 'https'; + +import type { FetchFn, FetchRequestOptions, FetchResponse } from '@launchdarkly/eventsource'; + +function wrapResponse(res: http.IncomingMessage): FetchResponse { + // The async iterator hands out one chunk per read. It resolves done when the server ends the + // stream, and it rejects when the connection drops or the request is destroyed. + const iterator = res[Symbol.asyncIterator](); + return { + status: res.statusCode ?? 0, + statusText: res.statusMessage ?? '', + headers: { + forEach(callback: (value: string, key: string) => void): void { + Object.entries(res.headers).forEach(([key, value]) => { + if (value === undefined) { + return; + } + callback(Array.isArray(value) ? value.join(', ') : value, key); + }); + }, + }, + body: { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value as Uint8Array }; + }, + }), + }, + }; +} + +/** + * Creates a `fetch()`-shaped function over `node:http`/`node:https` for the SSE stream. + * + * The Electron SDK exposes no agent, proxy, or TLS options: the running machine's own network + * configuration applies, and TLS verification follows the platform default. The returned function + * never follows a redirect: a redirect status resolves like any other non-200 response, and the + * caller decides whether to retry the original URL. It applies no read or socket timeout; the + * caller owns the read timeout and cancels through the request's `AbortSignal`. + */ +export default function createElectronFetch(): FetchFn { + return async (url: string, init: FetchRequestOptions): Promise => { + const isSecure = url.startsWith('https://'); + const impl = isSecure ? https : http; + const requestOptions: https.RequestOptions = { + method: init.method, + headers: init.headers, + }; + return new Promise((resolve, reject) => { + const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); + // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. + req.setNoDelay(true); + const { signal } = init; + if (signal) { + const abort = () => req.destroy(new Error('The stream request was aborted')); + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + // stays attached after resolve, so a later socket error becomes a harmless no-op reject + // instead of an unhandled 'error' event that would crash the process + req.on('error', reject); + if (init.body !== undefined) { + req.write(init.body); + } + req.end(); + }); + }; +} diff --git a/packages/sdk/electron/src/platform/ElectronRequests.ts b/packages/sdk/electron/src/platform/ElectronRequests.ts index 18a3fd7712..dc628d7898 100644 --- a/packages/sdk/electron/src/platform/ElectronRequests.ts +++ b/packages/sdk/electron/src/platform/ElectronRequests.ts @@ -1,21 +1,23 @@ import * as http from 'http'; import * as https from 'https'; -// No types for the event source. -// @ts-ignore -import { EventSource as LDEventSource } from 'launchdarkly-eventsource'; import { promisify } from 'util'; import * as zlib from 'zlib'; +import { createEventSource, FetchFn } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; +import createElectronFetch from './ElectronFetch'; import ElectronResponse from './ElectronResponse'; const gzip = promisify(zlib.gzip); export default class ElectronRequests implements platform.Requests { + private _eventSourceFetch: FetchFn; + private _enableBodyCompression: boolean = false; constructor(enableEventCompression?: boolean) { + this._eventSourceFetch = createElectronFetch(); this._enableBodyCompression = !!enableEventCompression; } @@ -71,12 +73,12 @@ export default class ElectronRequests implements platform.Requests { url: string, eventSourceInitDict: platform.EventSourceInitDict, ): platform.EventSource { - const expandedOptions = { + return createEventSource(url, { ...eventSourceInitDict, maxBackoffMillis: 30 * 1000, jitterRatio: 0.5, - }; - return new LDEventSource(url, expandedOptions); + fetch: this._eventSourceFetch, + }); } getEventSourceCapabilities(): EventSourceCapabilities { diff --git a/packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts b/packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts new file mode 100644 index 0000000000..a11cb19e24 --- /dev/null +++ b/packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts @@ -0,0 +1,153 @@ +import * as http from 'http'; +import { + AsyncQueue, + TestHttpHandlers, + TestHttpServer, + TestHttpServers, +} from 'launchdarkly-js-test-helpers'; + +import createNodeFetch from '../../src/platform/NodeFetch'; +import NodeRequests from '../../src/platform/NodeRequests'; + +describe('given a running HTTP server', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.start(); + }); + + afterEach(async () => { + await server.closeAndWait(); + }); + + it('forwards the request and exposes the status, headers, and body chunks', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const nodeFetch = createNodeFetch(); + const res = await nodeFetch(`${server.url}/stream`, { + method: 'REPORT', + headers: { authorization: 'sdk-key' }, + body: '{"kind":"user"}', + }); + + expect(res.status).toEqual(200); + const headers: Record = {}; + res.headers.forEach((value, key) => { + headers[key] = value; + }); + expect(headers['content-type']).toEqual('text/event-stream'); + + const reader = res.body?.getReader(); + const first = await reader?.read(); + expect(first?.done).toBe(false); + expect(Buffer.from(first?.value ?? []).toString()).toEqual('first'); + + const received = await server.nextRequest(); + expect(received.method.toUpperCase()).toEqual('REPORT'); + expect(received.headers.authorization).toEqual('sdk-key'); + expect(received.body).toEqual('{"kind":"user"}'); + }); + + it('does not follow redirects', async () => { + server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); + + const nodeFetch = createNodeFetch(); + const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); + + expect(res.status).toEqual(301); + expect(server.requestCount()).toEqual(1); + }); + + it('uses the supplied agent', async () => { + server.byDefault(TestHttpHandlers.respond(200)); + const agent = new http.Agent({ keepAlive: false }); + // addRequest exists at runtime but is not part of the public http.Agent type. + // @ts-ignore + const addRequestSpy = jest.spyOn(agent, 'addRequest'); + + // The runtime accepts an http.Agent even though this SDK only ever builds an https.Agent. + // @ts-ignore + const nodeFetch = createNodeFetch(agent); + await nodeFetch(server.url, { method: 'GET', headers: {} }); + + expect(addRequestSpy).toHaveBeenCalled(); + }); + + it('streams SSE events through createEventSource', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: hello\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests(); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('hello'); + } finally { + es.close(); + } + }); +}); + +describe('given a running HTTPS server with a self-signed certificate', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.startSecure(); + server.byDefault(TestHttpHandlers.respond(200)); + }); + + afterEach(async () => { + await server.closeAndWait(); + }); + + it('connects when the CA is supplied in the TLS options', async () => { + const nodeFetch = createNodeFetch(undefined, { ca: server.certificate }); + const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); + expect(res.status).toEqual(200); + }); + + it('rejects the connection when no CA is supplied', async () => { + const nodeFetch = createNodeFetch(); + await expect(nodeFetch(server.url, { method: 'GET', headers: {} })).rejects.toThrow(); + }); + + it('streams over TLS when the CA is supplied through tlsParams', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: secure\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests({ ca: server.certificate }); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('secure'); + } finally { + es.close(); + } + }); +}); diff --git a/packages/sdk/node-client/package.json b/packages/sdk/node-client/package.json index cfa0f6aa7f..7106824c31 100644 --- a/packages/sdk/node-client/package.json +++ b/packages/sdk/node-client/package.json @@ -45,7 +45,7 @@ }, "dependencies": { "@launchdarkly/js-client-sdk-common": "1.32.5", - "launchdarkly-eventsource": "2.2.0" + "@launchdarkly/eventsource": "0.1.0" }, "devDependencies": { "@types/jest": "^29.4.0", diff --git a/packages/sdk/node-client/src/platform/NodeFetch.ts b/packages/sdk/node-client/src/platform/NodeFetch.ts new file mode 100644 index 0000000000..bc7aad5aa2 --- /dev/null +++ b/packages/sdk/node-client/src/platform/NodeFetch.ts @@ -0,0 +1,114 @@ +import * as http from 'http'; +import * as https from 'https'; + +import type { FetchFn, FetchRequestOptions, FetchResponse } from '@launchdarkly/eventsource'; + +import type { LDTLSOptions } from '../NodeOptions'; + +/** + * The TLS options that are copied onto each `https` request. The names match the options of + * `https.request()`. + */ +const TLS_OPTION_NAMES = [ + 'pfx', + 'key', + 'passphrase', + 'cert', + 'ca', + 'ciphers', + 'rejectUnauthorized', + 'secureProtocol', + 'servername', + 'checkServerIdentity', +] as const; + +function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { + const merged: Record = {}; + if (!tlsOptions) { + return merged; + } + const bag = tlsOptions as unknown as Record; + TLS_OPTION_NAMES.forEach((name) => { + if (bag[name] !== undefined) { + merged[name] = bag[name]; + } + }); + return merged; +} + +function wrapResponse(res: http.IncomingMessage): FetchResponse { + // The async iterator hands out one chunk per read. It resolves done when the server ends the + // stream, and it rejects when the connection drops or the request is destroyed. + const iterator = res[Symbol.asyncIterator](); + return { + status: res.statusCode ?? 0, + statusText: res.statusMessage ?? '', + headers: { + forEach(callback: (value: string, key: string) => void): void { + Object.entries(res.headers).forEach(([key, value]) => { + if (value === undefined) { + return; + } + callback(Array.isArray(value) ? value.join(', ') : value, key); + }); + }, + }, + body: { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value as Uint8Array }; + }, + }), + }, + }; +} + +/** + * Creates a `fetch()`-shaped function over `node:http`/`node:https` for the SSE stream. + * + * The returned function bakes in the connection configuration: the agent (which can carry proxy + * or TLS setup) and the TLS parameters for `https` URLs. It never follows a redirect: a redirect + * status resolves like any other non-200 response, and the caller decides whether to retry the + * original URL. It applies no read or socket timeout; the caller owns the read timeout and + * cancels through the request's `AbortSignal`. + */ +export default function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): FetchFn { + const tlsParams = tlsRequestOptions(tlsOptions); + return async (url: string, init: FetchRequestOptions): Promise => { + const isSecure = url.startsWith('https://'); + const impl = isSecure ? https : http; + const requestOptions: https.RequestOptions & Record = { + method: init.method, + headers: init.headers, + agent, + }; + if (isSecure) { + Object.assign(requestOptions, tlsParams); + } + return new Promise((resolve, reject) => { + const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); + // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. + req.setNoDelay(true); + const { signal } = init; + if (signal) { + const abort = () => req.destroy(new Error('The stream request was aborted')); + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + // stays attached after resolve, so a later socket error becomes a harmless no-op reject + // instead of an unhandled 'error' event that would crash the process + req.on('error', reject); + if (init.body !== undefined) { + req.write(init.body); + } + req.end(); + }); + }; +} diff --git a/packages/sdk/node-client/src/platform/NodeRequests.ts b/packages/sdk/node-client/src/platform/NodeRequests.ts index 1ed819e94d..166dc8bb80 100644 --- a/packages/sdk/node-client/src/platform/NodeRequests.ts +++ b/packages/sdk/node-client/src/platform/NodeRequests.ts @@ -1,14 +1,13 @@ import * as http from 'http'; import * as https from 'https'; -// No types for the event source. -// @ts-ignore -import { EventSource as LDEventSource } from 'launchdarkly-eventsource'; import { promisify } from 'util'; import * as zlib from 'zlib'; +import { createEventSource, FetchFn } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; import type { LDTLSOptions } from '../NodeOptions'; +import createNodeFetch from './NodeFetch'; import NodeResponse from './NodeResponse'; const gzip = promisify(zlib.gzip); @@ -45,11 +44,14 @@ export default class NodeRequests implements platform.Requests { private _tlsOptions: LDTLSOptions | undefined; + private _eventSourceFetch: FetchFn; + private _enableBodyCompression: boolean = false; constructor(tlsOptions?: LDTLSOptions, enableEventCompression?: boolean) { this._agent = tlsOptions ? new https.Agent(processTlsOptions(tlsOptions)) : undefined; this._tlsOptions = tlsOptions; + this._eventSourceFetch = createNodeFetch(this._agent, this._tlsOptions); this._enableBodyCompression = !!enableEventCompression; } @@ -106,12 +108,11 @@ export default class NodeRequests implements platform.Requests { ): platform.EventSource { const expandedOptions = { ...eventSourceInitDict, - agent: this._agent, - tlsParams: this._tlsOptions, maxBackoffMillis: 30 * 1000, jitterRatio: 0.5, + fetch: this._eventSourceFetch, }; - return new LDEventSource(url, expandedOptions); + return createEventSource(url, expandedOptions); } getEventSourceCapabilities(): EventSourceCapabilities { diff --git a/packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts b/packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts new file mode 100644 index 0000000000..aa6530618f --- /dev/null +++ b/packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts @@ -0,0 +1,188 @@ +import * as http from 'http'; +import { + AsyncQueue, + TestHttpHandlers, + TestHttpServer, + TestHttpServers, +} from 'launchdarkly-js-test-helpers'; + +import createNodeFetch from '../../src/platform/NodeFetch'; +import NodeRequests from '../../src/platform/NodeRequests'; + +describe('given a running HTTP server', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.start(); + }); + + afterEach(async () => { + await server.closeAndWait(); + }); + + it('forwards the request and exposes the status, headers, and body chunks', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const nodeFetch = createNodeFetch(); + const res = await nodeFetch(`${server.url}/stream`, { + method: 'REPORT', + headers: { authorization: 'sdk-key' }, + body: '{"kind":"user"}', + }); + + expect(res.status).toEqual(200); + const headers: Record = {}; + res.headers.forEach((value, key) => { + headers[key] = value; + }); + expect(headers['content-type']).toEqual('text/event-stream'); + + const reader = res.body?.getReader(); + const first = await reader?.read(); + expect(first?.done).toBe(false); + expect(Buffer.from(first?.value ?? []).toString()).toEqual('first'); + + const received = await server.nextRequest(); + expect(received.method.toUpperCase()).toEqual('REPORT'); + expect(received.headers.authorization).toEqual('sdk-key'); + expect(received.body).toEqual('{"kind":"user"}'); + }); + + it('does not follow redirects', async () => { + server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); + + const nodeFetch = createNodeFetch(); + const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); + + expect(res.status).toEqual(301); + expect(server.requestCount()).toEqual(1); + }); + + it('stops the stream when the signal aborts', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); + + const controller = new AbortController(); + const nodeFetch = createNodeFetch(); + const res = await nodeFetch(server.url, { + method: 'GET', + headers: {}, + signal: controller.signal, + }); + const reader = res.body?.getReader(); + await reader?.read(); + const pending = reader?.read(); + controller.abort(); + await expect(pending).rejects.toThrow(); + }); + + it('uses the supplied agent', async () => { + server.byDefault(TestHttpHandlers.respond(200)); + const agent = new http.Agent({ keepAlive: false }); + // addRequest exists at runtime but is not part of the public http.Agent type. + // @ts-ignore + const addRequestSpy = jest.spyOn(agent, 'addRequest'); + + const nodeFetch = createNodeFetch(agent); + await nodeFetch(server.url, { method: 'GET', headers: {} }); + + expect(addRequestSpy).toHaveBeenCalled(); + }); + + it('streams SSE events through createEventSource', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: hello\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests(); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('hello'); + } finally { + es.close(); + } + }); +}); + +describe('given a running HTTPS server with a self-signed certificate', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.startSecure(); + server.byDefault(TestHttpHandlers.respond(200)); + }); + + afterEach(async () => { + await server.closeAndWait(); + }); + + it('connects when the CA is supplied in the TLS options', async () => { + const nodeFetch = createNodeFetch(undefined, { ca: server.certificate }); + const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); + expect(res.status).toEqual(200); + }); + + it('rejects the connection when no CA is supplied', async () => { + const nodeFetch = createNodeFetch(); + await expect(nodeFetch(server.url, { method: 'GET', headers: {} })).rejects.toThrow(); + }); + + it('fails the stream against an untrusted certificate by default', async () => { + const requests = new NodeRequests(); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const errors = new AsyncQueue(); + es.addEventListener('error', (event) => errors.add(event)); + expect(await errors.take()).toBeDefined(); + } finally { + es.close(); + } + }); + + it('streams over TLS when the CA is supplied through tlsParams', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: secure\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests({ ca: server.certificate }); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('secure'); + } finally { + es.close(); + } + }); +}); diff --git a/packages/sdk/server-node/package.json b/packages/sdk/server-node/package.json index c34f048029..34ac8f1cd4 100644 --- a/packages/sdk/server-node/package.json +++ b/packages/sdk/server-node/package.json @@ -49,7 +49,7 @@ "dependencies": { "@launchdarkly/js-server-sdk-common": "2.24.1", "https-proxy-agent": "^7.0.6", - "launchdarkly-eventsource": "2.3.0" + "@launchdarkly/eventsource": "0.1.0" }, "devDependencies": { "@types/jest": "^29.4.0", diff --git a/packages/sdk/server-node/src/platform/NodeFetch.ts b/packages/sdk/server-node/src/platform/NodeFetch.ts new file mode 100644 index 0000000000..870d983f37 --- /dev/null +++ b/packages/sdk/server-node/src/platform/NodeFetch.ts @@ -0,0 +1,116 @@ +import * as http from 'http'; +import * as https from 'https'; + +import type { FetchFn, FetchRequestOptions, FetchResponse } from '@launchdarkly/eventsource'; +import type { LDTLSOptions } from '@launchdarkly/js-server-sdk-common'; + +/** + * The TLS options that are copied onto each `https` request. The names match the options of + * `https.request()`. + */ +const TLS_OPTION_NAMES = [ + 'pfx', + 'key', + 'passphrase', + 'cert', + 'ca', + 'ciphers', + 'rejectUnauthorized', + 'secureProtocol', + 'servername', + 'checkServerIdentity', +] as const; + +function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { + const merged: Record = {}; + if (!tlsOptions) { + return merged; + } + const bag = tlsOptions as unknown as Record; + TLS_OPTION_NAMES.forEach((name) => { + if (bag[name] !== undefined) { + merged[name] = bag[name]; + } + }); + return merged; +} + +function wrapResponse(res: http.IncomingMessage): FetchResponse { + // The async iterator hands out one chunk per read. It resolves done when the server ends the + // stream, and it rejects when the connection drops or the request is destroyed. + const iterator = res[Symbol.asyncIterator](); + return { + status: res.statusCode ?? 0, + statusText: res.statusMessage ?? '', + headers: { + forEach(callback: (value: string, key: string) => void): void { + Object.entries(res.headers).forEach(([key, value]) => { + if (value === undefined) { + return; + } + callback(Array.isArray(value) ? value.join(', ') : value, key); + }); + }, + }, + body: { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value as Uint8Array }; + }, + }), + }, + }; +} + +/** + * Creates a `fetch()`-shaped function over `node:http`/`node:https` for the SSE stream. + * + * The returned function bakes in the connection configuration: the agent (which can carry proxy + * or TLS setup) and the TLS parameters for `https` URLs. It never follows a redirect: a redirect + * status resolves like any other non-200 response, and the caller decides whether to retry the + * original URL. It applies no read or socket timeout; the caller owns the read timeout and + * cancels through the request's `AbortSignal`. + */ +export default function createNodeFetch( + agent?: https.Agent | http.Agent, + tlsOptions?: LDTLSOptions, +): FetchFn { + const tlsParams = tlsRequestOptions(tlsOptions); + return async (url: string, init: FetchRequestOptions): Promise => { + const isSecure = url.startsWith('https://'); + const impl = isSecure ? https : http; + const requestOptions: https.RequestOptions & Record = { + method: init.method, + headers: init.headers, + agent, + }; + if (isSecure) { + Object.assign(requestOptions, tlsParams); + } + return new Promise((resolve, reject) => { + const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); + // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. + req.setNoDelay(true); + const { signal } = init; + if (signal) { + const abort = () => req.destroy(new Error('The stream request was aborted')); + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + // stays attached after resolve, so a later socket error becomes a harmless no-op reject + // instead of an unhandled 'error' event that would crash the process + req.on('error', reject); + if (init.body !== undefined) { + req.write(init.body); + } + req.end(); + }); + }; +} diff --git a/packages/sdk/server-node/src/platform/NodeRequests.ts b/packages/sdk/server-node/src/platform/NodeRequests.ts index fddce38559..335a552d26 100644 --- a/packages/sdk/server-node/src/platform/NodeRequests.ts +++ b/packages/sdk/server-node/src/platform/NodeRequests.ts @@ -1,13 +1,11 @@ import * as http from 'http'; import * as https from 'https'; import { HttpsProxyAgent, HttpsProxyAgentOptions } from 'https-proxy-agent'; -// No types for the event source. -// @ts-ignore -import { EventSource as LDEventSource } from 'launchdarkly-eventsource'; import { format as formatUrl } from 'url'; import { promisify } from 'util'; import * as zlib from 'zlib'; +import { createEventSource, FetchFn } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, LDLogger, @@ -16,6 +14,7 @@ import { platform, } from '@launchdarkly/js-server-sdk-common'; +import createNodeFetch from './NodeFetch'; import NodeResponse from './NodeResponse'; const gzip = promisify(zlib.gzip); @@ -124,6 +123,8 @@ export default class NodeRequests implements platform.Requests { private _tlsOptions: LDTLSOptions | undefined; + private _eventSourceFetch: FetchFn; + private _hasProxy: boolean = false; private _hasProxyAuth: boolean = false; @@ -138,6 +139,10 @@ export default class NodeRequests implements platform.Requests { enableEventCompression?: boolean, ) { this._agent = resolveAgent(tlsOptions, proxyOptions, proxyAgent, logger); + // The agent owns connection setup when the caller supplies one, so the TLS parameters are + // only forwarded when this class built the agent itself (or no agent exists). + this._tlsOptions = proxyAgent ? undefined : tlsOptions; + this._eventSourceFetch = createNodeFetch(this._agent, this._tlsOptions); // A caller-supplied proxyAgent is treated as a best-effort proxy signal: the SDK cannot // inspect an opaque agent to know whether it actually proxies (it could just as easily be a // certificate-only agent for mTLS). Reporting true is the better default here because @@ -211,12 +216,11 @@ export default class NodeRequests implements platform.Requests { ): platform.EventSource { const expandedOptions = { ...eventSourceInitDict, - agent: this._agent, - tlsParams: this._tlsOptions, maxBackoffMillis: 30 * 1000, jitterRatio: 0.5, + fetch: this._eventSourceFetch, }; - return new LDEventSource(url, expandedOptions); + return createEventSource(url, expandedOptions); } getEventSourceCapabilities(): EventSourceCapabilities { From fbc2048777dbd6119f6b193fa3b21de25c10acff Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 23 Sep 2026 15:47:05 -0400 Subject: [PATCH 2/6] refactor: consolidate node eventsource implementation --- ...eFetch.test.ts => NodeEventSource.test.ts} | 2 +- .../{NodeFetch.ts => NodeEventSource.ts} | 35 ++++++++++++++++-- .../node-client/src/platform/NodeRequests.ts | 15 +++----- ...eFetch.test.ts => NodeEventSource.test.ts} | 2 +- .../{NodeFetch.ts => NodeEventSource.ts} | 36 +++++++++++++++++-- .../server-node/src/platform/NodeRequests.ts | 15 +++----- 6 files changed, 76 insertions(+), 29 deletions(-) rename packages/sdk/node-client/__tests__/platform/{NodeFetch.test.ts => NodeEventSource.test.ts} (98%) rename packages/sdk/node-client/src/platform/{NodeFetch.ts => NodeEventSource.ts} (77%) rename packages/sdk/server-node/__tests__/platform/{NodeFetch.test.ts => NodeEventSource.test.ts} (98%) rename packages/sdk/server-node/src/platform/{NodeFetch.ts => NodeEventSource.ts} (78%) diff --git a/packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts b/packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts similarity index 98% rename from packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts rename to packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts index a11cb19e24..00906a17f9 100644 --- a/packages/sdk/node-client/__tests__/platform/NodeFetch.test.ts +++ b/packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts @@ -6,7 +6,7 @@ import { TestHttpServers, } from 'launchdarkly-js-test-helpers'; -import createNodeFetch from '../../src/platform/NodeFetch'; +import { createNodeFetch } from '../../src/platform/NodeEventSource'; import NodeRequests from '../../src/platform/NodeRequests'; describe('given a running HTTP server', () => { diff --git a/packages/sdk/node-client/src/platform/NodeFetch.ts b/packages/sdk/node-client/src/platform/NodeEventSource.ts similarity index 77% rename from packages/sdk/node-client/src/platform/NodeFetch.ts rename to packages/sdk/node-client/src/platform/NodeEventSource.ts index bc7aad5aa2..9c32261c65 100644 --- a/packages/sdk/node-client/src/platform/NodeFetch.ts +++ b/packages/sdk/node-client/src/platform/NodeEventSource.ts @@ -1,7 +1,13 @@ import * as http from 'http'; import * as https from 'https'; -import type { FetchFn, FetchRequestOptions, FetchResponse } from '@launchdarkly/eventsource'; +import { + createEventSource, + type FetchFn, + type FetchRequestOptions, + type FetchResponse, +} from '@launchdarkly/eventsource'; +import type { platform } from '@launchdarkly/js-client-sdk-common'; import type { LDTLSOptions } from '../NodeOptions'; @@ -76,7 +82,7 @@ function wrapResponse(res: http.IncomingMessage): FetchResponse { * original URL. It applies no read or socket timeout; the caller owns the read timeout and * cancels through the request's `AbortSignal`. */ -export default function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): FetchFn { +export function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): FetchFn { const tlsParams = tlsRequestOptions(tlsOptions); return async (url: string, init: FetchRequestOptions): Promise => { const isSecure = url.startsWith('https://'); @@ -112,3 +118,28 @@ export default function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSO }); }; } + +export type NodeEventSourceFactory = ( + url: string, + eventSourceInitDict: platform.EventSourceInitDict, +) => platform.EventSource; + +/** + * Creates the event source factory used by `NodeRequests.createEventSource`. + * + * The factory bakes in the connection configuration (agent and TLS parameters, through the fetch + * adapter above) and the SDK retry policy (backoff cap and jitter). + */ +export default function createNodeEventSourceFactory( + agent?: https.Agent, + tlsOptions?: LDTLSOptions, +): NodeEventSourceFactory { + const fetch = createNodeFetch(agent, tlsOptions); + return (url, eventSourceInitDict) => + createEventSource(url, { + ...eventSourceInitDict, + maxBackoffMillis: 30 * 1000, + jitterRatio: 0.5, + fetch, + }); +} diff --git a/packages/sdk/node-client/src/platform/NodeRequests.ts b/packages/sdk/node-client/src/platform/NodeRequests.ts index 166dc8bb80..fd44b857db 100644 --- a/packages/sdk/node-client/src/platform/NodeRequests.ts +++ b/packages/sdk/node-client/src/platform/NodeRequests.ts @@ -3,11 +3,10 @@ import * as https from 'https'; import { promisify } from 'util'; import * as zlib from 'zlib'; -import { createEventSource, FetchFn } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; import type { LDTLSOptions } from '../NodeOptions'; -import createNodeFetch from './NodeFetch'; +import createNodeEventSourceFactory, { NodeEventSourceFactory } from './NodeEventSource'; import NodeResponse from './NodeResponse'; const gzip = promisify(zlib.gzip); @@ -44,14 +43,14 @@ export default class NodeRequests implements platform.Requests { private _tlsOptions: LDTLSOptions | undefined; - private _eventSourceFetch: FetchFn; + private _createEventSource: NodeEventSourceFactory; private _enableBodyCompression: boolean = false; constructor(tlsOptions?: LDTLSOptions, enableEventCompression?: boolean) { this._agent = tlsOptions ? new https.Agent(processTlsOptions(tlsOptions)) : undefined; this._tlsOptions = tlsOptions; - this._eventSourceFetch = createNodeFetch(this._agent, this._tlsOptions); + this._createEventSource = createNodeEventSourceFactory(this._agent, this._tlsOptions); this._enableBodyCompression = !!enableEventCompression; } @@ -106,13 +105,7 @@ export default class NodeRequests implements platform.Requests { url: string, eventSourceInitDict: platform.EventSourceInitDict, ): platform.EventSource { - const expandedOptions = { - ...eventSourceInitDict, - maxBackoffMillis: 30 * 1000, - jitterRatio: 0.5, - fetch: this._eventSourceFetch, - }; - return createEventSource(url, expandedOptions); + return this._createEventSource(url, eventSourceInitDict); } getEventSourceCapabilities(): EventSourceCapabilities { diff --git a/packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts b/packages/sdk/server-node/__tests__/platform/NodeEventSource.test.ts similarity index 98% rename from packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts rename to packages/sdk/server-node/__tests__/platform/NodeEventSource.test.ts index aa6530618f..b1e855677e 100644 --- a/packages/sdk/server-node/__tests__/platform/NodeFetch.test.ts +++ b/packages/sdk/server-node/__tests__/platform/NodeEventSource.test.ts @@ -6,7 +6,7 @@ import { TestHttpServers, } from 'launchdarkly-js-test-helpers'; -import createNodeFetch from '../../src/platform/NodeFetch'; +import { createNodeFetch } from '../../src/platform/NodeEventSource'; import NodeRequests from '../../src/platform/NodeRequests'; describe('given a running HTTP server', () => { diff --git a/packages/sdk/server-node/src/platform/NodeFetch.ts b/packages/sdk/server-node/src/platform/NodeEventSource.ts similarity index 78% rename from packages/sdk/server-node/src/platform/NodeFetch.ts rename to packages/sdk/server-node/src/platform/NodeEventSource.ts index 870d983f37..fc313cfdf1 100644 --- a/packages/sdk/server-node/src/platform/NodeFetch.ts +++ b/packages/sdk/server-node/src/platform/NodeEventSource.ts @@ -1,8 +1,13 @@ import * as http from 'http'; import * as https from 'https'; -import type { FetchFn, FetchRequestOptions, FetchResponse } from '@launchdarkly/eventsource'; -import type { LDTLSOptions } from '@launchdarkly/js-server-sdk-common'; +import { + createEventSource, + type FetchFn, + type FetchRequestOptions, + type FetchResponse, +} from '@launchdarkly/eventsource'; +import type { LDTLSOptions, platform } from '@launchdarkly/js-server-sdk-common'; /** * The TLS options that are copied onto each `https` request. The names match the options of @@ -75,7 +80,7 @@ function wrapResponse(res: http.IncomingMessage): FetchResponse { * original URL. It applies no read or socket timeout; the caller owns the read timeout and * cancels through the request's `AbortSignal`. */ -export default function createNodeFetch( +export function createNodeFetch( agent?: https.Agent | http.Agent, tlsOptions?: LDTLSOptions, ): FetchFn { @@ -114,3 +119,28 @@ export default function createNodeFetch( }); }; } + +export type NodeEventSourceFactory = ( + url: string, + eventSourceInitDict: platform.EventSourceInitDict, +) => platform.EventSource; + +/** + * Creates the event source factory used by `NodeRequests.createEventSource`. + * + * The factory bakes in the connection configuration (agent and TLS parameters, through the fetch + * adapter above) and the SDK retry policy (backoff cap and jitter). + */ +export default function createNodeEventSourceFactory( + agent?: https.Agent | http.Agent, + tlsOptions?: LDTLSOptions, +): NodeEventSourceFactory { + const fetch = createNodeFetch(agent, tlsOptions); + return (url, eventSourceInitDict) => + createEventSource(url, { + ...eventSourceInitDict, + maxBackoffMillis: 30 * 1000, + jitterRatio: 0.5, + fetch, + }); +} diff --git a/packages/sdk/server-node/src/platform/NodeRequests.ts b/packages/sdk/server-node/src/platform/NodeRequests.ts index 335a552d26..b210e6443a 100644 --- a/packages/sdk/server-node/src/platform/NodeRequests.ts +++ b/packages/sdk/server-node/src/platform/NodeRequests.ts @@ -5,7 +5,6 @@ import { format as formatUrl } from 'url'; import { promisify } from 'util'; import * as zlib from 'zlib'; -import { createEventSource, FetchFn } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, LDLogger, @@ -14,7 +13,7 @@ import { platform, } from '@launchdarkly/js-server-sdk-common'; -import createNodeFetch from './NodeFetch'; +import createNodeEventSourceFactory, { NodeEventSourceFactory } from './NodeEventSource'; import NodeResponse from './NodeResponse'; const gzip = promisify(zlib.gzip); @@ -123,7 +122,7 @@ export default class NodeRequests implements platform.Requests { private _tlsOptions: LDTLSOptions | undefined; - private _eventSourceFetch: FetchFn; + private _createEventSource: NodeEventSourceFactory; private _hasProxy: boolean = false; @@ -142,7 +141,7 @@ export default class NodeRequests implements platform.Requests { // The agent owns connection setup when the caller supplies one, so the TLS parameters are // only forwarded when this class built the agent itself (or no agent exists). this._tlsOptions = proxyAgent ? undefined : tlsOptions; - this._eventSourceFetch = createNodeFetch(this._agent, this._tlsOptions); + this._createEventSource = createNodeEventSourceFactory(this._agent, this._tlsOptions); // A caller-supplied proxyAgent is treated as a best-effort proxy signal: the SDK cannot // inspect an opaque agent to know whether it actually proxies (it could just as easily be a // certificate-only agent for mTLS). Reporting true is the better default here because @@ -214,13 +213,7 @@ export default class NodeRequests implements platform.Requests { url: string, eventSourceInitDict: platform.EventSourceInitDict, ): platform.EventSource { - const expandedOptions = { - ...eventSourceInitDict, - maxBackoffMillis: 30 * 1000, - jitterRatio: 0.5, - fetch: this._eventSourceFetch, - }; - return createEventSource(url, expandedOptions); + return this._createEventSource(url, eventSourceInitDict); } getEventSourceCapabilities(): EventSourceCapabilities { From c5becb0401ac7ead71b1b32c4bc8e1e46db6dc08 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 30 Sep 2026 14:05:24 -0400 Subject: [PATCH 3/6] refactor: Rename the fetch transport types to the FetchLike names --- .../sdk/electron/src/platform/ElectronFetch.ts | 10 +++++----- .../sdk/electron/src/platform/ElectronRequests.ts | 4 ++-- .../node-client/src/platform/NodeEventSource.ts | 14 +++++++------- .../server-node/src/platform/NodeEventSource.ts | 14 +++++++------- 4 files changed, 21 insertions(+), 21 deletions(-) diff --git a/packages/sdk/electron/src/platform/ElectronFetch.ts b/packages/sdk/electron/src/platform/ElectronFetch.ts index 1a30115dff..b3762ef4e9 100644 --- a/packages/sdk/electron/src/platform/ElectronFetch.ts +++ b/packages/sdk/electron/src/platform/ElectronFetch.ts @@ -1,9 +1,9 @@ import * as http from 'http'; import * as https from 'https'; -import type { FetchFn, FetchRequestOptions, FetchResponse } from '@launchdarkly/eventsource'; +import type { FetchLike, FetchLikeOptions, FetchLikeResponse } from '@launchdarkly/eventsource'; -function wrapResponse(res: http.IncomingMessage): FetchResponse { +function wrapResponse(res: http.IncomingMessage): FetchLikeResponse { // The async iterator hands out one chunk per read. It resolves done when the server ends the // stream, and it rejects when the connection drops or the request is destroyed. const iterator = res[Symbol.asyncIterator](); @@ -43,15 +43,15 @@ function wrapResponse(res: http.IncomingMessage): FetchResponse { * caller decides whether to retry the original URL. It applies no read or socket timeout; the * caller owns the read timeout and cancels through the request's `AbortSignal`. */ -export default function createElectronFetch(): FetchFn { - return async (url: string, init: FetchRequestOptions): Promise => { +export default function createElectronFetch(): FetchLike { + return async (url: string, init: FetchLikeOptions): Promise => { const isSecure = url.startsWith('https://'); const impl = isSecure ? https : http; const requestOptions: https.RequestOptions = { method: init.method, headers: init.headers, }; - return new Promise((resolve, reject) => { + return new Promise((resolve, reject) => { const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. req.setNoDelay(true); diff --git a/packages/sdk/electron/src/platform/ElectronRequests.ts b/packages/sdk/electron/src/platform/ElectronRequests.ts index dc628d7898..fe3516c578 100644 --- a/packages/sdk/electron/src/platform/ElectronRequests.ts +++ b/packages/sdk/electron/src/platform/ElectronRequests.ts @@ -3,7 +3,7 @@ import * as https from 'https'; import { promisify } from 'util'; import * as zlib from 'zlib'; -import { createEventSource, FetchFn } from '@launchdarkly/eventsource'; +import { createEventSource, FetchLike } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; import createElectronFetch from './ElectronFetch'; @@ -12,7 +12,7 @@ import ElectronResponse from './ElectronResponse'; const gzip = promisify(zlib.gzip); export default class ElectronRequests implements platform.Requests { - private _eventSourceFetch: FetchFn; + private _eventSourceFetch: FetchLike; private _enableBodyCompression: boolean = false; diff --git a/packages/sdk/node-client/src/platform/NodeEventSource.ts b/packages/sdk/node-client/src/platform/NodeEventSource.ts index 9c32261c65..003d47521f 100644 --- a/packages/sdk/node-client/src/platform/NodeEventSource.ts +++ b/packages/sdk/node-client/src/platform/NodeEventSource.ts @@ -3,9 +3,9 @@ import * as https from 'https'; import { createEventSource, - type FetchFn, - type FetchRequestOptions, - type FetchResponse, + type FetchLike, + type FetchLikeOptions, + type FetchLikeResponse, } from '@launchdarkly/eventsource'; import type { platform } from '@launchdarkly/js-client-sdk-common'; @@ -42,7 +42,7 @@ function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { return merged; } -function wrapResponse(res: http.IncomingMessage): FetchResponse { +function wrapResponse(res: http.IncomingMessage): FetchLikeResponse { // The async iterator hands out one chunk per read. It resolves done when the server ends the // stream, and it rejects when the connection drops or the request is destroyed. const iterator = res[Symbol.asyncIterator](); @@ -82,9 +82,9 @@ function wrapResponse(res: http.IncomingMessage): FetchResponse { * original URL. It applies no read or socket timeout; the caller owns the read timeout and * cancels through the request's `AbortSignal`. */ -export function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): FetchFn { +export function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): FetchLike { const tlsParams = tlsRequestOptions(tlsOptions); - return async (url: string, init: FetchRequestOptions): Promise => { + return async (url: string, init: FetchLikeOptions): Promise => { const isSecure = url.startsWith('https://'); const impl = isSecure ? https : http; const requestOptions: https.RequestOptions & Record = { @@ -95,7 +95,7 @@ export function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): if (isSecure) { Object.assign(requestOptions, tlsParams); } - return new Promise((resolve, reject) => { + return new Promise((resolve, reject) => { const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. req.setNoDelay(true); diff --git a/packages/sdk/server-node/src/platform/NodeEventSource.ts b/packages/sdk/server-node/src/platform/NodeEventSource.ts index fc313cfdf1..ed6db34849 100644 --- a/packages/sdk/server-node/src/platform/NodeEventSource.ts +++ b/packages/sdk/server-node/src/platform/NodeEventSource.ts @@ -3,9 +3,9 @@ import * as https from 'https'; import { createEventSource, - type FetchFn, - type FetchRequestOptions, - type FetchResponse, + type FetchLike, + type FetchLikeOptions, + type FetchLikeResponse, } from '@launchdarkly/eventsource'; import type { LDTLSOptions, platform } from '@launchdarkly/js-server-sdk-common'; @@ -40,7 +40,7 @@ function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { return merged; } -function wrapResponse(res: http.IncomingMessage): FetchResponse { +function wrapResponse(res: http.IncomingMessage): FetchLikeResponse { // The async iterator hands out one chunk per read. It resolves done when the server ends the // stream, and it rejects when the connection drops or the request is destroyed. const iterator = res[Symbol.asyncIterator](); @@ -83,9 +83,9 @@ function wrapResponse(res: http.IncomingMessage): FetchResponse { export function createNodeFetch( agent?: https.Agent | http.Agent, tlsOptions?: LDTLSOptions, -): FetchFn { +): FetchLike { const tlsParams = tlsRequestOptions(tlsOptions); - return async (url: string, init: FetchRequestOptions): Promise => { + return async (url: string, init: FetchLikeOptions): Promise => { const isSecure = url.startsWith('https://'); const impl = isSecure ? https : http; const requestOptions: https.RequestOptions & Record = { @@ -96,7 +96,7 @@ export function createNodeFetch( if (isSecure) { Object.assign(requestOptions, tlsParams); } - return new Promise((resolve, reject) => { + return new Promise((resolve, reject) => { const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. req.setNoDelay(true); From 51039dacb5fd8f454d4820c03af0fd90b563ace6 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Thu, 8 Oct 2026 15:54:55 -0400 Subject: [PATCH 4/6] chore: fixing deps --- packages/sdk/node-client/package.json | 4 ++-- packages/sdk/server-node/package.json | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/sdk/node-client/package.json b/packages/sdk/node-client/package.json index 7106824c31..589239807b 100644 --- a/packages/sdk/node-client/package.json +++ b/packages/sdk/node-client/package.json @@ -44,8 +44,8 @@ "contract-tests": "./contract-tests/run-contract-tests.sh" }, "dependencies": { - "@launchdarkly/js-client-sdk-common": "1.32.5", - "@launchdarkly/eventsource": "0.1.0" + "@launchdarkly/eventsource": "0.1.0", + "@launchdarkly/js-client-sdk-common": "1.32.5" }, "devDependencies": { "@types/jest": "^29.4.0", diff --git a/packages/sdk/server-node/package.json b/packages/sdk/server-node/package.json index 34ac8f1cd4..c21ee19c29 100644 --- a/packages/sdk/server-node/package.json +++ b/packages/sdk/server-node/package.json @@ -47,9 +47,9 @@ }, "license": "Apache-2.0", "dependencies": { + "@launchdarkly/eventsource": "0.1.0", "@launchdarkly/js-server-sdk-common": "2.24.1", - "https-proxy-agent": "^7.0.6", - "@launchdarkly/eventsource": "0.1.0" + "https-proxy-agent": "^7.0.6" }, "devDependencies": { "@types/jest": "^29.4.0", From e10f9a94ac2df26ec56015e94bc846021f8f562c Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Thu, 8 Oct 2026 22:27:38 -0400 Subject: [PATCH 5/6] refactor: rebase to common fetch --- ....ts => ElectronRequests.streaming.test.ts} | 71 +++-- .../ElectronStreamingResponse.test.ts | 80 ++++++ .../__tests__/platform/HeaderWrapper.test.ts | 12 + .../electron/src/platform/ElectronFetch.ts | 76 ------ .../electron/src/platform/ElectronRequests.ts | 54 +++- .../src/platform/ElectronStreamingResponse.ts | 55 ++++ .../electron/src/platform/HeaderWrapper.ts | 11 + .../__tests__/platform/HeaderWrapper.test.ts | 15 ++ .../platform/NodeEventSource.test.ts | 153 ----------- .../platform/NodeRequests.streaming.test.ts} | 108 ++++---- .../platform/NodeStreamingResponse.test.ts | 80 ++++++ .../node-client/src/platform/HeaderWrapper.ts | 11 + .../src/platform/NodeEventSource.ts | 145 ---------- .../node-client/src/platform/NodeRequests.ts | 93 ++++++- .../src/platform/NodeStreamingResponse.ts | 55 ++++ .../__tests__/platform/HeaderWrapper.test.ts | 12 + .../platform/NodeRequests.streaming.test.ts | 253 ++++++++++++++++++ .../platform/NodeStreamingResponse.test.ts | 80 ++++++ .../server-node/src/platform/HeaderWrapper.ts | 11 + .../src/platform/NodeEventSource.ts | 146 ---------- .../server-node/src/platform/NodeRequests.ts | 99 ++++++- .../src/platform/NodeStreamingResponse.ts | 55 ++++ 22 files changed, 1071 insertions(+), 604 deletions(-) rename packages/sdk/electron/__tests__/platform/{ElectronFetch.test.ts => ElectronRequests.streaming.test.ts} (52%) create mode 100644 packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts delete mode 100644 packages/sdk/electron/src/platform/ElectronFetch.ts create mode 100644 packages/sdk/electron/src/platform/ElectronStreamingResponse.ts delete mode 100644 packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts rename packages/sdk/{server-node/__tests__/platform/NodeEventSource.test.ts => node-client/__tests__/platform/NodeRequests.streaming.test.ts} (55%) create mode 100644 packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts delete mode 100644 packages/sdk/node-client/src/platform/NodeEventSource.ts create mode 100644 packages/sdk/node-client/src/platform/NodeStreamingResponse.ts create mode 100644 packages/sdk/server-node/__tests__/platform/NodeRequests.streaming.test.ts create mode 100644 packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts delete mode 100644 packages/sdk/server-node/src/platform/NodeEventSource.ts create mode 100644 packages/sdk/server-node/src/platform/NodeStreamingResponse.ts diff --git a/packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts b/packages/sdk/electron/__tests__/platform/ElectronRequests.streaming.test.ts similarity index 52% rename from packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts rename to packages/sdk/electron/__tests__/platform/ElectronRequests.streaming.test.ts index 8a138e4d66..e19bad3cc7 100644 --- a/packages/sdk/electron/__tests__/platform/ElectronFetch.test.ts +++ b/packages/sdk/electron/__tests__/platform/ElectronRequests.streaming.test.ts @@ -5,7 +5,6 @@ import { TestHttpServers, } from 'launchdarkly-js-test-helpers'; -import createElectronFetch from '../../src/platform/ElectronFetch'; import ElectronRequests from '../../src/platform/ElectronRequests'; describe('given a running HTTP server', () => { @@ -19,26 +18,27 @@ describe('given a running HTTP server', () => { await server.closeAndWait(); }); - it('forwards the request and exposes the status, headers, and body chunks', async () => { + it('forwards a streaming request and exposes the status, headers, and body chunks', async () => { const chunks = new AsyncQueue(); chunks.add('first'); server.byDefault( TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), ); - const electronFetch = createElectronFetch(); - const res = await electronFetch(`${server.url}/stream`, { + const requests = new ElectronRequests(); + const res = await requests.fetch(`${server.url}/stream`, { method: 'REPORT', headers: { authorization: 'sdk-key' }, body: '{"kind":"user"}', + streaming: true, }); expect(res.status).toEqual(200); - const headers: Record = {}; - res.headers.forEach((value, key) => { - headers[key] = value; + const collected: Record = {}; + res.headers.forEach?.((value, key) => { + collected[key] = value; }); - expect(headers['content-type']).toEqual('text/event-stream'); + expect(collected['content-type']).toEqual('text/event-stream'); const reader = res.body?.getReader(); const first = await reader?.read(); @@ -51,16 +51,57 @@ describe('given a running HTTP server', () => { expect(received.body).toEqual('{"kind":"user"}'); }); - it('does not follow redirects', async () => { + it('does not request compressed content for a streaming request', async () => { + const chunks = new AsyncQueue(); + server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); + + const requests = new ElectronRequests(); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + const received = await server.nextRequest(); + expect(received.headers['accept-encoding']).toBeUndefined(); + }); + + it('does not follow redirects for a streaming request', async () => { server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); - const electronFetch = createElectronFetch(); - const res = await electronFetch(server.url, { method: 'GET', headers: {} }); + const requests = new ElectronRequests(); + const res = await requests.fetch(server.url, { method: 'GET', streaming: true }); expect(res.status).toEqual(301); expect(server.requestCount()).toEqual(1); }); + it('stops the stream when the signal aborts', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); + + const controller = new AbortController(); + const requests = new ElectronRequests(); + const res = await requests.fetch(server.url, { + method: 'GET', + streaming: true, + signal: controller.signal, + }); + const reader = res.body?.getReader(); + await reader?.read(); + const pending = reader?.read(); + controller.abort(); + await expect(pending).rejects.toThrow(); + }); + + it('rejects a streaming request when the signal is already aborted', async () => { + server.byDefault(TestHttpHandlers.respond(200)); + const controller = new AbortController(); + controller.abort(); + + const requests = new ElectronRequests(); + await expect( + requests.fetch(server.url, { method: 'GET', streaming: true, signal: controller.signal }), + ).rejects.toThrow(); + }); + it('streams SSE events through createEventSource', async () => { const chunks = new AsyncQueue(); chunks.add('data: hello\n\n'); @@ -99,10 +140,10 @@ describe('given a running HTTPS server with a self-signed certificate', () => { await server.closeAndWait(); }); - it('rejects the connection when the certificate is not trusted', async () => { - // The Electron SDK exposes no TLS options. Verification follows the platform default, so a + it('rejects a streaming request when the certificate is not trusted', async () => { + // This SDK exposes no TLS options. Verification follows the platform default, so a // self-signed certificate the machine does not trust must fail the request. - const electronFetch = createElectronFetch(); - await expect(electronFetch(server.url, { method: 'GET', headers: {} })).rejects.toThrow(); + const requests = new ElectronRequests(); + await expect(requests.fetch(server.url, { method: 'GET', streaming: true })).rejects.toThrow(); }); }); diff --git a/packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts b/packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts new file mode 100644 index 0000000000..853b0c7002 --- /dev/null +++ b/packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts @@ -0,0 +1,80 @@ +import * as http from 'http'; +import { Readable } from 'stream'; + +import ElectronStreamingResponse from '../../src/platform/ElectronStreamingResponse'; + +function makeIncomingMessage( + chunks: Buffer[], + options: { + headers?: http.IncomingHttpHeaders; + statusCode?: number; + statusMessage?: string; + } = {}, +): http.IncomingMessage { + const stream = Readable.from(chunks) as unknown as http.IncomingMessage; + stream.headers = options.headers ?? {}; + stream.statusCode = 'statusCode' in options ? options.statusCode : 200; + if (options.statusMessage !== undefined) { + stream.statusMessage = options.statusMessage; + } + return stream; +} + +it('exposes the status code and status message', () => { + const res = new ElectronStreamingResponse( + makeIncomingMessage([], { statusCode: 301, statusMessage: 'Moved Permanently' }), + ); + expect(res.status).toBe(301); + expect(res.statusText).toBe('Moved Permanently'); +}); + +it('defaults the status to 0 and the status message to an empty string when absent', () => { + const res = new ElectronStreamingResponse(makeIncomingMessage([], { statusCode: undefined })); + expect(res.status).toBe(0); + expect(res.statusText).toBe(''); +}); + +it('wraps the response headers in a HeaderWrapper', () => { + const res = new ElectronStreamingResponse( + makeIncomingMessage([], { headers: { 'content-type': 'text/event-stream' } }), + ); + expect(res.headers.get('content-type')).toBe('text/event-stream'); +}); + +it('hands out one chunk per read and then reports done', async () => { + const res = new ElectronStreamingResponse( + makeIncomingMessage([Buffer.from('first'), Buffer.from('second')]), + ); + const reader = res.body.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(Buffer.from(first.value ?? []).toString()).toBe('first'); + const second = await reader.read(); + expect(Buffer.from(second.value ?? []).toString()).toBe('second'); + const end = await reader.read(); + expect(end.done).toBe(true); +}); + +it('reads the remaining stream as text', async () => { + const res = new ElectronStreamingResponse( + makeIncomingMessage([Buffer.from('hello '), Buffer.from('world')]), + ); + await expect(res.text()).resolves.toBe('hello world'); +}); + +it('parses the remaining stream as JSON', async () => { + const res = new ElectronStreamingResponse(makeIncomingMessage([Buffer.from('{"a":1}')])); + await expect(res.json()).resolves.toEqual({ a: 1 }); +}); + +it('rejects a pending read when the stream errors', async () => { + const erroring = new Readable({ + read() { + this.destroy(new Error('boom')); + }, + }) as unknown as http.IncomingMessage; + erroring.headers = {}; + erroring.statusCode = 200; + const res = new ElectronStreamingResponse(erroring); + await expect(res.body.getReader().read()).rejects.toThrow('boom'); +}); diff --git a/packages/sdk/electron/__tests__/platform/HeaderWrapper.test.ts b/packages/sdk/electron/__tests__/platform/HeaderWrapper.test.ts index d2c89ae6b9..bc1f40f1b8 100644 --- a/packages/sdk/electron/__tests__/platform/HeaderWrapper.test.ts +++ b/packages/sdk/electron/__tests__/platform/HeaderWrapper.test.ts @@ -50,4 +50,16 @@ describe('given header values', () => { } expect(values).toEqual(['anything', 'some-value', 'a, b']); }); + + it('iterates each header with the value before the key', () => { + const collected: [string, string][] = []; + wrapper.forEach((value, key) => { + collected.push([value, key]); + }); + expect(collected).toEqual([ + ['anything', 'accept'], + ['some-value', 'some-header'], + ['a, b', 'some-array'], + ]); + }); }); diff --git a/packages/sdk/electron/src/platform/ElectronFetch.ts b/packages/sdk/electron/src/platform/ElectronFetch.ts deleted file mode 100644 index b3762ef4e9..0000000000 --- a/packages/sdk/electron/src/platform/ElectronFetch.ts +++ /dev/null @@ -1,76 +0,0 @@ -import * as http from 'http'; -import * as https from 'https'; - -import type { FetchLike, FetchLikeOptions, FetchLikeResponse } from '@launchdarkly/eventsource'; - -function wrapResponse(res: http.IncomingMessage): FetchLikeResponse { - // The async iterator hands out one chunk per read. It resolves done when the server ends the - // stream, and it rejects when the connection drops or the request is destroyed. - const iterator = res[Symbol.asyncIterator](); - return { - status: res.statusCode ?? 0, - statusText: res.statusMessage ?? '', - headers: { - forEach(callback: (value: string, key: string) => void): void { - Object.entries(res.headers).forEach(([key, value]) => { - if (value === undefined) { - return; - } - callback(Array.isArray(value) ? value.join(', ') : value, key); - }); - }, - }, - body: { - getReader: () => ({ - read: async () => { - const next = await iterator.next(); - if (next.done) { - return { done: true }; - } - return { done: false, value: next.value as Uint8Array }; - }, - }), - }, - }; -} - -/** - * Creates a `fetch()`-shaped function over `node:http`/`node:https` for the SSE stream. - * - * The Electron SDK exposes no agent, proxy, or TLS options: the running machine's own network - * configuration applies, and TLS verification follows the platform default. The returned function - * never follows a redirect: a redirect status resolves like any other non-200 response, and the - * caller decides whether to retry the original URL. It applies no read or socket timeout; the - * caller owns the read timeout and cancels through the request's `AbortSignal`. - */ -export default function createElectronFetch(): FetchLike { - return async (url: string, init: FetchLikeOptions): Promise => { - const isSecure = url.startsWith('https://'); - const impl = isSecure ? https : http; - const requestOptions: https.RequestOptions = { - method: init.method, - headers: init.headers, - }; - return new Promise((resolve, reject) => { - const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); - // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. - req.setNoDelay(true); - const { signal } = init; - if (signal) { - const abort = () => req.destroy(new Error('The stream request was aborted')); - if (signal.aborted) { - abort(); - } else { - signal.addEventListener('abort', abort, { once: true }); - } - } - // stays attached after resolve, so a later socket error becomes a harmless no-op reject - // instead of an unhandled 'error' event that would crash the process - req.on('error', reject); - if (init.body !== undefined) { - req.write(init.body); - } - req.end(); - }); - }; -} diff --git a/packages/sdk/electron/src/platform/ElectronRequests.ts b/packages/sdk/electron/src/platform/ElectronRequests.ts index fe3516c578..f9bfe145fd 100644 --- a/packages/sdk/electron/src/platform/ElectronRequests.ts +++ b/packages/sdk/electron/src/platform/ElectronRequests.ts @@ -3,25 +3,25 @@ import * as https from 'https'; import { promisify } from 'util'; import * as zlib from 'zlib'; -import { createEventSource, FetchLike } from '@launchdarkly/eventsource'; +import { createEventSource } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; -import createElectronFetch from './ElectronFetch'; import ElectronResponse from './ElectronResponse'; +import ElectronStreamingResponse from './ElectronStreamingResponse'; const gzip = promisify(zlib.gzip); export default class ElectronRequests implements platform.Requests { - private _eventSourceFetch: FetchLike; - private _enableBodyCompression: boolean = false; constructor(enableEventCompression?: boolean) { - this._eventSourceFetch = createElectronFetch(); this._enableBodyCompression = !!enableEventCompression; } async fetch(url: string, options: platform.Options = {}): Promise { + if (options.streaming) { + return this._streamingFetch(url, options); + } const isSecure = url.startsWith('https://'); const impl = isSecure ? https : http; @@ -29,7 +29,7 @@ export default class ElectronRequests implements platform.Requests { let bodyData: string | Buffer | undefined = options.body; // For get requests we are going to automatically support compressed responses. - // Note this does not affect SSE as the event source is not using this fetch implementation. + // Note this does not affect SSE as streaming requests take the branch above. if (options.method?.toLowerCase() === 'get') { headers['accept-encoding'] = 'gzip'; } @@ -69,6 +69,46 @@ export default class ElectronRequests implements platform.Requests { }); } + /** + * The transport for a streaming request. This SDK exposes no agent, proxy, or TLS options, so + * the running machine's own network configuration applies and TLS verification follows the + * platform default. It does not request compressed content, and it never follows a redirect. + * A redirect status resolves like any other non-200 response, and the caller decides whether + * to retry the original URL. It applies no read or socket timeout. The caller owns the read + * timeout and cancels through the abort signal. + */ + private _streamingFetch(url: string, options: platform.Options): Promise { + const isSecure = url.startsWith('https://'); + const impl = isSecure ? https : http; + const requestOptions: https.RequestOptions = { + method: options.method, + headers: options.headers, + }; + return new Promise((resolve, reject) => { + const req = impl.request(url, requestOptions, (res) => + resolve(new ElectronStreamingResponse(res)), + ); + // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. + req.setNoDelay(true); + const { signal } = options; + if (signal) { + const abort = () => req.destroy(new Error('The stream request was aborted')); + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + // This listener stays attached after resolve. A later socket error then becomes a harmless + // no-op reject instead of an unhandled 'error' event that would crash the process + req.on('error', reject); + if (options.body !== undefined) { + req.write(options.body); + } + req.end(); + }); + } + createEventSource( url: string, eventSourceInitDict: platform.EventSourceInitDict, @@ -77,7 +117,7 @@ export default class ElectronRequests implements platform.Requests { ...eventSourceInitDict, maxBackoffMillis: 30 * 1000, jitterRatio: 0.5, - fetch: this._eventSourceFetch, + fetch: (fetchUrl, init) => this.fetch(fetchUrl, { ...init, streaming: true }), }); } diff --git a/packages/sdk/electron/src/platform/ElectronStreamingResponse.ts b/packages/sdk/electron/src/platform/ElectronStreamingResponse.ts new file mode 100644 index 0000000000..656f6c7b75 --- /dev/null +++ b/packages/sdk/electron/src/platform/ElectronStreamingResponse.ts @@ -0,0 +1,55 @@ +import * as http from 'http'; + +import { platform } from '@launchdarkly/js-client-sdk-common'; + +import HeaderWrapper from './HeaderWrapper'; + +/** + * A response for a streaming request. It does not buffer the body: each read hands out the next + * chunk from the socket. The `text` and `json` methods read the remainder of the stream, so a + * caller uses either the reader or those methods, not both. + */ +export default class ElectronStreamingResponse implements platform.Response { + headers: platform.Headers; + + status: number; + + statusText: string; + + body: platform.ResponseBody; + + private _iterator: AsyncIterableIterator; + + constructor(res: http.IncomingMessage) { + this.headers = new HeaderWrapper(res.headers); + this.status = res.statusCode ?? 0; + this.statusText = res.statusMessage ?? ''; + // The async iterator hands out one chunk per read. It resolves done when the server ends the + // stream, and it rejects when the connection drops or the request is destroyed. + const iterator = res[Symbol.asyncIterator](); + this._iterator = iterator; + this.body = { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value }; + }, + }), + }; + } + + async text(): Promise { + const chunks: Uint8Array[] = []; + for await (const chunk of this._iterator) { + chunks.push(chunk); + } + return Buffer.concat(chunks).toString(); + } + + async json(): Promise { + return JSON.parse(await this.text()); + } +} diff --git a/packages/sdk/electron/src/platform/HeaderWrapper.ts b/packages/sdk/electron/src/platform/HeaderWrapper.ts index 9c447c2a8f..f6922bf30d 100644 --- a/packages/sdk/electron/src/platform/HeaderWrapper.ts +++ b/packages/sdk/electron/src/platform/HeaderWrapper.ts @@ -52,6 +52,17 @@ export default class HeaderWrapper implements platform.Headers { } } + /** + * Executes the callback once for each header, with the value first. The order matches the + * fetch `Headers.forEach` signature. Multi-value headers are joined with a comma, and + * headers without a value are skipped, like `entries`. + */ + forEach(callback: (value: string, key: string) => void): void { + for (const [key, value] of this.entries()) { + callback(value, key); + } + } + has(name: string): boolean { return Object.prototype.hasOwnProperty.call(this._headers, name); } diff --git a/packages/sdk/node-client/__tests__/platform/HeaderWrapper.test.ts b/packages/sdk/node-client/__tests__/platform/HeaderWrapper.test.ts index 72233ae5b8..e55624f382 100644 --- a/packages/sdk/node-client/__tests__/platform/HeaderWrapper.test.ts +++ b/packages/sdk/node-client/__tests__/platform/HeaderWrapper.test.ts @@ -49,3 +49,18 @@ it('reports presence with has()', () => { expect(wrapper.has('content-type')).toBe(true); expect(wrapper.has('missing')).toBe(false); }); + +it('iterates each header with the value before the key', () => { + const wrapper = new HeaderWrapper({ + accept: 'anything', + 'some-array': ['a', 'b'], + }); + const collected: [string, string][] = []; + wrapper.forEach((value, key) => { + collected.push([value, key]); + }); + expect(collected).toEqual([ + ['anything', 'accept'], + ['a, b', 'some-array'], + ]); +}); diff --git a/packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts b/packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts deleted file mode 100644 index 00906a17f9..0000000000 --- a/packages/sdk/node-client/__tests__/platform/NodeEventSource.test.ts +++ /dev/null @@ -1,153 +0,0 @@ -import * as http from 'http'; -import { - AsyncQueue, - TestHttpHandlers, - TestHttpServer, - TestHttpServers, -} from 'launchdarkly-js-test-helpers'; - -import { createNodeFetch } from '../../src/platform/NodeEventSource'; -import NodeRequests from '../../src/platform/NodeRequests'; - -describe('given a running HTTP server', () => { - let server: TestHttpServer; - - beforeEach(async () => { - server = await TestHttpServers.start(); - }); - - afterEach(async () => { - await server.closeAndWait(); - }); - - it('forwards the request and exposes the status, headers, and body chunks', async () => { - const chunks = new AsyncQueue(); - chunks.add('first'); - server.byDefault( - TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), - ); - - const nodeFetch = createNodeFetch(); - const res = await nodeFetch(`${server.url}/stream`, { - method: 'REPORT', - headers: { authorization: 'sdk-key' }, - body: '{"kind":"user"}', - }); - - expect(res.status).toEqual(200); - const headers: Record = {}; - res.headers.forEach((value, key) => { - headers[key] = value; - }); - expect(headers['content-type']).toEqual('text/event-stream'); - - const reader = res.body?.getReader(); - const first = await reader?.read(); - expect(first?.done).toBe(false); - expect(Buffer.from(first?.value ?? []).toString()).toEqual('first'); - - const received = await server.nextRequest(); - expect(received.method.toUpperCase()).toEqual('REPORT'); - expect(received.headers.authorization).toEqual('sdk-key'); - expect(received.body).toEqual('{"kind":"user"}'); - }); - - it('does not follow redirects', async () => { - server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); - - const nodeFetch = createNodeFetch(); - const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); - - expect(res.status).toEqual(301); - expect(server.requestCount()).toEqual(1); - }); - - it('uses the supplied agent', async () => { - server.byDefault(TestHttpHandlers.respond(200)); - const agent = new http.Agent({ keepAlive: false }); - // addRequest exists at runtime but is not part of the public http.Agent type. - // @ts-ignore - const addRequestSpy = jest.spyOn(agent, 'addRequest'); - - // The runtime accepts an http.Agent even though this SDK only ever builds an https.Agent. - // @ts-ignore - const nodeFetch = createNodeFetch(agent); - await nodeFetch(server.url, { method: 'GET', headers: {} }); - - expect(addRequestSpy).toHaveBeenCalled(); - }); - - it('streams SSE events through createEventSource', async () => { - const chunks = new AsyncQueue(); - chunks.add('data: hello\n\n'); - server.byDefault( - TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), - ); - - const requests = new NodeRequests(); - const es = requests.createEventSource(`${server.url}/stream`, { - headers: {}, - initialRetryDelayMillis: 100, - readTimeoutMillis: 5000, - retryResetIntervalMillis: 30_000, - errorFilter: () => false, - }); - try { - const messages = new AsyncQueue<{ data?: string }>(); - es.addEventListener('message', (event) => messages.add(event ?? {})); - const message = await messages.take(); - expect(message.data).toEqual('hello'); - } finally { - es.close(); - } - }); -}); - -describe('given a running HTTPS server with a self-signed certificate', () => { - let server: TestHttpServer; - - beforeEach(async () => { - server = await TestHttpServers.startSecure(); - server.byDefault(TestHttpHandlers.respond(200)); - }); - - afterEach(async () => { - await server.closeAndWait(); - }); - - it('connects when the CA is supplied in the TLS options', async () => { - const nodeFetch = createNodeFetch(undefined, { ca: server.certificate }); - const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); - expect(res.status).toEqual(200); - }); - - it('rejects the connection when no CA is supplied', async () => { - const nodeFetch = createNodeFetch(); - await expect(nodeFetch(server.url, { method: 'GET', headers: {} })).rejects.toThrow(); - }); - - it('streams over TLS when the CA is supplied through tlsParams', async () => { - const chunks = new AsyncQueue(); - chunks.add('data: secure\n\n'); - server.byDefault( - TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), - ); - - const requests = new NodeRequests({ ca: server.certificate }); - const es = requests.createEventSource(`${server.url}/stream`, { - headers: {}, - initialRetryDelayMillis: 100, - readTimeoutMillis: 5000, - retryResetIntervalMillis: 30_000, - errorFilter: () => false, - }); - try { - const messages = new AsyncQueue<{ data?: string }>(); - es.addEventListener('message', (event) => messages.add(event ?? {})); - const message = await messages.take(); - expect(message.data).toEqual('secure'); - } finally { - es.close(); - } - }); -}); diff --git a/packages/sdk/server-node/__tests__/platform/NodeEventSource.test.ts b/packages/sdk/node-client/__tests__/platform/NodeRequests.streaming.test.ts similarity index 55% rename from packages/sdk/server-node/__tests__/platform/NodeEventSource.test.ts rename to packages/sdk/node-client/__tests__/platform/NodeRequests.streaming.test.ts index b1e855677e..135d5ef3d2 100644 --- a/packages/sdk/server-node/__tests__/platform/NodeEventSource.test.ts +++ b/packages/sdk/node-client/__tests__/platform/NodeRequests.streaming.test.ts @@ -1,4 +1,5 @@ -import * as http from 'http'; +import http from 'http'; +import https from 'https'; import { AsyncQueue, TestHttpHandlers, @@ -6,7 +7,6 @@ import { TestHttpServers, } from 'launchdarkly-js-test-helpers'; -import { createNodeFetch } from '../../src/platform/NodeEventSource'; import NodeRequests from '../../src/platform/NodeRequests'; describe('given a running HTTP server', () => { @@ -17,29 +17,31 @@ describe('given a running HTTP server', () => { }); afterEach(async () => { + jest.restoreAllMocks(); await server.closeAndWait(); }); - it('forwards the request and exposes the status, headers, and body chunks', async () => { + it('forwards a streaming request and exposes the status, headers, and body chunks', async () => { const chunks = new AsyncQueue(); chunks.add('first'); server.byDefault( TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), ); - const nodeFetch = createNodeFetch(); - const res = await nodeFetch(`${server.url}/stream`, { + const requests = new NodeRequests(); + const res = await requests.fetch(`${server.url}/stream`, { method: 'REPORT', headers: { authorization: 'sdk-key' }, body: '{"kind":"user"}', + streaming: true, }); expect(res.status).toEqual(200); - const headers: Record = {}; - res.headers.forEach((value, key) => { - headers[key] = value; + const collected: Record = {}; + res.headers.forEach?.((value, key) => { + collected[key] = value; }); - expect(headers['content-type']).toEqual('text/event-stream'); + expect(collected['content-type']).toEqual('text/event-stream'); const reader = res.body?.getReader(); const first = await reader?.read(); @@ -52,11 +54,22 @@ describe('given a running HTTP server', () => { expect(received.body).toEqual('{"kind":"user"}'); }); - it('does not follow redirects', async () => { + it('does not request compressed content for a streaming request', async () => { + const chunks = new AsyncQueue(); + server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); + + const requests = new NodeRequests(); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + const received = await server.nextRequest(); + expect(received.headers['accept-encoding']).toBeUndefined(); + }); + + it('does not follow redirects for a streaming request', async () => { server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); - const nodeFetch = createNodeFetch(); - const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); + const requests = new NodeRequests(); + const res = await requests.fetch(server.url, { method: 'GET', streaming: true }); expect(res.status).toEqual(301); expect(server.requestCount()).toEqual(1); @@ -68,10 +81,10 @@ describe('given a running HTTP server', () => { server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); const controller = new AbortController(); - const nodeFetch = createNodeFetch(); - const res = await nodeFetch(server.url, { + const requests = new NodeRequests(); + const res = await requests.fetch(server.url, { method: 'GET', - headers: {}, + streaming: true, signal: controller.signal, }); const reader = res.body?.getReader(); @@ -81,17 +94,27 @@ describe('given a running HTTP server', () => { await expect(pending).rejects.toThrow(); }); - it('uses the supplied agent', async () => { + it('rejects a streaming request when the signal is already aborted', async () => { server.byDefault(TestHttpHandlers.respond(200)); - const agent = new http.Agent({ keepAlive: false }); - // addRequest exists at runtime but is not part of the public http.Agent type. - // @ts-ignore - const addRequestSpy = jest.spyOn(agent, 'addRequest'); + const controller = new AbortController(); + controller.abort(); + + const requests = new NodeRequests(); + await expect( + requests.fetch(server.url, { method: 'GET', streaming: true, signal: controller.signal }), + ).rejects.toThrow(); + }); - const nodeFetch = createNodeFetch(agent); - await nodeFetch(server.url, { method: 'GET', headers: {} }); + it('does not pass the CA to the per-request options of an http streaming request', async () => { + const requestSpy = jest.spyOn(http, 'request'); - expect(addRequestSpy).toHaveBeenCalled(); + const requests = new NodeRequests({ ca: 'unused-ca' }); + // The https agent built from the TLS options is rejected for an http URL, so the request + // fails after http.request has already received its options. + await requests.fetch(server.url, { method: 'GET', streaming: true }).catch(() => undefined); + + expect(requestSpy).toHaveBeenCalledTimes(1); + expect(requestSpy.mock.calls[0][1]).not.toHaveProperty('ca'); }); it('streams SSE events through createEventSource', async () => { @@ -129,39 +152,34 @@ describe('given a running HTTPS server with a self-signed certificate', () => { }); afterEach(async () => { + jest.restoreAllMocks(); await server.closeAndWait(); }); - it('connects when the CA is supplied in the TLS options', async () => { - const nodeFetch = createNodeFetch(undefined, { ca: server.certificate }); - const res = await nodeFetch(server.url, { method: 'GET', headers: {} }); + it('connects a streaming request when the CA is supplied in the TLS options', async () => { + const requests = new NodeRequests({ ca: server.certificate }); + const res = await requests.fetch(server.url, { method: 'GET', streaming: true }); expect(res.status).toEqual(200); }); - it('rejects the connection when no CA is supplied', async () => { - const nodeFetch = createNodeFetch(); - await expect(nodeFetch(server.url, { method: 'GET', headers: {} })).rejects.toThrow(); + it('passes the CA to the per-request options of an https streaming request', async () => { + const requestSpy = jest.spyOn(https, 'request'); + + const requests = new NodeRequests({ ca: server.certificate }); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + expect(requestSpy).toHaveBeenCalledTimes(1); + expect(requestSpy.mock.calls[0][1]).toEqual( + expect.objectContaining({ ca: server.certificate }), + ); }); - it('fails the stream against an untrusted certificate by default', async () => { + it('rejects a streaming request when no CA is supplied', async () => { const requests = new NodeRequests(); - const es = requests.createEventSource(`${server.url}/stream`, { - headers: {}, - initialRetryDelayMillis: 100, - readTimeoutMillis: 5000, - retryResetIntervalMillis: 30_000, - errorFilter: () => false, - }); - try { - const errors = new AsyncQueue(); - es.addEventListener('error', (event) => errors.add(event)); - expect(await errors.take()).toBeDefined(); - } finally { - es.close(); - } + await expect(requests.fetch(server.url, { method: 'GET', streaming: true })).rejects.toThrow(); }); - it('streams over TLS when the CA is supplied through tlsParams', async () => { + it('streams over TLS when the CA is supplied in the TLS options', async () => { const chunks = new AsyncQueue(); chunks.add('data: secure\n\n'); server.byDefault( diff --git a/packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts b/packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts new file mode 100644 index 0000000000..c7ec88fc08 --- /dev/null +++ b/packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts @@ -0,0 +1,80 @@ +import * as http from 'http'; +import { Readable } from 'stream'; + +import NodeStreamingResponse from '../../src/platform/NodeStreamingResponse'; + +function makeIncomingMessage( + chunks: Buffer[], + options: { + headers?: http.IncomingHttpHeaders; + statusCode?: number; + statusMessage?: string; + } = {}, +): http.IncomingMessage { + const stream = Readable.from(chunks) as unknown as http.IncomingMessage; + stream.headers = options.headers ?? {}; + stream.statusCode = 'statusCode' in options ? options.statusCode : 200; + if (options.statusMessage !== undefined) { + stream.statusMessage = options.statusMessage; + } + return stream; +} + +it('exposes the status code and status message', () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([], { statusCode: 301, statusMessage: 'Moved Permanently' }), + ); + expect(res.status).toBe(301); + expect(res.statusText).toBe('Moved Permanently'); +}); + +it('defaults the status to 0 and the status message to an empty string when absent', () => { + const res = new NodeStreamingResponse(makeIncomingMessage([], { statusCode: undefined })); + expect(res.status).toBe(0); + expect(res.statusText).toBe(''); +}); + +it('wraps the response headers in a HeaderWrapper', () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([], { headers: { 'content-type': 'text/event-stream' } }), + ); + expect(res.headers.get('content-type')).toBe('text/event-stream'); +}); + +it('hands out one chunk per read and then reports done', async () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([Buffer.from('first'), Buffer.from('second')]), + ); + const reader = res.body.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(Buffer.from(first.value ?? []).toString()).toBe('first'); + const second = await reader.read(); + expect(Buffer.from(second.value ?? []).toString()).toBe('second'); + const end = await reader.read(); + expect(end.done).toBe(true); +}); + +it('reads the remaining stream as text', async () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([Buffer.from('hello '), Buffer.from('world')]), + ); + await expect(res.text()).resolves.toBe('hello world'); +}); + +it('parses the remaining stream as JSON', async () => { + const res = new NodeStreamingResponse(makeIncomingMessage([Buffer.from('{"a":1}')])); + await expect(res.json()).resolves.toEqual({ a: 1 }); +}); + +it('rejects a pending read when the stream errors', async () => { + const erroring = new Readable({ + read() { + this.destroy(new Error('boom')); + }, + }) as unknown as http.IncomingMessage; + erroring.headers = {}; + erroring.statusCode = 200; + const res = new NodeStreamingResponse(erroring); + await expect(res.body.getReader().read()).rejects.toThrow('boom'); +}); diff --git a/packages/sdk/node-client/src/platform/HeaderWrapper.ts b/packages/sdk/node-client/src/platform/HeaderWrapper.ts index d3812f1937..9b30516989 100644 --- a/packages/sdk/node-client/src/platform/HeaderWrapper.ts +++ b/packages/sdk/node-client/src/platform/HeaderWrapper.ts @@ -50,6 +50,17 @@ export default class HeaderWrapper implements platform.Headers { } } + /** + * Executes the callback once for each header, with the value first. The order matches the + * fetch `Headers.forEach` signature. Multi-value headers are joined with a comma, and + * headers without a value are skipped, like `entries`. + */ + forEach(callback: (value: string, key: string) => void): void { + for (const [key, value] of this.entries()) { + callback(value, key); + } + } + has(name: string): boolean { return Object.prototype.hasOwnProperty.call(this._headers, name); } diff --git a/packages/sdk/node-client/src/platform/NodeEventSource.ts b/packages/sdk/node-client/src/platform/NodeEventSource.ts deleted file mode 100644 index 003d47521f..0000000000 --- a/packages/sdk/node-client/src/platform/NodeEventSource.ts +++ /dev/null @@ -1,145 +0,0 @@ -import * as http from 'http'; -import * as https from 'https'; - -import { - createEventSource, - type FetchLike, - type FetchLikeOptions, - type FetchLikeResponse, -} from '@launchdarkly/eventsource'; -import type { platform } from '@launchdarkly/js-client-sdk-common'; - -import type { LDTLSOptions } from '../NodeOptions'; - -/** - * The TLS options that are copied onto each `https` request. The names match the options of - * `https.request()`. - */ -const TLS_OPTION_NAMES = [ - 'pfx', - 'key', - 'passphrase', - 'cert', - 'ca', - 'ciphers', - 'rejectUnauthorized', - 'secureProtocol', - 'servername', - 'checkServerIdentity', -] as const; - -function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { - const merged: Record = {}; - if (!tlsOptions) { - return merged; - } - const bag = tlsOptions as unknown as Record; - TLS_OPTION_NAMES.forEach((name) => { - if (bag[name] !== undefined) { - merged[name] = bag[name]; - } - }); - return merged; -} - -function wrapResponse(res: http.IncomingMessage): FetchLikeResponse { - // The async iterator hands out one chunk per read. It resolves done when the server ends the - // stream, and it rejects when the connection drops or the request is destroyed. - const iterator = res[Symbol.asyncIterator](); - return { - status: res.statusCode ?? 0, - statusText: res.statusMessage ?? '', - headers: { - forEach(callback: (value: string, key: string) => void): void { - Object.entries(res.headers).forEach(([key, value]) => { - if (value === undefined) { - return; - } - callback(Array.isArray(value) ? value.join(', ') : value, key); - }); - }, - }, - body: { - getReader: () => ({ - read: async () => { - const next = await iterator.next(); - if (next.done) { - return { done: true }; - } - return { done: false, value: next.value as Uint8Array }; - }, - }), - }, - }; -} - -/** - * Creates a `fetch()`-shaped function over `node:http`/`node:https` for the SSE stream. - * - * The returned function bakes in the connection configuration: the agent (which can carry proxy - * or TLS setup) and the TLS parameters for `https` URLs. It never follows a redirect: a redirect - * status resolves like any other non-200 response, and the caller decides whether to retry the - * original URL. It applies no read or socket timeout; the caller owns the read timeout and - * cancels through the request's `AbortSignal`. - */ -export function createNodeFetch(agent?: https.Agent, tlsOptions?: LDTLSOptions): FetchLike { - const tlsParams = tlsRequestOptions(tlsOptions); - return async (url: string, init: FetchLikeOptions): Promise => { - const isSecure = url.startsWith('https://'); - const impl = isSecure ? https : http; - const requestOptions: https.RequestOptions & Record = { - method: init.method, - headers: init.headers, - agent, - }; - if (isSecure) { - Object.assign(requestOptions, tlsParams); - } - return new Promise((resolve, reject) => { - const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); - // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. - req.setNoDelay(true); - const { signal } = init; - if (signal) { - const abort = () => req.destroy(new Error('The stream request was aborted')); - if (signal.aborted) { - abort(); - } else { - signal.addEventListener('abort', abort, { once: true }); - } - } - // stays attached after resolve, so a later socket error becomes a harmless no-op reject - // instead of an unhandled 'error' event that would crash the process - req.on('error', reject); - if (init.body !== undefined) { - req.write(init.body); - } - req.end(); - }); - }; -} - -export type NodeEventSourceFactory = ( - url: string, - eventSourceInitDict: platform.EventSourceInitDict, -) => platform.EventSource; - -/** - * Creates the event source factory used by `NodeRequests.createEventSource`. - * - * The factory bakes in the connection configuration (agent and TLS parameters, through the fetch - * adapter above) and the SDK retry policy (backoff cap and jitter). - */ -export default function createNodeEventSourceFactory( - agent?: https.Agent, - tlsOptions?: LDTLSOptions, -): NodeEventSourceFactory { - const fetch = createNodeFetch(agent, tlsOptions); - return (url, eventSourceInitDict) => - createEventSource(url, { - ...eventSourceInitDict, - maxBackoffMillis: 30 * 1000, - jitterRatio: 0.5, - fetch, - }); -} diff --git a/packages/sdk/node-client/src/platform/NodeRequests.ts b/packages/sdk/node-client/src/platform/NodeRequests.ts index fd44b857db..9825425685 100644 --- a/packages/sdk/node-client/src/platform/NodeRequests.ts +++ b/packages/sdk/node-client/src/platform/NodeRequests.ts @@ -3,14 +3,46 @@ import * as https from 'https'; import { promisify } from 'util'; import * as zlib from 'zlib'; +import { createEventSource } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; import type { LDTLSOptions } from '../NodeOptions'; -import createNodeEventSourceFactory, { NodeEventSourceFactory } from './NodeEventSource'; import NodeResponse from './NodeResponse'; +import NodeStreamingResponse from './NodeStreamingResponse'; const gzip = promisify(zlib.gzip); +/** + * The TLS options that are copied onto each streaming `https` request. The names match the + * options of `https.request()`. + */ +const TLS_OPTION_NAMES = [ + 'pfx', + 'key', + 'passphrase', + 'cert', + 'ca', + 'ciphers', + 'rejectUnauthorized', + 'secureProtocol', + 'servername', + 'checkServerIdentity', +] as const; + +function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { + const merged: Record = {}; + if (!tlsOptions) { + return merged; + } + const bag = tlsOptions as unknown as Record; + TLS_OPTION_NAMES.forEach((name) => { + if (bag[name] !== undefined) { + merged[name] = bag[name]; + } + }); + return merged; +} + function processTlsOptions(tlsOptions: LDTLSOptions): https.AgentOptions { const options: https.AgentOptions & { [index: string]: any } = { ca: tlsOptions.ca, @@ -41,20 +73,20 @@ function processTlsOptions(tlsOptions: LDTLSOptions): https.AgentOptions { export default class NodeRequests implements platform.Requests { private _agent: https.Agent | undefined; - private _tlsOptions: LDTLSOptions | undefined; - - private _createEventSource: NodeEventSourceFactory; + private _tlsParams: Record; private _enableBodyCompression: boolean = false; constructor(tlsOptions?: LDTLSOptions, enableEventCompression?: boolean) { this._agent = tlsOptions ? new https.Agent(processTlsOptions(tlsOptions)) : undefined; - this._tlsOptions = tlsOptions; - this._createEventSource = createNodeEventSourceFactory(this._agent, this._tlsOptions); + this._tlsParams = tlsRequestOptions(tlsOptions); this._enableBodyCompression = !!enableEventCompression; } async fetch(url: string, options: platform.Options = {}): Promise { + if (options.streaming) { + return this._streamingFetch(url, options); + } const isSecure = url.startsWith('https://'); const impl = isSecure ? https : http; @@ -101,11 +133,58 @@ export default class NodeRequests implements platform.Requests { }); } + /** + * The transport for a streaming request. It does not request compressed content, and it never + * follows a redirect. A redirect status resolves like any other non-200 response, and the + * caller decides whether to retry the original URL. It applies no read or socket timeout. The + * caller owns the read timeout and cancels through the abort signal. + */ + private _streamingFetch(url: string, options: platform.Options): Promise { + const isSecure = url.startsWith('https://'); + const impl = isSecure ? https : http; + const requestOptions: https.RequestOptions & Record = { + method: options.method, + headers: options.headers, + agent: this._agent, + }; + if (isSecure) { + Object.assign(requestOptions, this._tlsParams); + } + return new Promise((resolve, reject) => { + const req = impl.request(url, requestOptions, (res) => + resolve(new NodeStreamingResponse(res)), + ); + // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. + req.setNoDelay(true); + const { signal } = options; + if (signal) { + const abort = () => req.destroy(new Error('The stream request was aborted')); + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + // This listener stays attached after resolve. A later socket error then becomes a harmless + // no-op reject instead of an unhandled 'error' event that would crash the process + req.on('error', reject); + if (options.body !== undefined) { + req.write(options.body); + } + req.end(); + }); + } + createEventSource( url: string, eventSourceInitDict: platform.EventSourceInitDict, ): platform.EventSource { - return this._createEventSource(url, eventSourceInitDict); + return createEventSource(url, { + ...eventSourceInitDict, + maxBackoffMillis: 30 * 1000, + jitterRatio: 0.5, + fetch: (fetchUrl, init) => this.fetch(fetchUrl, { ...init, streaming: true }), + }); } getEventSourceCapabilities(): EventSourceCapabilities { diff --git a/packages/sdk/node-client/src/platform/NodeStreamingResponse.ts b/packages/sdk/node-client/src/platform/NodeStreamingResponse.ts new file mode 100644 index 0000000000..876855c59a --- /dev/null +++ b/packages/sdk/node-client/src/platform/NodeStreamingResponse.ts @@ -0,0 +1,55 @@ +import * as http from 'http'; + +import { platform } from '@launchdarkly/js-client-sdk-common'; + +import HeaderWrapper from './HeaderWrapper'; + +/** + * A response for a streaming request. It does not buffer the body: each read hands out the next + * chunk from the socket. The `text` and `json` methods read the remainder of the stream, so a + * caller uses either the reader or those methods, not both. + */ +export default class NodeStreamingResponse implements platform.Response { + headers: platform.Headers; + + status: number; + + statusText: string; + + body: platform.ResponseBody; + + private _iterator: AsyncIterableIterator; + + constructor(res: http.IncomingMessage) { + this.headers = new HeaderWrapper(res.headers); + this.status = res.statusCode ?? 0; + this.statusText = res.statusMessage ?? ''; + // The async iterator hands out one chunk per read. It resolves done when the server ends the + // stream, and it rejects when the connection drops or the request is destroyed. + const iterator = res[Symbol.asyncIterator](); + this._iterator = iterator; + this.body = { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value }; + }, + }), + }; + } + + async text(): Promise { + const chunks: Uint8Array[] = []; + for await (const chunk of this._iterator) { + chunks.push(chunk); + } + return Buffer.concat(chunks).toString(); + } + + async json(): Promise { + return JSON.parse(await this.text()); + } +} diff --git a/packages/sdk/server-node/__tests__/platform/HeaderWrapper.test.ts b/packages/sdk/server-node/__tests__/platform/HeaderWrapper.test.ts index bf1cfd0b50..680cefc3c5 100644 --- a/packages/sdk/server-node/__tests__/platform/HeaderWrapper.test.ts +++ b/packages/sdk/server-node/__tests__/platform/HeaderWrapper.test.ts @@ -50,4 +50,16 @@ describe('given header values', () => { } expect(values).toEqual(['anything', 'some-value', 'a, b']); }); + + it('iterates each header with the value before the key', () => { + const collected: [string, string][] = []; + wrapper.forEach((value, key) => { + collected.push([value, key]); + }); + expect(collected).toEqual([ + ['anything', 'accept'], + ['some-value', 'some-header'], + ['a, b', 'some-array'], + ]); + }); }); diff --git a/packages/sdk/server-node/__tests__/platform/NodeRequests.streaming.test.ts b/packages/sdk/server-node/__tests__/platform/NodeRequests.streaming.test.ts new file mode 100644 index 0000000000..fa2633883c --- /dev/null +++ b/packages/sdk/server-node/__tests__/platform/NodeRequests.streaming.test.ts @@ -0,0 +1,253 @@ +import * as http from 'http'; +import * as https from 'https'; +import { + AsyncQueue, + TestHttpHandlers, + TestHttpServer, + TestHttpServers, +} from 'launchdarkly-js-test-helpers'; + +import NodeRequests from '../../src/platform/NodeRequests'; + +describe('given a running HTTP server', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.start(); + }); + + afterEach(async () => { + jest.restoreAllMocks(); + await server.closeAndWait(); + }); + + it('forwards a streaming request and exposes the status, headers, and body chunks', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests(); + const res = await requests.fetch(`${server.url}/stream`, { + method: 'REPORT', + headers: { authorization: 'sdk-key' }, + body: '{"kind":"user"}', + streaming: true, + }); + + expect(res.status).toEqual(200); + const collected: Record = {}; + res.headers.forEach?.((value, key) => { + collected[key] = value; + }); + expect(collected['content-type']).toEqual('text/event-stream'); + + const reader = res.body?.getReader(); + const first = await reader?.read(); + expect(first?.done).toBe(false); + expect(Buffer.from(first?.value ?? []).toString()).toEqual('first'); + + const received = await server.nextRequest(); + expect(received.method.toUpperCase()).toEqual('REPORT'); + expect(received.headers.authorization).toEqual('sdk-key'); + expect(received.body).toEqual('{"kind":"user"}'); + }); + + it('does not request compressed content for a streaming request', async () => { + const chunks = new AsyncQueue(); + server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); + + const requests = new NodeRequests(); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + const received = await server.nextRequest(); + expect(received.headers['accept-encoding']).toBeUndefined(); + }); + + it('does not follow redirects for a streaming request', async () => { + server.byDefault(TestHttpHandlers.respond(301, { location: `${server.url}/other` })); + + const requests = new NodeRequests(); + const res = await requests.fetch(server.url, { method: 'GET', streaming: true }); + + expect(res.status).toEqual(301); + expect(server.requestCount()).toEqual(1); + }); + + it('stops the stream when the signal aborts', async () => { + const chunks = new AsyncQueue(); + chunks.add('first'); + server.byDefault(TestHttpHandlers.chunkedStream(200, {}, chunks)); + + const controller = new AbortController(); + const requests = new NodeRequests(); + const res = await requests.fetch(server.url, { + method: 'GET', + streaming: true, + signal: controller.signal, + }); + const reader = res.body?.getReader(); + await reader?.read(); + const pending = reader?.read(); + controller.abort(); + await expect(pending).rejects.toThrow(); + }); + + it('rejects a streaming request when the signal is already aborted', async () => { + server.byDefault(TestHttpHandlers.respond(200)); + const controller = new AbortController(); + controller.abort(); + + const requests = new NodeRequests(); + await expect( + requests.fetch(server.url, { method: 'GET', streaming: true, signal: controller.signal }), + ).rejects.toThrow(); + }); + + it('uses the supplied agent for a streaming request', async () => { + server.byDefault(TestHttpHandlers.respond(200)); + const agent = new http.Agent({ keepAlive: false }); + // addRequest exists at runtime but is not part of the public http.Agent type. + // @ts-ignore + const addRequestSpy = jest.spyOn(agent, 'addRequest'); + + const requests = new NodeRequests(undefined, undefined, agent); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + expect(addRequestSpy).toHaveBeenCalled(); + }); + + it('does not pass the CA to the per-request options of an http streaming request', async () => { + const requestSpy = jest.spyOn(http, 'request'); + + const requests = new NodeRequests({ ca: 'unused-ca' }); + // The https agent built from the TLS options is rejected for an http URL, so the request + // fails after http.request has already received its options. + await requests.fetch(server.url, { method: 'GET', streaming: true }).catch(() => undefined); + + expect(requestSpy).toHaveBeenCalledTimes(1); + expect(requestSpy.mock.calls[0][1]).not.toHaveProperty('ca'); + }); + + it('streams SSE events through createEventSource', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: hello\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests(); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('hello'); + } finally { + es.close(); + } + }); +}); + +describe('given a running HTTPS server with a self-signed certificate', () => { + let server: TestHttpServer; + + beforeEach(async () => { + server = await TestHttpServers.startSecure(); + server.byDefault(TestHttpHandlers.respond(200)); + }); + + afterEach(async () => { + jest.restoreAllMocks(); + await server.closeAndWait(); + }); + + it('connects a streaming request when the CA is supplied in the TLS options', async () => { + const requests = new NodeRequests({ ca: server.certificate }); + const res = await requests.fetch(server.url, { method: 'GET', streaming: true }); + expect(res.status).toEqual(200); + }); + + it('passes the CA to the per-request options of an https streaming request', async () => { + const requestSpy = jest.spyOn(https, 'request'); + + const requests = new NodeRequests({ ca: server.certificate }); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + expect(requestSpy).toHaveBeenCalledTimes(1); + expect(requestSpy.mock.calls[0][1]).toEqual( + expect.objectContaining({ ca: server.certificate }), + ); + }); + + it('rejects a streaming request when no CA is supplied', async () => { + const requests = new NodeRequests(); + await expect(requests.fetch(server.url, { method: 'GET', streaming: true })).rejects.toThrow(); + }); + + it('fails the stream against an untrusted certificate by default', async () => { + const requests = new NodeRequests(); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const errors = new AsyncQueue(); + es.addEventListener('error', (event) => errors.add(event)); + expect(await errors.take()).toBeDefined(); + } finally { + es.close(); + } + }); + + it('streams over TLS when the CA is supplied in the TLS options', async () => { + const chunks = new AsyncQueue(); + chunks.add('data: secure\n\n'); + server.byDefault( + TestHttpHandlers.chunkedStream(200, { 'content-type': 'text/event-stream' }, chunks), + ); + + const requests = new NodeRequests({ ca: server.certificate }); + const es = requests.createEventSource(`${server.url}/stream`, { + headers: {}, + initialRetryDelayMillis: 100, + readTimeoutMillis: 5000, + retryResetIntervalMillis: 30_000, + errorFilter: () => false, + }); + try { + const messages = new AsyncQueue<{ data?: string }>(); + es.addEventListener('message', (event) => messages.add(event ?? {})); + const message = await messages.take(); + expect(message.data).toEqual('secure'); + } finally { + es.close(); + } + }); + + it('does not pass the CA to the per-request options when a proxyAgent is supplied', async () => { + const requestSpy = jest.spyOn(https, 'request'); + const logger = { warn: jest.fn(), error: jest.fn(), info: jest.fn(), debug: jest.fn() }; + + const requests = new NodeRequests( + { ca: server.certificate }, + undefined, + new https.Agent({ ca: server.certificate }), + logger, + ); + await requests.fetch(server.url, { method: 'GET', streaming: true }); + + expect(requestSpy).toHaveBeenCalledTimes(1); + expect(requestSpy.mock.calls[0][1]).not.toHaveProperty('ca'); + }); +}); diff --git a/packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts b/packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts new file mode 100644 index 0000000000..c7ec88fc08 --- /dev/null +++ b/packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts @@ -0,0 +1,80 @@ +import * as http from 'http'; +import { Readable } from 'stream'; + +import NodeStreamingResponse from '../../src/platform/NodeStreamingResponse'; + +function makeIncomingMessage( + chunks: Buffer[], + options: { + headers?: http.IncomingHttpHeaders; + statusCode?: number; + statusMessage?: string; + } = {}, +): http.IncomingMessage { + const stream = Readable.from(chunks) as unknown as http.IncomingMessage; + stream.headers = options.headers ?? {}; + stream.statusCode = 'statusCode' in options ? options.statusCode : 200; + if (options.statusMessage !== undefined) { + stream.statusMessage = options.statusMessage; + } + return stream; +} + +it('exposes the status code and status message', () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([], { statusCode: 301, statusMessage: 'Moved Permanently' }), + ); + expect(res.status).toBe(301); + expect(res.statusText).toBe('Moved Permanently'); +}); + +it('defaults the status to 0 and the status message to an empty string when absent', () => { + const res = new NodeStreamingResponse(makeIncomingMessage([], { statusCode: undefined })); + expect(res.status).toBe(0); + expect(res.statusText).toBe(''); +}); + +it('wraps the response headers in a HeaderWrapper', () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([], { headers: { 'content-type': 'text/event-stream' } }), + ); + expect(res.headers.get('content-type')).toBe('text/event-stream'); +}); + +it('hands out one chunk per read and then reports done', async () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([Buffer.from('first'), Buffer.from('second')]), + ); + const reader = res.body.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(Buffer.from(first.value ?? []).toString()).toBe('first'); + const second = await reader.read(); + expect(Buffer.from(second.value ?? []).toString()).toBe('second'); + const end = await reader.read(); + expect(end.done).toBe(true); +}); + +it('reads the remaining stream as text', async () => { + const res = new NodeStreamingResponse( + makeIncomingMessage([Buffer.from('hello '), Buffer.from('world')]), + ); + await expect(res.text()).resolves.toBe('hello world'); +}); + +it('parses the remaining stream as JSON', async () => { + const res = new NodeStreamingResponse(makeIncomingMessage([Buffer.from('{"a":1}')])); + await expect(res.json()).resolves.toEqual({ a: 1 }); +}); + +it('rejects a pending read when the stream errors', async () => { + const erroring = new Readable({ + read() { + this.destroy(new Error('boom')); + }, + }) as unknown as http.IncomingMessage; + erroring.headers = {}; + erroring.statusCode = 200; + const res = new NodeStreamingResponse(erroring); + await expect(res.body.getReader().read()).rejects.toThrow('boom'); +}); diff --git a/packages/sdk/server-node/src/platform/HeaderWrapper.ts b/packages/sdk/server-node/src/platform/HeaderWrapper.ts index dee99edc92..fe7781baf0 100644 --- a/packages/sdk/server-node/src/platform/HeaderWrapper.ts +++ b/packages/sdk/server-node/src/platform/HeaderWrapper.ts @@ -52,6 +52,17 @@ export default class HeaderWrapper implements platform.Headers { } } + /** + * Executes the callback once for each header, with the value first. The order matches the + * fetch `Headers.forEach` signature. Multi-value headers are joined with a comma, and + * headers without a value are skipped, like `entries`. + */ + forEach(callback: (value: string, key: string) => void): void { + for (const [key, value] of this.entries()) { + callback(value, key); + } + } + has(name: string): boolean { return Object.prototype.hasOwnProperty.call(this._headers, name); } diff --git a/packages/sdk/server-node/src/platform/NodeEventSource.ts b/packages/sdk/server-node/src/platform/NodeEventSource.ts deleted file mode 100644 index ed6db34849..0000000000 --- a/packages/sdk/server-node/src/platform/NodeEventSource.ts +++ /dev/null @@ -1,146 +0,0 @@ -import * as http from 'http'; -import * as https from 'https'; - -import { - createEventSource, - type FetchLike, - type FetchLikeOptions, - type FetchLikeResponse, -} from '@launchdarkly/eventsource'; -import type { LDTLSOptions, platform } from '@launchdarkly/js-server-sdk-common'; - -/** - * The TLS options that are copied onto each `https` request. The names match the options of - * `https.request()`. - */ -const TLS_OPTION_NAMES = [ - 'pfx', - 'key', - 'passphrase', - 'cert', - 'ca', - 'ciphers', - 'rejectUnauthorized', - 'secureProtocol', - 'servername', - 'checkServerIdentity', -] as const; - -function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { - const merged: Record = {}; - if (!tlsOptions) { - return merged; - } - const bag = tlsOptions as unknown as Record; - TLS_OPTION_NAMES.forEach((name) => { - if (bag[name] !== undefined) { - merged[name] = bag[name]; - } - }); - return merged; -} - -function wrapResponse(res: http.IncomingMessage): FetchLikeResponse { - // The async iterator hands out one chunk per read. It resolves done when the server ends the - // stream, and it rejects when the connection drops or the request is destroyed. - const iterator = res[Symbol.asyncIterator](); - return { - status: res.statusCode ?? 0, - statusText: res.statusMessage ?? '', - headers: { - forEach(callback: (value: string, key: string) => void): void { - Object.entries(res.headers).forEach(([key, value]) => { - if (value === undefined) { - return; - } - callback(Array.isArray(value) ? value.join(', ') : value, key); - }); - }, - }, - body: { - getReader: () => ({ - read: async () => { - const next = await iterator.next(); - if (next.done) { - return { done: true }; - } - return { done: false, value: next.value as Uint8Array }; - }, - }), - }, - }; -} - -/** - * Creates a `fetch()`-shaped function over `node:http`/`node:https` for the SSE stream. - * - * The returned function bakes in the connection configuration: the agent (which can carry proxy - * or TLS setup) and the TLS parameters for `https` URLs. It never follows a redirect: a redirect - * status resolves like any other non-200 response, and the caller decides whether to retry the - * original URL. It applies no read or socket timeout; the caller owns the read timeout and - * cancels through the request's `AbortSignal`. - */ -export function createNodeFetch( - agent?: https.Agent | http.Agent, - tlsOptions?: LDTLSOptions, -): FetchLike { - const tlsParams = tlsRequestOptions(tlsOptions); - return async (url: string, init: FetchLikeOptions): Promise => { - const isSecure = url.startsWith('https://'); - const impl = isSecure ? https : http; - const requestOptions: https.RequestOptions & Record = { - method: init.method, - headers: init.headers, - agent, - }; - if (isSecure) { - Object.assign(requestOptions, tlsParams); - } - return new Promise((resolve, reject) => { - const req = impl.request(url, requestOptions, (res) => resolve(wrapResponse(res))); - // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. - req.setNoDelay(true); - const { signal } = init; - if (signal) { - const abort = () => req.destroy(new Error('The stream request was aborted')); - if (signal.aborted) { - abort(); - } else { - signal.addEventListener('abort', abort, { once: true }); - } - } - // stays attached after resolve, so a later socket error becomes a harmless no-op reject - // instead of an unhandled 'error' event that would crash the process - req.on('error', reject); - if (init.body !== undefined) { - req.write(init.body); - } - req.end(); - }); - }; -} - -export type NodeEventSourceFactory = ( - url: string, - eventSourceInitDict: platform.EventSourceInitDict, -) => platform.EventSource; - -/** - * Creates the event source factory used by `NodeRequests.createEventSource`. - * - * The factory bakes in the connection configuration (agent and TLS parameters, through the fetch - * adapter above) and the SDK retry policy (backoff cap and jitter). - */ -export default function createNodeEventSourceFactory( - agent?: https.Agent | http.Agent, - tlsOptions?: LDTLSOptions, -): NodeEventSourceFactory { - const fetch = createNodeFetch(agent, tlsOptions); - return (url, eventSourceInitDict) => - createEventSource(url, { - ...eventSourceInitDict, - maxBackoffMillis: 30 * 1000, - jitterRatio: 0.5, - fetch, - }); -} diff --git a/packages/sdk/server-node/src/platform/NodeRequests.ts b/packages/sdk/server-node/src/platform/NodeRequests.ts index b210e6443a..2b89b58fa2 100644 --- a/packages/sdk/server-node/src/platform/NodeRequests.ts +++ b/packages/sdk/server-node/src/platform/NodeRequests.ts @@ -5,6 +5,7 @@ import { format as formatUrl } from 'url'; import { promisify } from 'util'; import * as zlib from 'zlib'; +import { createEventSource } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, LDLogger, @@ -13,11 +14,42 @@ import { platform, } from '@launchdarkly/js-server-sdk-common'; -import createNodeEventSourceFactory, { NodeEventSourceFactory } from './NodeEventSource'; import NodeResponse from './NodeResponse'; +import NodeStreamingResponse from './NodeStreamingResponse'; const gzip = promisify(zlib.gzip); +/** + * The TLS options that are copied onto each streaming `https` request. The names match the + * options of `https.request()`. + */ +const TLS_OPTION_NAMES = [ + 'pfx', + 'key', + 'passphrase', + 'cert', + 'ca', + 'ciphers', + 'rejectUnauthorized', + 'secureProtocol', + 'servername', + 'checkServerIdentity', +] as const; + +function tlsRequestOptions(tlsOptions?: LDTLSOptions): Record { + const merged: Record = {}; + if (!tlsOptions) { + return merged; + } + const bag = tlsOptions as unknown as Record; + TLS_OPTION_NAMES.forEach((name) => { + if (bag[name] !== undefined) { + merged[name] = bag[name]; + } + }); + return merged; +} + function processTlsOptions(tlsOptions: LDTLSOptions): https.AgentOptions { const options: https.AgentOptions & { [index: string]: any } = { ca: tlsOptions.ca, @@ -120,9 +152,7 @@ function resolveAgent( export default class NodeRequests implements platform.Requests { private _agent: https.Agent | http.Agent | undefined; - private _tlsOptions: LDTLSOptions | undefined; - - private _createEventSource: NodeEventSourceFactory; + private _tlsParams: Record; private _hasProxy: boolean = false; @@ -138,10 +168,9 @@ export default class NodeRequests implements platform.Requests { enableEventCompression?: boolean, ) { this._agent = resolveAgent(tlsOptions, proxyOptions, proxyAgent, logger); - // The agent owns connection setup when the caller supplies one, so the TLS parameters are - // only forwarded when this class built the agent itself (or no agent exists). - this._tlsOptions = proxyAgent ? undefined : tlsOptions; - this._createEventSource = createNodeEventSourceFactory(this._agent, this._tlsOptions); + // The agent owns connection setup when the caller supplies one, so the per-request TLS + // parameters are only forwarded when this class built the agent itself (or no agent exists). + this._tlsParams = tlsRequestOptions(proxyAgent ? undefined : tlsOptions); // A caller-supplied proxyAgent is treated as a best-effort proxy signal: the SDK cannot // inspect an opaque agent to know whether it actually proxies (it could just as easily be a // certificate-only agent for mTLS). Reporting true is the better default here because @@ -157,6 +186,9 @@ export default class NodeRequests implements platform.Requests { } async fetch(url: string, options: platform.Options = {}): Promise { + if (options.streaming) { + return this._streamingFetch(url, options); + } const isSecure = url.startsWith('https://'); const impl = isSecure ? https : http; @@ -164,7 +196,7 @@ export default class NodeRequests implements platform.Requests { let bodyData: string | Buffer | undefined = options.body; // For get requests we are going to automatically support compressed responses. - // Note this does not affect SSE as the event source is not using this fetch implementation. + // Note this does not affect SSE as streaming requests take the branch above. if (options.method?.toLowerCase() === 'get') { headers['accept-encoding'] = 'gzip'; } @@ -209,11 +241,58 @@ export default class NodeRequests implements platform.Requests { }); } + /** + * The transport for a streaming request. It does not request compressed content, and it never + * follows a redirect. A redirect status resolves like any other non-200 response, and the + * caller decides whether to retry the original URL. It applies no read or socket timeout. The + * caller owns the read timeout and cancels through the abort signal. + */ + private _streamingFetch(url: string, options: platform.Options): Promise { + const isSecure = url.startsWith('https://'); + const impl = isSecure ? https : http; + const requestOptions: https.RequestOptions & Record = { + method: options.method, + headers: options.headers, + agent: this._agent, + }; + if (isSecure) { + Object.assign(requestOptions, this._tlsParams); + } + return new Promise((resolve, reject) => { + const req = impl.request(url, requestOptions, (res) => + resolve(new NodeStreamingResponse(res)), + ); + // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. + req.setNoDelay(true); + const { signal } = options; + if (signal) { + const abort = () => req.destroy(new Error('The stream request was aborted')); + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + // This listener stays attached after resolve. A later socket error then becomes a harmless + // no-op reject instead of an unhandled 'error' event that would crash the process + req.on('error', reject); + if (options.body !== undefined) { + req.write(options.body); + } + req.end(); + }); + } + createEventSource( url: string, eventSourceInitDict: platform.EventSourceInitDict, ): platform.EventSource { - return this._createEventSource(url, eventSourceInitDict); + return createEventSource(url, { + ...eventSourceInitDict, + maxBackoffMillis: 30 * 1000, + jitterRatio: 0.5, + fetch: (fetchUrl, init) => this.fetch(fetchUrl, { ...init, streaming: true }), + }); } getEventSourceCapabilities(): EventSourceCapabilities { diff --git a/packages/sdk/server-node/src/platform/NodeStreamingResponse.ts b/packages/sdk/server-node/src/platform/NodeStreamingResponse.ts new file mode 100644 index 0000000000..36fb992c76 --- /dev/null +++ b/packages/sdk/server-node/src/platform/NodeStreamingResponse.ts @@ -0,0 +1,55 @@ +import * as http from 'http'; + +import { platform } from '@launchdarkly/js-server-sdk-common'; + +import HeaderWrapper from './HeaderWrapper'; + +/** + * A response for a streaming request. It does not buffer the body: each read hands out the next + * chunk from the socket. The `text` and `json` methods read the remainder of the stream, so a + * caller uses either the reader or those methods, not both. + */ +export default class NodeStreamingResponse implements platform.Response { + headers: platform.Headers; + + status: number; + + statusText: string; + + body: platform.ResponseBody; + + private _iterator: AsyncIterableIterator; + + constructor(res: http.IncomingMessage) { + this.headers = new HeaderWrapper(res.headers); + this.status = res.statusCode ?? 0; + this.statusText = res.statusMessage ?? ''; + // The async iterator hands out one chunk per read. It resolves done when the server ends the + // stream, and it rejects when the connection drops or the request is destroyed. + const iterator = res[Symbol.asyncIterator](); + this._iterator = iterator; + this.body = { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value }; + }, + }), + }; + } + + async text(): Promise { + const chunks: Uint8Array[] = []; + for await (const chunk of this._iterator) { + chunks.push(chunk); + } + return Buffer.concat(chunks).toString(); + } + + async json(): Promise { + return JSON.parse(await this.text()); + } +} From 7ff36a7df68b9405f96bef1d3823f05436383650 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 9 Oct 2026 15:52:20 -0400 Subject: [PATCH 6/6] refactor: Use the common streaming response in the node-family SDKs --- .../ElectronStreamingResponse.test.ts | 80 ------------------- .../electron/src/platform/ElectronRequests.ts | 5 +- .../src/platform/ElectronStreamingResponse.ts | 55 ------------- .../platform/NodeStreamingResponse.test.ts | 80 ------------------- .../node-client/src/platform/NodeRequests.ts | 5 +- .../src/platform/NodeStreamingResponse.ts | 55 ------------- .../platform/NodeStreamingResponse.test.ts | 80 ------------------- .../server-node/src/platform/NodeRequests.ts | 4 +- .../src/platform/NodeStreamingResponse.ts | 55 ------------- 9 files changed, 6 insertions(+), 413 deletions(-) delete mode 100644 packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts delete mode 100644 packages/sdk/electron/src/platform/ElectronStreamingResponse.ts delete mode 100644 packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts delete mode 100644 packages/sdk/node-client/src/platform/NodeStreamingResponse.ts delete mode 100644 packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts delete mode 100644 packages/sdk/server-node/src/platform/NodeStreamingResponse.ts diff --git a/packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts b/packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts deleted file mode 100644 index 853b0c7002..0000000000 --- a/packages/sdk/electron/__tests__/platform/ElectronStreamingResponse.test.ts +++ /dev/null @@ -1,80 +0,0 @@ -import * as http from 'http'; -import { Readable } from 'stream'; - -import ElectronStreamingResponse from '../../src/platform/ElectronStreamingResponse'; - -function makeIncomingMessage( - chunks: Buffer[], - options: { - headers?: http.IncomingHttpHeaders; - statusCode?: number; - statusMessage?: string; - } = {}, -): http.IncomingMessage { - const stream = Readable.from(chunks) as unknown as http.IncomingMessage; - stream.headers = options.headers ?? {}; - stream.statusCode = 'statusCode' in options ? options.statusCode : 200; - if (options.statusMessage !== undefined) { - stream.statusMessage = options.statusMessage; - } - return stream; -} - -it('exposes the status code and status message', () => { - const res = new ElectronStreamingResponse( - makeIncomingMessage([], { statusCode: 301, statusMessage: 'Moved Permanently' }), - ); - expect(res.status).toBe(301); - expect(res.statusText).toBe('Moved Permanently'); -}); - -it('defaults the status to 0 and the status message to an empty string when absent', () => { - const res = new ElectronStreamingResponse(makeIncomingMessage([], { statusCode: undefined })); - expect(res.status).toBe(0); - expect(res.statusText).toBe(''); -}); - -it('wraps the response headers in a HeaderWrapper', () => { - const res = new ElectronStreamingResponse( - makeIncomingMessage([], { headers: { 'content-type': 'text/event-stream' } }), - ); - expect(res.headers.get('content-type')).toBe('text/event-stream'); -}); - -it('hands out one chunk per read and then reports done', async () => { - const res = new ElectronStreamingResponse( - makeIncomingMessage([Buffer.from('first'), Buffer.from('second')]), - ); - const reader = res.body.getReader(); - const first = await reader.read(); - expect(first.done).toBe(false); - expect(Buffer.from(first.value ?? []).toString()).toBe('first'); - const second = await reader.read(); - expect(Buffer.from(second.value ?? []).toString()).toBe('second'); - const end = await reader.read(); - expect(end.done).toBe(true); -}); - -it('reads the remaining stream as text', async () => { - const res = new ElectronStreamingResponse( - makeIncomingMessage([Buffer.from('hello '), Buffer.from('world')]), - ); - await expect(res.text()).resolves.toBe('hello world'); -}); - -it('parses the remaining stream as JSON', async () => { - const res = new ElectronStreamingResponse(makeIncomingMessage([Buffer.from('{"a":1}')])); - await expect(res.json()).resolves.toEqual({ a: 1 }); -}); - -it('rejects a pending read when the stream errors', async () => { - const erroring = new Readable({ - read() { - this.destroy(new Error('boom')); - }, - }) as unknown as http.IncomingMessage; - erroring.headers = {}; - erroring.statusCode = 200; - const res = new ElectronStreamingResponse(erroring); - await expect(res.body.getReader().read()).rejects.toThrow('boom'); -}); diff --git a/packages/sdk/electron/src/platform/ElectronRequests.ts b/packages/sdk/electron/src/platform/ElectronRequests.ts index f9bfe145fd..2e05928c58 100644 --- a/packages/sdk/electron/src/platform/ElectronRequests.ts +++ b/packages/sdk/electron/src/platform/ElectronRequests.ts @@ -4,10 +4,9 @@ import { promisify } from 'util'; import * as zlib from 'zlib'; import { createEventSource } from '@launchdarkly/eventsource'; -import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; +import { EventSourceCapabilities, internal, platform } from '@launchdarkly/js-client-sdk-common'; import ElectronResponse from './ElectronResponse'; -import ElectronStreamingResponse from './ElectronStreamingResponse'; const gzip = promisify(zlib.gzip); @@ -86,7 +85,7 @@ export default class ElectronRequests implements platform.Requests { }; return new Promise((resolve, reject) => { const req = impl.request(url, requestOptions, (res) => - resolve(new ElectronStreamingResponse(res)), + resolve(internal.createStreamingResponse(res)), ); // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. req.setNoDelay(true); diff --git a/packages/sdk/electron/src/platform/ElectronStreamingResponse.ts b/packages/sdk/electron/src/platform/ElectronStreamingResponse.ts deleted file mode 100644 index 656f6c7b75..0000000000 --- a/packages/sdk/electron/src/platform/ElectronStreamingResponse.ts +++ /dev/null @@ -1,55 +0,0 @@ -import * as http from 'http'; - -import { platform } from '@launchdarkly/js-client-sdk-common'; - -import HeaderWrapper from './HeaderWrapper'; - -/** - * A response for a streaming request. It does not buffer the body: each read hands out the next - * chunk from the socket. The `text` and `json` methods read the remainder of the stream, so a - * caller uses either the reader or those methods, not both. - */ -export default class ElectronStreamingResponse implements platform.Response { - headers: platform.Headers; - - status: number; - - statusText: string; - - body: platform.ResponseBody; - - private _iterator: AsyncIterableIterator; - - constructor(res: http.IncomingMessage) { - this.headers = new HeaderWrapper(res.headers); - this.status = res.statusCode ?? 0; - this.statusText = res.statusMessage ?? ''; - // The async iterator hands out one chunk per read. It resolves done when the server ends the - // stream, and it rejects when the connection drops or the request is destroyed. - const iterator = res[Symbol.asyncIterator](); - this._iterator = iterator; - this.body = { - getReader: () => ({ - read: async () => { - const next = await iterator.next(); - if (next.done) { - return { done: true }; - } - return { done: false, value: next.value }; - }, - }), - }; - } - - async text(): Promise { - const chunks: Uint8Array[] = []; - for await (const chunk of this._iterator) { - chunks.push(chunk); - } - return Buffer.concat(chunks).toString(); - } - - async json(): Promise { - return JSON.parse(await this.text()); - } -} diff --git a/packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts b/packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts deleted file mode 100644 index c7ec88fc08..0000000000 --- a/packages/sdk/node-client/__tests__/platform/NodeStreamingResponse.test.ts +++ /dev/null @@ -1,80 +0,0 @@ -import * as http from 'http'; -import { Readable } from 'stream'; - -import NodeStreamingResponse from '../../src/platform/NodeStreamingResponse'; - -function makeIncomingMessage( - chunks: Buffer[], - options: { - headers?: http.IncomingHttpHeaders; - statusCode?: number; - statusMessage?: string; - } = {}, -): http.IncomingMessage { - const stream = Readable.from(chunks) as unknown as http.IncomingMessage; - stream.headers = options.headers ?? {}; - stream.statusCode = 'statusCode' in options ? options.statusCode : 200; - if (options.statusMessage !== undefined) { - stream.statusMessage = options.statusMessage; - } - return stream; -} - -it('exposes the status code and status message', () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([], { statusCode: 301, statusMessage: 'Moved Permanently' }), - ); - expect(res.status).toBe(301); - expect(res.statusText).toBe('Moved Permanently'); -}); - -it('defaults the status to 0 and the status message to an empty string when absent', () => { - const res = new NodeStreamingResponse(makeIncomingMessage([], { statusCode: undefined })); - expect(res.status).toBe(0); - expect(res.statusText).toBe(''); -}); - -it('wraps the response headers in a HeaderWrapper', () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([], { headers: { 'content-type': 'text/event-stream' } }), - ); - expect(res.headers.get('content-type')).toBe('text/event-stream'); -}); - -it('hands out one chunk per read and then reports done', async () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([Buffer.from('first'), Buffer.from('second')]), - ); - const reader = res.body.getReader(); - const first = await reader.read(); - expect(first.done).toBe(false); - expect(Buffer.from(first.value ?? []).toString()).toBe('first'); - const second = await reader.read(); - expect(Buffer.from(second.value ?? []).toString()).toBe('second'); - const end = await reader.read(); - expect(end.done).toBe(true); -}); - -it('reads the remaining stream as text', async () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([Buffer.from('hello '), Buffer.from('world')]), - ); - await expect(res.text()).resolves.toBe('hello world'); -}); - -it('parses the remaining stream as JSON', async () => { - const res = new NodeStreamingResponse(makeIncomingMessage([Buffer.from('{"a":1}')])); - await expect(res.json()).resolves.toEqual({ a: 1 }); -}); - -it('rejects a pending read when the stream errors', async () => { - const erroring = new Readable({ - read() { - this.destroy(new Error('boom')); - }, - }) as unknown as http.IncomingMessage; - erroring.headers = {}; - erroring.statusCode = 200; - const res = new NodeStreamingResponse(erroring); - await expect(res.body.getReader().read()).rejects.toThrow('boom'); -}); diff --git a/packages/sdk/node-client/src/platform/NodeRequests.ts b/packages/sdk/node-client/src/platform/NodeRequests.ts index 9825425685..d7cefd269f 100644 --- a/packages/sdk/node-client/src/platform/NodeRequests.ts +++ b/packages/sdk/node-client/src/platform/NodeRequests.ts @@ -4,11 +4,10 @@ import { promisify } from 'util'; import * as zlib from 'zlib'; import { createEventSource } from '@launchdarkly/eventsource'; -import { EventSourceCapabilities, platform } from '@launchdarkly/js-client-sdk-common'; +import { EventSourceCapabilities, internal, platform } from '@launchdarkly/js-client-sdk-common'; import type { LDTLSOptions } from '../NodeOptions'; import NodeResponse from './NodeResponse'; -import NodeStreamingResponse from './NodeStreamingResponse'; const gzip = promisify(zlib.gzip); @@ -152,7 +151,7 @@ export default class NodeRequests implements platform.Requests { } return new Promise((resolve, reject) => { const req = impl.request(url, requestOptions, (res) => - resolve(new NodeStreamingResponse(res)), + resolve(internal.createStreamingResponse(res)), ); // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. req.setNoDelay(true); diff --git a/packages/sdk/node-client/src/platform/NodeStreamingResponse.ts b/packages/sdk/node-client/src/platform/NodeStreamingResponse.ts deleted file mode 100644 index 876855c59a..0000000000 --- a/packages/sdk/node-client/src/platform/NodeStreamingResponse.ts +++ /dev/null @@ -1,55 +0,0 @@ -import * as http from 'http'; - -import { platform } from '@launchdarkly/js-client-sdk-common'; - -import HeaderWrapper from './HeaderWrapper'; - -/** - * A response for a streaming request. It does not buffer the body: each read hands out the next - * chunk from the socket. The `text` and `json` methods read the remainder of the stream, so a - * caller uses either the reader or those methods, not both. - */ -export default class NodeStreamingResponse implements platform.Response { - headers: platform.Headers; - - status: number; - - statusText: string; - - body: platform.ResponseBody; - - private _iterator: AsyncIterableIterator; - - constructor(res: http.IncomingMessage) { - this.headers = new HeaderWrapper(res.headers); - this.status = res.statusCode ?? 0; - this.statusText = res.statusMessage ?? ''; - // The async iterator hands out one chunk per read. It resolves done when the server ends the - // stream, and it rejects when the connection drops or the request is destroyed. - const iterator = res[Symbol.asyncIterator](); - this._iterator = iterator; - this.body = { - getReader: () => ({ - read: async () => { - const next = await iterator.next(); - if (next.done) { - return { done: true }; - } - return { done: false, value: next.value }; - }, - }), - }; - } - - async text(): Promise { - const chunks: Uint8Array[] = []; - for await (const chunk of this._iterator) { - chunks.push(chunk); - } - return Buffer.concat(chunks).toString(); - } - - async json(): Promise { - return JSON.parse(await this.text()); - } -} diff --git a/packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts b/packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts deleted file mode 100644 index c7ec88fc08..0000000000 --- a/packages/sdk/server-node/__tests__/platform/NodeStreamingResponse.test.ts +++ /dev/null @@ -1,80 +0,0 @@ -import * as http from 'http'; -import { Readable } from 'stream'; - -import NodeStreamingResponse from '../../src/platform/NodeStreamingResponse'; - -function makeIncomingMessage( - chunks: Buffer[], - options: { - headers?: http.IncomingHttpHeaders; - statusCode?: number; - statusMessage?: string; - } = {}, -): http.IncomingMessage { - const stream = Readable.from(chunks) as unknown as http.IncomingMessage; - stream.headers = options.headers ?? {}; - stream.statusCode = 'statusCode' in options ? options.statusCode : 200; - if (options.statusMessage !== undefined) { - stream.statusMessage = options.statusMessage; - } - return stream; -} - -it('exposes the status code and status message', () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([], { statusCode: 301, statusMessage: 'Moved Permanently' }), - ); - expect(res.status).toBe(301); - expect(res.statusText).toBe('Moved Permanently'); -}); - -it('defaults the status to 0 and the status message to an empty string when absent', () => { - const res = new NodeStreamingResponse(makeIncomingMessage([], { statusCode: undefined })); - expect(res.status).toBe(0); - expect(res.statusText).toBe(''); -}); - -it('wraps the response headers in a HeaderWrapper', () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([], { headers: { 'content-type': 'text/event-stream' } }), - ); - expect(res.headers.get('content-type')).toBe('text/event-stream'); -}); - -it('hands out one chunk per read and then reports done', async () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([Buffer.from('first'), Buffer.from('second')]), - ); - const reader = res.body.getReader(); - const first = await reader.read(); - expect(first.done).toBe(false); - expect(Buffer.from(first.value ?? []).toString()).toBe('first'); - const second = await reader.read(); - expect(Buffer.from(second.value ?? []).toString()).toBe('second'); - const end = await reader.read(); - expect(end.done).toBe(true); -}); - -it('reads the remaining stream as text', async () => { - const res = new NodeStreamingResponse( - makeIncomingMessage([Buffer.from('hello '), Buffer.from('world')]), - ); - await expect(res.text()).resolves.toBe('hello world'); -}); - -it('parses the remaining stream as JSON', async () => { - const res = new NodeStreamingResponse(makeIncomingMessage([Buffer.from('{"a":1}')])); - await expect(res.json()).resolves.toEqual({ a: 1 }); -}); - -it('rejects a pending read when the stream errors', async () => { - const erroring = new Readable({ - read() { - this.destroy(new Error('boom')); - }, - }) as unknown as http.IncomingMessage; - erroring.headers = {}; - erroring.statusCode = 200; - const res = new NodeStreamingResponse(erroring); - await expect(res.body.getReader().read()).rejects.toThrow('boom'); -}); diff --git a/packages/sdk/server-node/src/platform/NodeRequests.ts b/packages/sdk/server-node/src/platform/NodeRequests.ts index 2b89b58fa2..45cde40c41 100644 --- a/packages/sdk/server-node/src/platform/NodeRequests.ts +++ b/packages/sdk/server-node/src/platform/NodeRequests.ts @@ -8,6 +8,7 @@ import * as zlib from 'zlib'; import { createEventSource } from '@launchdarkly/eventsource'; import { EventSourceCapabilities, + internal, LDLogger, LDProxyOptions, LDTLSOptions, @@ -15,7 +16,6 @@ import { } from '@launchdarkly/js-server-sdk-common'; import NodeResponse from './NodeResponse'; -import NodeStreamingResponse from './NodeStreamingResponse'; const gzip = promisify(zlib.gzip); @@ -260,7 +260,7 @@ export default class NodeRequests implements platform.Requests { } return new Promise((resolve, reject) => { const req = impl.request(url, requestOptions, (res) => - resolve(new NodeStreamingResponse(res)), + resolve(internal.createStreamingResponse(res)), ); // An SSE consumer wants each chunk as soon as it arrives; do not batch small writes. req.setNoDelay(true); diff --git a/packages/sdk/server-node/src/platform/NodeStreamingResponse.ts b/packages/sdk/server-node/src/platform/NodeStreamingResponse.ts deleted file mode 100644 index 36fb992c76..0000000000 --- a/packages/sdk/server-node/src/platform/NodeStreamingResponse.ts +++ /dev/null @@ -1,55 +0,0 @@ -import * as http from 'http'; - -import { platform } from '@launchdarkly/js-server-sdk-common'; - -import HeaderWrapper from './HeaderWrapper'; - -/** - * A response for a streaming request. It does not buffer the body: each read hands out the next - * chunk from the socket. The `text` and `json` methods read the remainder of the stream, so a - * caller uses either the reader or those methods, not both. - */ -export default class NodeStreamingResponse implements platform.Response { - headers: platform.Headers; - - status: number; - - statusText: string; - - body: platform.ResponseBody; - - private _iterator: AsyncIterableIterator; - - constructor(res: http.IncomingMessage) { - this.headers = new HeaderWrapper(res.headers); - this.status = res.statusCode ?? 0; - this.statusText = res.statusMessage ?? ''; - // The async iterator hands out one chunk per read. It resolves done when the server ends the - // stream, and it rejects when the connection drops or the request is destroyed. - const iterator = res[Symbol.asyncIterator](); - this._iterator = iterator; - this.body = { - getReader: () => ({ - read: async () => { - const next = await iterator.next(); - if (next.done) { - return { done: true }; - } - return { done: false, value: next.value }; - }, - }), - }; - } - - async text(): Promise { - const chunks: Uint8Array[] = []; - for await (const chunk of this._iterator) { - chunks.push(chunk); - } - return Buffer.concat(chunks).toString(); - } - - async json(): Promise { - return JSON.parse(await this.text()); - } -}