// This file is auto-generated by @hey-api/openapi-ts import type { Config } from "./types.gen.js" export type ServerSentEventsOptions = Omit & Pick & { /** * Callback invoked when a network or parsing error occurs during streaming. * * This option applies only if the endpoint returns a stream of events. * * @param error The error that occurred. */ onSseError?: (error: unknown) => void /** * Callback invoked when an event is streamed from the server. * * This option applies only if the endpoint returns a stream of events. * * @param event Event streamed from the server. * @returns Nothing (void). */ onSseEvent?: (event: StreamEvent) => void /** * Default retry delay in milliseconds. * * This option applies only if the endpoint returns a stream of events. * * @default 3000 */ sseDefaultRetryDelay?: number /** * Maximum number of retry attempts before giving up. */ sseMaxRetryAttempts?: number /** * Maximum retry delay in milliseconds. * * Applies only when exponential backoff is used. * * This option applies only if the endpoint returns a stream of events. * * @default 30000 */ sseMaxRetryDelay?: number /** * Optional sleep function for retry backoff. * * Defaults to using `setTimeout`. */ sseSleepFn?: (ms: number) => Promise url: string } export interface StreamEvent { data: TData event?: string id?: string retry?: number } export type ServerSentEventsResult = { stream: AsyncGenerator ? TData[keyof TData] : TData, TReturn, TNext> } export const createSseClient = ({ onSseError, onSseEvent, responseTransformer, responseValidator, sseDefaultRetryDelay, sseMaxRetryAttempts, sseMaxRetryDelay, sseSleepFn, url, ...options }: ServerSentEventsOptions): ServerSentEventsResult => { let lastEventId: string | undefined const sleep = sseSleepFn ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))) const createStream = async function* () { let retryDelay: number = sseDefaultRetryDelay ?? 3000 let attempt = 0 const signal = options.signal ?? new AbortController().signal while (true) { if (signal.aborted) break attempt++ const headers = options.headers instanceof Headers ? options.headers : new Headers(options.headers as Record | undefined) if (lastEventId !== undefined) { headers.set("Last-Event-ID", lastEventId) } try { const response = await fetch(url, { ...options, headers, signal }) if (!response.ok) throw new Error(`SSE failed: ${response.status} ${response.statusText}`) if (!response.body) throw new Error("No body in SSE response") const reader = response.body.pipeThrough(new TextDecoderStream()).getReader() let buffer = "" const abortHandler = () => { try { reader.cancel() } catch { // noop } } signal.addEventListener("abort", abortHandler) try { while (true) { const { done, value } = await reader.read() if (done) break buffer += value const chunks = buffer.split("\n\n") buffer = chunks.pop() ?? "" for (const chunk of chunks) { const lines = chunk.split("\n") const dataLines: Array = [] let eventName: string | undefined for (const line of lines) { if (line.startsWith("data:")) { dataLines.push(line.replace(/^data:\s*/, "")) } else if (line.startsWith("event:")) { eventName = line.replace(/^event:\s*/, "") } else if (line.startsWith("id:")) { lastEventId = line.replace(/^id:\s*/, "") } else if (line.startsWith("retry:")) { const parsed = Number.parseInt(line.replace(/^retry:\s*/, ""), 10) if (!Number.isNaN(parsed)) { retryDelay = parsed } } } let data: unknown let parsedJson = false if (dataLines.length) { const rawData = dataLines.join("\n") try { data = JSON.parse(rawData) parsedJson = true } catch { data = rawData } } if (parsedJson) { if (responseValidator) { await responseValidator(data) } if (responseTransformer) { data = await responseTransformer(data) } } onSseEvent?.({ data, event: eventName, id: lastEventId, retry: retryDelay, }) if (dataLines.length) { yield data as any } } } } finally { signal.removeEventListener("abort", abortHandler) reader.releaseLock() } break // exit loop on normal completion } catch (error) { // connection failed or aborted; retry after delay onSseError?.(error) if (sseMaxRetryAttempts !== undefined && attempt >= sseMaxRetryAttempts) { break // stop after firing error } // exponential backoff: double retry each attempt, cap at 30s const backoff = Math.min(retryDelay * 2 ** (attempt - 1), sseMaxRetryDelay ?? 30000) await sleep(backoff) } } } const stream = createStream() return { stream } }