95 lines
2.6 KiB
TypeScript
95 lines
2.6 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import type { WSContext } from 'hono/ws';
|
|
import { toExtendedJson } from '@xo-cash/utils';
|
|
|
|
import { BaseStream, type StreamMessage } from './base-stream.ts';
|
|
|
|
/** WebSocket readyState value indicating an open socket. */
|
|
const OPEN = 1;
|
|
|
|
/** Adapts one WebSocket connection to the shared connection-stream contract. */
|
|
export class WSStream extends BaseStream {
|
|
readonly streaming = true;
|
|
readonly bidirectional = true;
|
|
|
|
/** Stable server-side identity used only for diagnostics. */
|
|
readonly id: string;
|
|
|
|
/** Observers notified when the socket closes. */
|
|
private closeCallbacks: Array<() => void> = [];
|
|
|
|
/** Guards against sends after local or remote closure. */
|
|
private closed = false;
|
|
|
|
/**
|
|
* @param ws - Minimal WebSocket surface required for send/close operations.
|
|
*/
|
|
constructor(private readonly ws: Pick<WSContext, 'send' | 'close' | 'readyState'>) {
|
|
super();
|
|
this.id = randomUUID();
|
|
}
|
|
|
|
/**
|
|
* Encode and send one Extended JSON message frame.
|
|
*
|
|
* @param message - Neutral response or event envelope to send.
|
|
*/
|
|
async send(message: StreamMessage): Promise<void> {
|
|
if (this.closed || this.ws.readyState !== OPEN) {
|
|
throw new Error('Cannot send to a closed WebSocket stream');
|
|
}
|
|
|
|
// WebSocket carries the complete neutral response or event envelope.
|
|
this.ws.send(toExtendedJson(message));
|
|
}
|
|
|
|
/** Close the socket from the server side and notify observers. */
|
|
close(): void {
|
|
if (this.closed) {
|
|
return;
|
|
}
|
|
|
|
this.closed = true;
|
|
this.ws.close();
|
|
this.emitClose();
|
|
}
|
|
|
|
/**
|
|
* Register a callback invoked when the socket closes.
|
|
*
|
|
* @param callback - Called immediately if the stream is already closed.
|
|
*/
|
|
onClose(callback: () => void): void {
|
|
if (this.closed) {
|
|
callback();
|
|
|
|
return;
|
|
}
|
|
|
|
this.closeCallbacks.push(callback);
|
|
}
|
|
|
|
/**
|
|
* Record closure reported by WebSocket callbacks without closing the socket
|
|
* again. This shares the same observer notification path as local closure.
|
|
*/
|
|
markClosed(): void {
|
|
if (this.closed) {
|
|
return;
|
|
}
|
|
|
|
this.closed = true;
|
|
this.emitClose();
|
|
}
|
|
|
|
/** Drain and invoke all registered close observers. */
|
|
private emitClose(): void {
|
|
const callbacks = this.closeCallbacks;
|
|
this.closeCallbacks = [];
|
|
|
|
for (const callback of callbacks) {
|
|
callback();
|
|
}
|
|
}
|
|
}
|