81 lines
2.3 KiB
TypeScript
81 lines
2.3 KiB
TypeScript
import { toExtendedJson } from '@xo-cash/utils';
|
|
import { BaseStream, type StreamMessage } from './base-stream.ts';
|
|
import type { SSEStreamingApi } from 'hono/streaming';
|
|
|
|
/** Maps neutral stream messages onto Hono's Server-Sent Events API. */
|
|
export class HonoSSEStream extends BaseStream {
|
|
readonly streaming = true;
|
|
readonly bidirectional = false;
|
|
|
|
/** Observers notified when the SSE connection closes. */
|
|
private readonly closeCallbacks: Array<() => void> = [];
|
|
|
|
/** Guards against sends after local closure. */
|
|
private closed = false;
|
|
|
|
/**
|
|
* @param stream - Hono SSE writer bound to the active HTTP response.
|
|
*/
|
|
constructor(private readonly stream: SSEStreamingApi) {
|
|
super();
|
|
}
|
|
|
|
/**
|
|
* Encode and write one SSE event frame.
|
|
*
|
|
* @param message - Neutral response or event envelope to send.
|
|
*/
|
|
async send(message: StreamMessage): Promise<void> {
|
|
if (this.closed) {
|
|
throw new Error('Cannot send to a closed SSE stream');
|
|
}
|
|
|
|
// SSE carries type and ID as protocol fields; the payload remains Extended JSON.
|
|
await this.stream.writeSSE({
|
|
event: message.type,
|
|
data: toExtendedJson('body' in message ? message.body : message.data),
|
|
...(message.id === undefined ? {} : { id: message.id }),
|
|
});
|
|
}
|
|
|
|
/** Close the SSE response and notify lifecycle observers. */
|
|
async close(): Promise<void> {
|
|
if (this.closed) {
|
|
return;
|
|
}
|
|
|
|
this.closed = true;
|
|
try {
|
|
await this.stream.close();
|
|
} catch (error) {
|
|
const errorInstance = error instanceof Error ? error : new Error(String(error));
|
|
throw errorInstance;
|
|
}
|
|
|
|
this.emitClose();
|
|
}
|
|
|
|
/**
|
|
* Register a callback invoked when the stream closes.
|
|
*
|
|
* @param callback - Called immediately if the stream is already closed.
|
|
*/
|
|
onClose(callback: () => void): void {
|
|
if (this.closed) {
|
|
callback();
|
|
|
|
return;
|
|
}
|
|
|
|
this.closeCallbacks.push(callback);
|
|
}
|
|
|
|
/** Drain and invoke all registered close observers. */
|
|
private emitClose(): void {
|
|
const callbacks = this.closeCallbacks.splice(0);
|
|
for (const callback of callbacks) {
|
|
callback();
|
|
}
|
|
}
|
|
}
|