From 62553b2d647c0ead0ed629e929bb4ecdc44f6782 Mon Sep 17 00:00:00 2001 From: markstuart-oai <323302876+markstuart-oai@users.noreply.github.com> Date: Thu, 15 Oct 2026 04:24:34 +0000 Subject: [PATCH] refactor: share existing WebSocket send and open handling Castiron-Internal-PR: https://github.com/openai/openai-node-internal/pull/306 Castiron-Source-SHA: 68a9ec540ee28500d40b6258020e04da6c6010ff Castiron-Public-Base-SHA: 4ca7bd61c05effec8d2ace880a000047da37f1f3 --- src/internal/ws-connection.ts | 116 ++++++++++++++++++ src/resources/beta/responses/ws-base.ts | 99 ++++----------- src/resources/live/forks/ws-base.ts | 101 ++++----------- src/resources/live/sideband/ws-base.ts | 101 ++++----------- src/resources/live/ws-base.ts | 101 ++++----------- src/resources/responses/ws-base.ts | 99 ++++----------- .../lib/websocket-send-compatibility.test.ts | 100 +++++++++++++++ 7 files changed, 332 insertions(+), 385 deletions(-) create mode 100644 src/internal/ws-connection.ts create mode 100644 tests/lib/websocket-send-compatibility.test.ts diff --git a/src/internal/ws-connection.ts b/src/internal/ws-connection.ts new file mode 100644 index 0000000000..dc537733a1 --- /dev/null +++ b/src/internal/ws-connection.ts @@ -0,0 +1,116 @@ +import { OpenAIError } from '../core/error'; +import { ReadyState } from './ws-adapter'; +import type { WebSocketLike } from './ws-adapter'; +import { flattenRawData } from './ws'; +import type { RawWebSocketData, SendQueue } from './ws'; + +// Read the public socket on each access; callers may replace it through an accessor. +interface SocketOwner { + socket: WebSocketLike; +} +type OnError = (message: string, cause?: unknown) => void; + +/** Sends or queues an application event using the existing connection state. */ +export function sendWebSocketEvent( + owner: SocketOwner, + event: T, + queue: SendQueue, + isReconnecting: () => boolean, + onError: OnError, +): void { + if (!owner.socket) { + throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); + } + + if (isReconnecting() || owner.socket.readyState === ReadyState.CONNECTING) { + if (!queue.enqueue(event)) { + onError('send queue is full, message discarded'); + } + return; + } + if (owner.socket.readyState !== ReadyState.OPEN) { + onError('cannot send on a closed WebSocket'); + return; + } + try { + owner.socket.send(JSON.stringify(event)); + } catch (error) { + onError('could not send data', error); + } +} + +/** Sends or snapshots a raw frame using the same admission rules as application events. */ +export function sendRawWebSocketData( + owner: SocketOwner, + data: RawWebSocketData, + queue: SendQueue, + isReconnecting: () => boolean, + onError: OnError, +): void { + if (!owner.socket) { + throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); + } + + if (isReconnecting() || owner.socket.readyState === ReadyState.CONNECTING) { + if (!queue.enqueueRaw(data)) { + onError('send queue is full, message discarded'); + } + return; + } + if (owner.socket.readyState !== ReadyState.OPEN) { + onError('cannot send on a closed WebSocket'); + return; + } + try { + owner.socket.send(flattenRawData(data)); + } catch (error) { + onError('could not send data', error); + } +} + +/** Flushes queued frames with the caller's existing failed-send policy. */ +export function flushWebSocketSendQueue( + owner: SocketOwner, + queue: SendQueue, + onError: OnError, + options?: { requeueFailed?: boolean }, +): void { + if (!owner.socket) { + throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); + } + + try { + queue.flush((data) => owner.socket.send(flattenRawData(data)), options); + } catch (error) { + onError('could not send queued data', error); + } +} + +/** Resolves when the socket opens, or rejects if it errors or closes first. */ +export function waitForWebSocketOpen(socket: WebSocketLike): Promise { + // oxlint-disable-next-line promise/avoid-new -- Convert the existing socket event callbacks into the connection attempt's promise. + return new Promise((resolve, reject) => { + /* oxlint-disable eslint/no-use-before-define -- All mutually dependent callbacks are initialized before socket listeners are registered. */ + const cleanup = () => { + socket.off('open', onOpen); + socket.off('error', onError); + socket.off('close', onFail); + }; + /* oxlint-enable eslint/no-use-before-define */ + const onOpen = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const onFail = () => { + cleanup(); + reject(new Error('socket closed before open')); + }; + socket.once('open', onOpen); + socket.once('error', onError); + socket.once('close', onFail); + }); +} diff --git a/src/resources/beta/responses/ws-base.ts b/src/resources/beta/responses/ws-base.ts index 90f14429a0..4b0fec011e 100644 --- a/src/resources/beta/responses/ws-base.ts +++ b/src/resources/beta/responses/ws-base.ts @@ -1,6 +1,12 @@ // File generated from our OpenAPI spec by Castiron. See CONTRIBUTING.md for details. import { ResponsesEmitter, ResponsesStreamMessage, WebSocketError, buildURL } from './internal-base'; +import { + sendWebSocketEvent, + sendRawWebSocketData, + flushWebSocketSendQueue, + waitForWebSocketOpen, +} from '../../../internal/ws-connection'; import { InternalEventEmitter } from '../../../core/EventEmitter'; import { sleep } from '../../../internal/utils/sleep'; import { type WebSocketLike, ReadyState } from '../../../internal/ws-adapter'; @@ -10,7 +16,6 @@ import { recordWebSocketError, rawByteLength, type WebSocketStreamOptions, - flattenRawData, isRecoverableClose, type RawWebSocketData, type ReconnectingEvent, @@ -128,47 +133,23 @@ export abstract class ResponsesWSBase extends Res private _cancelRefresh: (() => void) | undefined; send(event: ResponsesAPI.BetaResponsesClientEvent) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueue(event)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(JSON.stringify(event)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendWebSocketEvent( + this, + event, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } sendRaw(data: RawWebSocketData) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueueRaw(data)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(flattenRawData(data)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendRawWebSocketData( + this, + data, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } close(props?: { code: number; reason: string }) { @@ -661,7 +642,7 @@ export abstract class ResponsesWSBase extends Res this.socket.once('close', resolve); }); - await this._awaitOpen(this.socket); + await waitForWebSocketOpen(this.socket); this._internalEvents._emit('socketSwap', oldSocket, this.socket); this._isReconnecting = false; @@ -698,44 +679,8 @@ export abstract class ResponsesWSBase extends Res this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); } - /** - * Resolves once the socket is open, rejects if it errors or closes first - */ - private _awaitOpen(socket: TSocket): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - socket.off('open', onOpen); - socket.off('error', onError); - socket.off('close', onFail); - }; - const onOpen = () => { - cleanup(); - resolve(); - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const onFail = () => { - cleanup(); - reject(new Error('socket closed before open')); - }; - socket.once('open', onOpen); - socket.once('error', onError); - socket.once('close', onFail); - }); - } - private _flushSendQueue(): void { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - try { - this._sendQueue.flush((data) => this.socket.send(flattenRawData(data))); - } catch (err) { - this._onError(null, 'could not send queued data', err); - } + flushWebSocketSendQueue(this, this._sendQueue, (message, cause) => this._onError(null, message, cause)); } /** diff --git a/src/resources/live/forks/ws-base.ts b/src/resources/live/forks/ws-base.ts index 02721045a4..68c9cd980a 100644 --- a/src/resources/live/forks/ws-base.ts +++ b/src/resources/live/forks/ws-base.ts @@ -1,6 +1,12 @@ // File generated from our OpenAPI spec by Castiron. See CONTRIBUTING.md for details. import { ForksEmitter, ForksStreamMessage, WebSocketError, buildURL } from './internal-base'; +import { + sendWebSocketEvent, + sendRawWebSocketData, + flushWebSocketSendQueue, + waitForWebSocketOpen, +} from '../../../internal/ws-connection'; import { InternalEventEmitter } from '../../../core/EventEmitter'; import { sleep } from '../../../internal/utils/sleep'; import { type WebSocketLike, ReadyState } from '../../../internal/ws-adapter'; @@ -8,7 +14,6 @@ import { SendQueue, getMaxBufferedEvents, type WebSocketStreamOptions, - flattenRawData, isRecoverableClose, type RawWebSocketData, type ReconnectingEvent, @@ -107,47 +112,23 @@ export abstract class ForksWSBase extends ForksEm protected abstract _createSocket(url: URL, authHeaders: Record): TSocket; send(event: ForksAPI.ForkClientEvent) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueue(event)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(JSON.stringify(event)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendWebSocketEvent( + this, + event, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } sendRaw(data: RawWebSocketData) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueueRaw(data)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(flattenRawData(data)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendRawWebSocketData( + this, + data, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } close(props?: { code: number; reason: string }) { @@ -609,7 +590,7 @@ export abstract class ForksWSBase extends ForksEm this.socket.once('close', resolve); }); - await this._awaitOpen(this.socket); + await waitForWebSocketOpen(this.socket); this._internalEvents._emit('socketSwap', oldSocket, this.socket); this._isReconnecting = false; @@ -637,44 +618,10 @@ export abstract class ForksWSBase extends ForksEm this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); } - /** - * Resolves once the socket is open, rejects if it errors or closes first - */ - private _awaitOpen(socket: TSocket): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - socket.off('open', onOpen); - socket.off('error', onError); - socket.off('close', onFail); - }; - const onOpen = () => { - cleanup(); - resolve(); - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const onFail = () => { - cleanup(); - reject(new Error('socket closed before open')); - }; - socket.once('open', onOpen); - socket.once('error', onError); - socket.once('close', onFail); - }); - } - private _flushSendQueue(): void { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - try { - this._sendQueue.flush((data) => this.socket.send(flattenRawData(data)), { requeueFailed: false }); - } catch (err) { - this._onError(null, 'could not send queued data', err); - } + flushWebSocketSendQueue(this, this._sendQueue, (message, cause) => this._onError(null, message, cause), { + requeueFailed: false, + }); } /** diff --git a/src/resources/live/sideband/ws-base.ts b/src/resources/live/sideband/ws-base.ts index 7be76b5c7d..2b530be584 100644 --- a/src/resources/live/sideband/ws-base.ts +++ b/src/resources/live/sideband/ws-base.ts @@ -1,6 +1,12 @@ // File generated from our OpenAPI spec by Castiron. See CONTRIBUTING.md for details. import { SidebandEmitter, SidebandStreamMessage, WebSocketError, buildURL } from './internal-base'; +import { + sendWebSocketEvent, + sendRawWebSocketData, + flushWebSocketSendQueue, + waitForWebSocketOpen, +} from '../../../internal/ws-connection'; import { InternalEventEmitter } from '../../../core/EventEmitter'; import { sleep } from '../../../internal/utils/sleep'; import { type WebSocketLike, ReadyState } from '../../../internal/ws-adapter'; @@ -8,7 +14,6 @@ import { SendQueue, getMaxBufferedEvents, type WebSocketStreamOptions, - flattenRawData, isRecoverableClose, type RawWebSocketData, type ReconnectingEvent, @@ -113,47 +118,23 @@ export abstract class SidebandWSBase extends Side protected abstract _createSocket(url: URL, authHeaders: Record): TSocket; send(event: SidebandAPI.ConnectClientEvent) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueue(event)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(JSON.stringify(event)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendWebSocketEvent( + this, + event, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } sendRaw(data: RawWebSocketData) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueueRaw(data)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(flattenRawData(data)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendRawWebSocketData( + this, + data, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } close(props?: { code: number; reason: string }) { @@ -619,7 +600,7 @@ export abstract class SidebandWSBase extends Side this.socket.once('close', resolve); }); - await this._awaitOpen(this.socket); + await waitForWebSocketOpen(this.socket); this._internalEvents._emit('socketSwap', oldSocket, this.socket); this._isReconnecting = false; @@ -647,44 +628,10 @@ export abstract class SidebandWSBase extends Side this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); } - /** - * Resolves once the socket is open, rejects if it errors or closes first - */ - private _awaitOpen(socket: TSocket): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - socket.off('open', onOpen); - socket.off('error', onError); - socket.off('close', onFail); - }; - const onOpen = () => { - cleanup(); - resolve(); - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const onFail = () => { - cleanup(); - reject(new Error('socket closed before open')); - }; - socket.once('open', onOpen); - socket.once('error', onError); - socket.once('close', onFail); - }); - } - private _flushSendQueue(): void { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - try { - this._sendQueue.flush((data) => this.socket.send(flattenRawData(data)), { requeueFailed: false }); - } catch (err) { - this._onError(null, 'could not send queued data', err); - } + flushWebSocketSendQueue(this, this._sendQueue, (message, cause) => this._onError(null, message, cause), { + requeueFailed: false, + }); } /** diff --git a/src/resources/live/ws-base.ts b/src/resources/live/ws-base.ts index 467a569617..15e4126934 100644 --- a/src/resources/live/ws-base.ts +++ b/src/resources/live/ws-base.ts @@ -1,6 +1,12 @@ // File generated from our OpenAPI spec by Castiron. See CONTRIBUTING.md for details. import { LiveEmitter, LiveStreamMessage, WebSocketError, buildURL } from './internal-base'; +import { + sendWebSocketEvent, + sendRawWebSocketData, + flushWebSocketSendQueue, + waitForWebSocketOpen, +} from '../../internal/ws-connection'; import { InternalEventEmitter } from '../../core/EventEmitter'; import { sleep } from '../../internal/utils/sleep'; import { type WebSocketLike, ReadyState } from '../../internal/ws-adapter'; @@ -8,7 +14,6 @@ import { SendQueue, getMaxBufferedEvents, type WebSocketStreamOptions, - flattenRawData, isRecoverableClose, type RawWebSocketData, type ReconnectingEvent, @@ -103,47 +108,23 @@ export abstract class LiveWSBase extends LiveEmit protected abstract _createSocket(url: URL, authHeaders: Record): TSocket; send(event: LiveAPI.ClientEvent) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueue(event)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(JSON.stringify(event)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendWebSocketEvent( + this, + event, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } sendRaw(data: RawWebSocketData) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueueRaw(data)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(flattenRawData(data)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendRawWebSocketData( + this, + data, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } close(props?: { code: number; reason: string }) { @@ -605,7 +586,7 @@ export abstract class LiveWSBase extends LiveEmit this.socket.once('close', resolve); }); - await this._awaitOpen(this.socket); + await waitForWebSocketOpen(this.socket); this._internalEvents._emit('socketSwap', oldSocket, this.socket); this._isReconnecting = false; @@ -633,44 +614,10 @@ export abstract class LiveWSBase extends LiveEmit this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); } - /** - * Resolves once the socket is open, rejects if it errors or closes first - */ - private _awaitOpen(socket: TSocket): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - socket.off('open', onOpen); - socket.off('error', onError); - socket.off('close', onFail); - }; - const onOpen = () => { - cleanup(); - resolve(); - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const onFail = () => { - cleanup(); - reject(new Error('socket closed before open')); - }; - socket.once('open', onOpen); - socket.once('error', onError); - socket.once('close', onFail); - }); - } - private _flushSendQueue(): void { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - try { - this._sendQueue.flush((data) => this.socket.send(flattenRawData(data)), { requeueFailed: false }); - } catch (err) { - this._onError(null, 'could not send queued data', err); - } + flushWebSocketSendQueue(this, this._sendQueue, (message, cause) => this._onError(null, message, cause), { + requeueFailed: false, + }); } /** diff --git a/src/resources/responses/ws-base.ts b/src/resources/responses/ws-base.ts index 92c4933feb..de463c422d 100644 --- a/src/resources/responses/ws-base.ts +++ b/src/resources/responses/ws-base.ts @@ -1,6 +1,12 @@ // File generated from our OpenAPI spec by Castiron. See CONTRIBUTING.md for details. import { ResponsesEmitter, ResponsesStreamMessage, WebSocketError, buildURL } from './internal-base'; +import { + sendWebSocketEvent, + sendRawWebSocketData, + flushWebSocketSendQueue, + waitForWebSocketOpen, +} from '../../internal/ws-connection'; import { InternalEventEmitter } from '../../core/EventEmitter'; import { sleep } from '../../internal/utils/sleep'; import { type WebSocketLike, ReadyState } from '../../internal/ws-adapter'; @@ -10,7 +16,6 @@ import { recordWebSocketError, rawByteLength, type WebSocketStreamOptions, - flattenRawData, isRecoverableClose, type RawWebSocketData, type ReconnectingEvent, @@ -126,47 +131,23 @@ export abstract class ResponsesWSBase extends Res private _cancelRefresh: (() => void) | undefined; send(event: ResponsesAPI.ResponsesClientEvent) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueue(event)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(JSON.stringify(event)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendWebSocketEvent( + this, + event, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } sendRaw(data: RawWebSocketData) { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - if (this._isReconnecting || this.socket.readyState === ReadyState.CONNECTING) { - if (!this._sendQueue.enqueueRaw(data)) { - this._onError(null, 'send queue is full, message discarded', undefined); - } - return; - } - if (this.socket.readyState !== ReadyState.OPEN) { - this._onError(null, 'cannot send on a closed WebSocket', undefined); - return; - } - try { - this.socket.send(flattenRawData(data)); - } catch (err) { - this._onError(null, 'could not send data', err); - } + sendRawWebSocketData( + this, + data, + this._sendQueue, + () => this._isReconnecting, + (message, cause) => this._onError(null, message, cause), + ); } close(props?: { code: number; reason: string }) { @@ -659,7 +640,7 @@ export abstract class ResponsesWSBase extends Res this.socket.once('close', resolve); }); - await this._awaitOpen(this.socket); + await waitForWebSocketOpen(this.socket); this._internalEvents._emit('socketSwap', oldSocket, this.socket); this._isReconnecting = false; @@ -696,44 +677,8 @@ export abstract class ResponsesWSBase extends Res this._emitPermanentClose(closeCode, `reconnect failed after ${maxRetries} attempts`); } - /** - * Resolves once the socket is open, rejects if it errors or closes first - */ - private _awaitOpen(socket: TSocket): Promise { - return new Promise((resolve, reject) => { - const cleanup = () => { - socket.off('open', onOpen); - socket.off('error', onError); - socket.off('close', onFail); - }; - const onOpen = () => { - cleanup(); - resolve(); - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const onFail = () => { - cleanup(); - reject(new Error('socket closed before open')); - }; - socket.once('open', onOpen); - socket.once('error', onError); - socket.once('close', onFail); - }); - } - private _flushSendQueue(): void { - if (!this.socket) { - throw new OpenAIError('Internal error: failed to initialize socket. Please report this issue.'); - } - - try { - this._sendQueue.flush((data) => this.socket.send(flattenRawData(data))); - } catch (err) { - this._onError(null, 'could not send queued data', err); - } + flushWebSocketSendQueue(this, this._sendQueue, (message, cause) => this._onError(null, message, cause)); } /** diff --git a/tests/lib/websocket-send-compatibility.test.ts b/tests/lib/websocket-send-compatibility.test.ts new file mode 100644 index 0000000000..ec169a3fb9 --- /dev/null +++ b/tests/lib/websocket-send-compatibility.test.ts @@ -0,0 +1,100 @@ +import OpenAI from 'openai'; +import { afterEach, expect, vi } from 'vitest'; +import { onWebSocketEvent, setMockSocketReadyState, websocketVariants } from './helpers/websocket-variants'; + +afterEach(() => { + setMockSocketReadyState(1); + vi.useRealTimers(); +}); + +describe.each(websocketVariants)('$name send compatibility', ({ name, create }) => { + test('queues a send when reading the public socket starts a reconnect', async () => { + vi.useFakeTimers(); + const connection = create(new OpenAI({ apiKey: 'synthetic-key', baseURL: 'https://example.test/v1' }), { + reconnect: { onReconnecting: () => {}, initialDelay: 10, maxRetries: 1 }, + }); + const { socket } = connection; + const errors = vi.fn(); + const close = vi.fn(); + onWebSocketEvent(connection, 'error', errors); + onWebSocketEvent(connection, 'close', close); + let interrupted = false; + Object.defineProperty(connection, 'socket', { + configurable: true, + get() { + if (!interrupted) { + interrupted = true; + Object.defineProperty(socket.platformSocket, 'readyState', { value: 3, writable: true }); + socket.platformSocket.emit('close', 1006, Buffer.from('synthetic interruption')); + } + return socket; + }, + }); + try { + connection.sendRaw('queued during reconnect'); + expect(errors).not.toHaveBeenCalled(); + connection.close(); + await vi.runAllTimersAsync(); + expect(close).toHaveBeenCalledExactlyOnceWith(1000, 'OK', [ + { type: 'raw', data: 'queued during reconnect' }, + ]); + } finally { + connection.close(); + await vi.runAllTimersAsync(); + } + }); + + test('uses the current public socket after a ready-state accessor replaces it', () => { + const client = new OpenAI({ apiKey: 'synthetic-key', baseURL: 'https://example.test/v1' }); + const connection = create(client); + const replacement = create(client); + const originalSocket = connection.socket; + Object.defineProperty(originalSocket, 'readyState', { + configurable: true, + get() { + connection.socket = replacement.socket; + return 1; + }, + }); + try { + connection.sendRaw('synthetic message'); + expect(originalSocket.platformSocket.send).not.toHaveBeenCalled(); + expect(replacement.socket.platformSocket.send).toHaveBeenCalledExactlyOnceWith('synthetic message'); + } finally { + connection.socket = originalSocket; + connection.close(); + replacement.close(); + } + }); + + test('preserves which queued messages remain unsent after a failed write', () => { + setMockSocketReadyState(0); + const connection = create(new OpenAI({ apiKey: 'synthetic-key', baseURL: 'https://example.test/v1' })); + const error = new Error('synthetic write failure'); + const errors = vi.fn(); + const close = vi.fn(); + onWebSocketEvent(connection, 'error', errors); + onWebSocketEvent(connection, 'close', close); + connection.sendRaw('attempted'); + connection.sendRaw('never attempted'); + vi.spyOn(connection.socket.platformSocket, 'send').mockImplementationOnce(() => { + throw error; + }); + try { + Object.defineProperty(connection.socket.platformSocket, 'readyState', { value: 1, writable: true }); + connection.socket.platformSocket.emit('open'); + expect(errors).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ message: 'could not send queued data', cause: error }), + ); + connection.close(); + const pending = name.endsWith('Responses') ? ['attempted', 'never attempted'] : ['never attempted']; + expect(close).toHaveBeenCalledExactlyOnceWith( + 1000, + 'OK', + pending.map((data) => ({ type: 'raw', data })), + ); + } finally { + connection.close(); + } + }); +});