307 lines
8.2 KiB
TypeScript
307 lines
8.2 KiB
TypeScript
// TODO: 1. Handle reconnection logic.
|
|
// 2. Allow specifying custom
|
|
// 3. Try to make it so that connect DOES NOT resolve until connection is actually establish.
|
|
// There's a very slim chance of a race-condition otherwise where we might miss an event.
|
|
// Note that this IS NOT trivial. Claude will probably NOT be able to do this for you!
|
|
|
|
/**
|
|
* A Server-Sent Events client implementation using fetch API.
|
|
* Supports custom headers, POST requests, and is non-blocking.
|
|
*/
|
|
export class SSEClient {
|
|
/**
|
|
* Creates and connects a new SSEClient instance.
|
|
* @param url The URL to connect to
|
|
* @param options Configuration options
|
|
* @returns A new connected SSEClient instance
|
|
*/
|
|
public static async from(
|
|
url: string,
|
|
options: Partial<SSEClientOptions> = {},
|
|
): Promise<SSEClient> {
|
|
const client = new SSEClient(url, options);
|
|
await client.connect();
|
|
return client;
|
|
}
|
|
|
|
// State.
|
|
private url: string;
|
|
private controller: AbortController;
|
|
private connected: boolean = false;
|
|
protected options: SSEClientOptions;
|
|
protected messageBuffer: Uint8Array = new Uint8Array();
|
|
|
|
// Text decoders and encoders for parsing the message buffer.
|
|
private textDecoder: TextDecoder = new TextDecoder();
|
|
private textEncoder: TextEncoder = new TextEncoder();
|
|
|
|
/**
|
|
* Creates a new SSEClient instance.
|
|
* @param url The URL to connect to
|
|
* @param options Configuration options
|
|
*/
|
|
constructor(url: string, options: Partial<SSEClientOptions> = {}) {
|
|
this.url = url;
|
|
this.options = {
|
|
// Use default fetch function.
|
|
fetch: fetch,
|
|
method: 'GET',
|
|
headers: {
|
|
Accept: 'text/event-stream',
|
|
'Cache-Control': 'no-cache',
|
|
},
|
|
onOpen: () => {},
|
|
onMessage: () => {},
|
|
onError: (error) => console.error('SSEClient error:', error),
|
|
onClose: () => {},
|
|
...options,
|
|
};
|
|
this.controller = new AbortController();
|
|
}
|
|
|
|
/**
|
|
* Connects to the SSE endpoint.
|
|
*/
|
|
public async connect(): Promise<void> {
|
|
if (this.connected) return;
|
|
|
|
this.connected = true;
|
|
const { method, headers, body } = this.options;
|
|
|
|
const fetchOptions: RequestInit = {
|
|
method,
|
|
headers,
|
|
body,
|
|
signal: this.controller.signal,
|
|
cache: 'no-store',
|
|
};
|
|
|
|
// Using a regular Promise executor (not async)
|
|
return new Promise((resolve, reject) => {
|
|
// Move the async logic inside a separate function
|
|
const establishConnection = async () => {
|
|
try {
|
|
// Fetch from the provided URL using our fetch function.
|
|
const response = await this.options.fetch(this.url, fetchOptions);
|
|
|
|
if (!response.ok) {
|
|
throw new Error(`HTTP error! Status: ${response.status}`);
|
|
}
|
|
|
|
if (!response.body) {
|
|
throw new Error('Response body is null');
|
|
}
|
|
|
|
// Notify that the connection has been established
|
|
if (this.options.onOpen) {
|
|
this.options.onOpen();
|
|
}
|
|
|
|
const reader = response.body.getReader();
|
|
|
|
// Start the read here so we can resolve and then start handling the stream
|
|
// We will await this at the start of the loop, and redeclare it at the end of each loop iteration
|
|
let readPromise = reader.read();
|
|
|
|
// Resolve before starting the loop so that we can start performing other actions
|
|
// Keep this out of the loop to avoid flags or `already resolved` errors.
|
|
resolve();
|
|
|
|
while (true) {
|
|
const { done, value } = await readPromise;
|
|
|
|
if (done) {
|
|
this.connected = false;
|
|
if (this.options.onClose) {
|
|
this.options.onClose();
|
|
}
|
|
break;
|
|
}
|
|
|
|
// Decode and parse the chunk
|
|
const events = this.parseEvents(value);
|
|
|
|
// Handle each complete event
|
|
for (const event of events) {
|
|
if (this.options.onMessage) {
|
|
this.options.onMessage(event);
|
|
}
|
|
}
|
|
|
|
// Start reading the next chunk
|
|
readPromise = reader.read();
|
|
}
|
|
} catch (error) {
|
|
this.connected = false;
|
|
|
|
if (this.options.onError) {
|
|
this.options.onError(error);
|
|
}
|
|
|
|
if (this.options.onClose) {
|
|
this.options.onClose();
|
|
}
|
|
|
|
// Now we'll reject the promise instead of resolving
|
|
reject(error);
|
|
}
|
|
};
|
|
|
|
// Execute the async function
|
|
establishConnection();
|
|
});
|
|
}
|
|
|
|
protected parseEvents(chunk: Uint8Array): SSEvent[] {
|
|
// Append new chunk to existing buffer
|
|
this.messageBuffer = new Uint8Array([...this.messageBuffer, ...chunk]);
|
|
|
|
const events: SSEvent[] = [];
|
|
const lines = this.textDecoder.decode(this.messageBuffer).split(/\r\n|\r|\n/);
|
|
|
|
let currentEvent: Partial<SSEvent> = {};
|
|
let completeEventCount = 0;
|
|
|
|
// Iterate over the lines to find complete events
|
|
for (let i = 0; i < lines.length; i++) {
|
|
const line = lines[i];
|
|
|
|
// Empty line signals the end of an event
|
|
if (line === '') {
|
|
if (currentEvent.data) {
|
|
// Remove trailing newline if present
|
|
currentEvent.data = currentEvent.data.replace(/\n$/, '');
|
|
events.push(currentEvent as SSEvent);
|
|
currentEvent = {};
|
|
completeEventCount = i + 1;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
// Parse field: value format
|
|
const colonIndex = line.indexOf(':');
|
|
if (colonIndex === -1) continue;
|
|
|
|
const field = line.slice(0, colonIndex);
|
|
// Skip initial space after colon if present
|
|
const valueStartIndex =
|
|
colonIndex + 1 + (line[colonIndex + 1] === ' ' ? 1 : 0);
|
|
const value = line.slice(valueStartIndex);
|
|
|
|
if (field === 'data') {
|
|
currentEvent.data = currentEvent.data
|
|
? currentEvent.data + '\n' + value
|
|
: value;
|
|
} else if (field === 'event') {
|
|
currentEvent.event = value;
|
|
} else if (field === 'id') {
|
|
currentEvent.id = value;
|
|
} else if (field === 'retry') {
|
|
const retryMs = parseInt(value, 10);
|
|
if (!isNaN(retryMs)) {
|
|
currentEvent.retry = retryMs;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Store the remainder of the buffer for the next chunk
|
|
const remainder = lines.slice(completeEventCount).join('\n');
|
|
this.messageBuffer = this.textEncoder.encode(remainder);
|
|
|
|
return events;
|
|
}
|
|
|
|
/**
|
|
* Closes the SSE connection.
|
|
*/
|
|
public close(): void {
|
|
if (!this.connected) return;
|
|
|
|
this.controller.abort();
|
|
this.connected = false;
|
|
}
|
|
|
|
/**
|
|
* Checks if the client is currently connected.
|
|
* @returns Whether the client is connected
|
|
*/
|
|
public isConnected(): boolean {
|
|
return this.connected;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Configuration options for the SSEClient.
|
|
*/
|
|
export interface SSEClientOptions {
|
|
/**
|
|
* The fetch function to use.
|
|
*
|
|
* NOTE: This is compatible with Browser/Node's native "fetcH" function.
|
|
* We use this in place of "typeof fetch" so that we can accept non-standard URLs ("url" is a "string" here).
|
|
* For example, a LibP2P adapter might not use a standardized URL format (and might only include "path").
|
|
* This would cause a type error as native fetch expects type "URL".
|
|
*/
|
|
fetch: (url: string, options: RequestInit) => Promise<Response>;
|
|
|
|
/**
|
|
* HTTP method to use (GET or POST).
|
|
*/
|
|
method: 'GET' | 'POST';
|
|
|
|
/**
|
|
* HTTP headers to send with the request.
|
|
*/
|
|
headers?: Record<string, string>;
|
|
|
|
/**
|
|
* Body to send with POST requests.
|
|
*/
|
|
body?: string | FormData | URLSearchParams;
|
|
|
|
/**
|
|
* Called when the connection is established.
|
|
*/
|
|
onOpen?: () => void;
|
|
|
|
/**
|
|
* Called when a message is received.
|
|
*/
|
|
onMessage?: (event: SSEvent) => void;
|
|
|
|
/**
|
|
* Called when an error occurs.
|
|
*/
|
|
onError?: (error: unknown) => void;
|
|
|
|
/**
|
|
* Called when the connection is closed.
|
|
*/
|
|
onClose?: () => void;
|
|
}
|
|
|
|
/**
|
|
* Represents a Server-Sent Event.
|
|
*/
|
|
export interface SSEvent {
|
|
/**
|
|
* Event data.
|
|
*/
|
|
data: string;
|
|
|
|
/**
|
|
* Event type.
|
|
*/
|
|
event?: string;
|
|
|
|
/**
|
|
* Event ID.
|
|
*/
|
|
id?: string;
|
|
|
|
/**
|
|
* Reconnection time in milliseconds.
|
|
*/
|
|
retry?: number;
|
|
}
|