feat: implement server-sent events consumer

Author: softsideof

src/streaming.ts

@@ -0,0 +1,127 @@
+/**
+ * FastFetch Server-Sent Events (SSE) streaming support.
+ *
+ * Consumes a `ReadableStream` from a `fetch()` Response body and calls
+ * `handler` for each well-formed SSE event received.
+ *
+ * SSE spec reference: https://html.spec.whatwg.org/#server-sent-events
+ */
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+export interface SSEEvent {
+  /** Event type field (defaults to `"message"` if not specified by server). */
+  type: string;
+  /** The `data` field value. Multi-line data fields are joined with `\n`. */
+  data: string;
+  /** Optional `id` field from the server. */
+  id?: string;
+  /** Optional `retry` field in milliseconds. */
+  retry?: number;
+}
+
+export type SSEHandler = (event: SSEEvent) => void | Promise<void>;
+
+// ---------------------------------------------------------------------------
+// consumeSSE
+// ---------------------------------------------------------------------------
+
+/**
+ * Consume a Server-Sent Events stream and invoke `handler` for each event.
+ *
+ * The function resolves when the stream ends (or `signal` is aborted).
+ * The underlying reader is always released in the `finally` block.
+ *
+ * @param reader  A `ReadableStreamDefaultReader<Uint8Array>` from `response.body.getReader()`.
+ * @param handler Called for each complete SSE event.
+ * @param signal  Optional `AbortSignal` to stop consuming early.
+ *
+ * @example
+ * ```ts
+ * const res = await fetch('/api/stream');
+ * if (!res.body) throw new Error('No body');
+ * await consumeSSE(res.body.getReader(), (event) => {
+ *   console.log(event.type, event.data);
+ * });
+ * ```
+ */
+export async function consumeSSE(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  handler: SSEHandler,
+  signal?: AbortSignal,
+): Promise<void> {
+  const decoder = new TextDecoder("utf-8");
+  let buffer = "";
+
+  function parseAndEmit(block: string): void {
+    // SSE field defaults
+    let eventType = "message";
+    let data = "";
+    let id: string | undefined;
+    let retry: number | undefined;
+
+    for (const line of block.split("\n")) {
+      const colon = line.indexOf(":");
+      if (colon === -1) continue; // ignore malformed lines
+
+      const field = line.slice(0, colon).trim();
+      // A space immediately after ":" is part of the spec and must be stripped.
+      const value = line.slice(colon + 1).replace(/^ /, "");
+
+      switch (field) {
+        case "event":
+          eventType = value;
+          break;
+        case "data":
+          data = data ? `${data}\n${value}` : value;
+          break;
+        case "id":
+          id = value;
+          break;
+        case "retry": {
+          const ms = parseInt(value, 10);
+          if (!isNaN(ms)) retry = ms;
+          break;
+        }
+        // "comment" lines (field === "") are silently ignored per spec
+      }
+    }
+
+    // Dispatch only if a data field was present (spec §9.2.6)
+    if (data !== "") {
+      void handler({ type: eventType, data, id, retry });
+    }
+  }
+
+  try {
+    while (true) {
+      if (signal?.aborted) break;
+
+      const { done, value } = await reader.read();
+      if (done) break;
+
+      buffer += decoder.decode(value, { stream: true });
+
+      // Events are separated by two newlines (\n\n or \r\n\r\n)
+      // Split on double-newline — keep trailing incomplete chunk in buffer.
+      const parts = buffer.split(/\n\n|\r\n\r\n/);
+      buffer = parts.pop() ?? "";
+
+      for (const block of parts) {
+        const trimmed = block.trim();
+        if (trimmed) {
+          parseAndEmit(trimmed);
+        }
+      }
+    }
+
+    // Flush any remaining data in the buffer
+    if (buffer.trim()) {
+      parseAndEmit(buffer.trim());
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}