🌊 StreamingHelper

EN: Static utilities for streaming JSON lines (NDJSON), Server-Sent Events (SSE), and file downloads with progress tracking. Includes fetchSSE — a fetch-based SSE consumer that supports POST, AbortSignal, multiple event types, and for await (async generator).

ES: Utilidades estáticas para streaming de líneas JSON (NDJSON), Server-Sent Events (SSE) y descargas de archivos con seguimiento de progreso. Incluye fetchSSE — un consumidor SSE basado en fetch que soporta POST, AbortSignal, múltiples tipos de evento y for await (generador asíncrono).

¿Por qué StreamingHelper? / Why StreamingHelper?

ES: Cuando trabajas con endpoints que devuelven datos incrementalmente — como logs en tiempo real, pipelines de IA, o datasets NDJSON grandes — necesitas procesar cada fragmento sin cargar todo en memoria. StreamingHelper provee dos APIs para SSE:

	`streamSSE` (legacy)	`fetchSSE` ⭐ (nuevo)
Transporte	`EventSource` (solo GET)	`fetch` + `ReadableStream`
POST con body	❌	✅
`AbortSignal`	❌	✅
Múltiples event types	Solo 1 `eventType`	Todos, o filtra con `eventTypes[]`
API de consumo	Callbacks `.subscribe()`	`for await … of` (async generator)
`id:` / `retry:` fields	❌	✅

EN: When working with endpoints that return data incrementally — like real-time logs, AI pipelines, or large NDJSON datasets — you need to process each fragment without loading everything into memory. StreamingHelper provides two APIs for SSE:

	`streamSSE` (legacy)	`fetchSSE` ⭐ (new)
Transport	`EventSource` (GET-only)	`fetch` + `ReadableStream`
POST with body	❌	✅
`AbortSignal`	❌	✅
Multiple event types	Single `eventType`	All, or filter with `eventTypes[]`
Consumer API	Callbacks `.subscribe()`	`for await … of` (async generator)
`id:` / `retry:` fields	❌	✅

📦 Import

import { StreamingHelper } from "bytekit/streaming";
// Types (if needed) | Tipos (si es necesario):
import type { SSEEvent, FetchSSEOptions, StreamOptions, StreamResponse } from "bytekit/streaming";

📋 Static Methods / Métodos Estáticos

Method / Método	Signature / Firma	Description (EN)	Descripción (ES)
⭐ `fetchSSE`	`fetchSSE<T>(endpoint, options?): AsyncGenerator<SSEEvent<T>>`	fetch-based SSE with POST, abort, multi-event, `for await`	SSE basado en fetch con POST, abort, multi-evento, `for await`
`streamSSE`	`streamSSE<T>(endpoint, options?): { subscribe, close }`	Legacy EventSource-based SSE (GET only)	SSE basado en EventSource (solo GET)
`streamJsonLines`	`streamJsonLines<T>(endpoint, options?): Promise<StreamResponse<T>>`	Stream NDJSON lines from an endpoint	Transmite líneas NDJSON desde un endpoint
`downloadStream`	`downloadStream(endpoint, options?): Promise<Blob>`	Download a file with progress tracking	Descarga un archivo con seguimiento de progreso

🧩 Types / Tipos

SSEEvent<T> ⭐

EN: A single parsed SSE event yielded by fetchSSE. ES: Un evento SSE parseado individual, emitido por fetchSSE.

interface SSEEvent<T = unknown> {
  event: string;        // Event type (default: "message") | Tipo de evento (default: "message")
  data: T | string;     // Parsed JSON or raw string | JSON parseado o string crudo
  id?: string;          // Optional server-sent ID | ID opcional enviado por el servidor
  retry?: number;       // Optional retry interval (ms) | Intervalo de reintento opcional (ms)
}

FetchSSEOptions ⭐

Property / Propiedad	Type / Tipo	Default	Description (EN)	Descripción (ES)
`method`	`string`	`"GET"`	HTTP method (POST, PUT, etc.)	Método HTTP (POST, PUT, etc.)
`body`	`BodyInit \| Record<string, unknown> \| unknown`	—	Request body. Objects are auto-JSON-stringified	Body de la petición. Los objetos se serializan a JSON automáticamente
`headers`	`Record<string, string>`	`{}`	Custom headers. `Accept: text/event-stream` is always set	Headers personalizados. `Accept: text/event-stream` se envía siempre
`signal`	`AbortSignal`	—	Cancellation signal	Señal de cancelación
`eventTypes`	`string[]`	—	Filter: only yield these event types. Omit = yield all	Filtro: solo emite estos tipos de evento. Omitir = emitir todos
`raw`	`boolean`	`false`	If true, `data` is always a raw string (no JSON parse)	Si es true, `data` siempre es string crudo (sin JSON.parse)

StreamOptions (for streamJsonLines, streamSSE, downloadStream)

Property / Propiedad	Type / Tipo	Default	Description (EN)	Descripción (ES)
`timeout`	`number`	`30000`	Request timeout in milliseconds	Timeout de la petición en milisegundos
`headers`	`Record<string, string>`	`{}`	Custom request headers	Headers personalizados
`onChunk`	`(chunk: string) => void`	—	Called for each successfully parsed chunk	Se llama por cada chunk parseado exitosamente
`onError`	`(error: Error) => void`	—	Called when an error occurs	Se llama cuando ocurre un error
`onComplete`	`() => void`	—	Called when the stream finishes	Se llama cuando el stream termina

StreamResponse<T>

interface StreamResponse<T> {
  data: T[];          // All parsed items collected | Todos los items parseados recolectados
  complete: boolean;  // Whether stream finished successfully | Si el stream terminó exitosamente
  error?: Error;      // Error object if stream failed | Objeto Error si el stream falló
}

🧠 How It Works / Cómo Funciona

fetchSSE (fetch + ReadableStream) ⭐

EN: fetchSSE() calls fetch() with the given method, body, and headers. It reads the response as a ReadableStream, parses lines following the SSE specification, and yields SSEEvent objects. The parser handles multi-line data: fields, event:, id:, retry: fields, and comment lines (: prefix). Because it returns an AsyncGenerator, you consume it with for await … of.

ES: fetchSSE() llama a fetch() con el método, body y headers dados. Lee la respuesta como ReadableStream, parsea las líneas siguiendo la especificación SSE, y hace yield de objetos SSEEvent. El parser maneja campos data: multi-línea, event:, id:, retry:, y líneas de comentario (prefijo :). Como retorna un AsyncGenerator, se consume con for await … of.

fetchSSE()  —  fetch + ReadableStream + SSE parser
──────────────────────────────────────────────────
fetch(endpoint, { method, body, headers, signal })
     │
ReadableStream.getReader()
     │
  ┌──▼───────────────────────────────────┐
  │ Buffer accumulates text chunks       │
  │                                      │
  │ Parse SSE fields line by line:       │
  │   event: data                        │
  │   data: {"token":"hello"}            │
  │   id: evt-42                         │
  │   retry: 5000                        │
  │   : comment (skipped)                │
  │                                      │
  │ On blank line → yield SSEEvent<T>    │
  └──┬───────────────────────────────────┘
     │
  for await (const ev of ...) { ... }

streamSSE (EventSource — legacy)

streamSSE()  —  EventSource (GET only)
─────────────────────────────────────────
new EventSource(endpoint)
     │
addEventListener(eventType, handler)
     │
  ┌──▼───────────────────────────────────┐
  │ SSE event → JSON.parse → subscribers │
  │ SSE event → JSON.parse → subscribers │
  └──┬───────────────────────────────────┘
     │
  close() to disconnect

💡 Examples / Ejemplos

⭐ fetchSSE: POST con body + cancellation + multi-event types

import { StreamingHelper } from "bytekit/streaming";
import type { SSEEvent } from "bytekit/streaming";

// EN: Your critical use case — POST to an AI orchestrator with body + abort + multi-event
// ES: Tu caso de uso crítico — POST a un orquestador de IA con body + abort + multi-evento

interface PipelineChunk { token?: string; text?: string; }

const ac = new AbortController();

// Cancel after 30 seconds if needed
setTimeout(() => ac.abort(), 30_000);

for await (const ev of StreamingHelper.fetchSSE<PipelineChunk>(
  "/bff/ai/orchestrator-stream",
  {
    method: "POST",
    body: { prompt: "Explain DDD in 3 sentences", model: "gpt-4o" },
    signal: ac.signal,
    eventTypes: ["data", "log", "heartbeat"],
  }
)) {
  switch (ev.event) {
    case "data":
      process.stdout.write(ev.data.token ?? "");
      break;
    case "log":
      console.debug("[server log]", ev.data);
      break;
    case "heartbeat":
      // keep-alive — ignore
      break;
  }
}

⭐ fetchSSE: React hook for AI streaming

import { StreamingHelper } from "bytekit/streaming";
import type { SSEEvent } from "bytekit/streaming";
import { useEffect, useRef, useState } from "react";

interface TokenChunk { token: string; done: boolean; }

function useAIStream(prompt: string) {
  const [text, setText] = useState("");
  const [loading, setLoading] = useState(false);
  const abortRef = useRef<AbortController | null>(null);

  const start = async () => {
    abortRef.current?.abort();
    const ac = new AbortController();
    abortRef.current = ac;
    setLoading(true);
    setText("");

    try {
      for await (const ev of StreamingHelper.fetchSSE<TokenChunk>(
        "/bff/ai/chat-stream",
        {
          method: "POST",
          body: { prompt },
          signal: ac.signal,
          eventTypes: ["data"],
        }
      )) {
        const chunk = ev.data as TokenChunk;
        setText((prev) => prev + chunk.token);
        if (chunk.done) break;
      }
    } catch (err) {
      if ((err as Error).name !== "AbortError") throw err;
    } finally {
      setLoading(false);
    }
  };

  const stop = () => abortRef.current?.abort();

  useEffect(() => () => abortRef.current?.abort(), []);

  return { text, loading, start, stop };
}

⭐ fetchSSE: simple GET (same as EventSource, but composable)

// EN: Even for simple GET, fetchSSE gives you for-await
// ES: Incluso para GET simple, fetchSSE te da for-await
for await (const ev of StreamingHelper.fetchSSE<{ price: number }>(
  "https://api.example.com/stock/AAPL"
)) {
  console.log(`AAPL: $${(ev.data as { price: number }).price}`);
}

Stream NDJSON / Transmitir NDJSON

import { StreamingHelper } from "bytekit/streaming";

interface LogEntry { timestamp: string; level: string; message: string; }

const result = await StreamingHelper.streamJsonLines<LogEntry>(
  "https://api.example.com/logs/stream",
  {
    timeout: 60000,
    headers: { Authorization: "Bearer token123" },
    onChunk: (raw) => console.log("Received line:", raw),
    onComplete: () => console.log("Stream finished!"),
    onError: (err) => console.error("Stream error:", err.message),
  }
);

console.log(`Received ${result.data.length} log entries`);

Legacy streamSSE (EventSource — GET only)

// EN: Still works for simple GET SSE. Use fetchSSE for anything more.
// ES: Sigue funcionando para SSE GET simple. Usa fetchSSE para cualquier otra cosa.
const stream = StreamingHelper.streamSSE<{ title: string }>(
  "https://api.example.com/events",
  { eventType: "notification" }
);

const unsub = stream.subscribe((data) => console.log(data.title));

// Later:
unsub();
stream.close();

Download with Progress / Descarga con Progreso

const blob = await StreamingHelper.downloadStream(
  "https://api.example.com/reports/annual.pdf",
  {
    timeout: 120000,
    onProgress: (pct) => console.log(`${pct}%`),
    onComplete: () => console.log("Download complete!"),
  }
);

const url = URL.createObjectURL(blob);
const a = document.createElement("a");
a.href = url;
a.download = "annual-report.pdf";
a.click();
URL.revokeObjectURL(url);

🔗 Véase También / See Also

ApiClient — HTTP client for standard request/response patterns / Cliente HTTP para patrones estándar
WebSocketHelper — Bidirectional real-time communication / Comunicación bidireccional en tiempo real
Async-Toolkit — withTimeout, retry — combine with fetchSSE for resilience / combina con fetchSSE para resiliencia
FileUploadHelper — Upload files with chunking and progress / Sube archivos con chunking y progreso

StreamingHelper - sebamar88/bytekit GitHub Wiki

🌊 StreamingHelper

¿Por qué StreamingHelper? / Why StreamingHelper?

📦 Import

📋 Static Methods / Métodos Estáticos

🧩 Types / Tipos

SSEEvent<T> ⭐

FetchSSEOptions ⭐

StreamOptions (for streamJsonLines, streamSSE, downloadStream)

StreamResponse<T>

🧠 How It Works / Cómo Funciona

fetchSSE (fetch + ReadableStream) ⭐

streamSSE (EventSource — legacy)

💡 Examples / Ejemplos

⭐ fetchSSE: POST con body + cancellation + multi-event types

⭐ fetchSSE: React hook for AI streaming

⭐ fetchSSE: simple GET (same as EventSource, but composable)

Stream NDJSON / Transmitir NDJSON

Legacy streamSSE (EventSource — GET only)

Download with Progress / Descarga con Progreso

🔗 Véase También / See Also

⚠️ GitHub.com Fallback ⚠️

StreamingHelper - sebamar88/bytekit GitHub Wiki

🌊 StreamingHelper

¿Por qué StreamingHelper? / Why StreamingHelper?

📦 Import

📋 Static Methods / Métodos Estáticos

🧩 Types / Tipos

SSEEvent<T> ⭐

FetchSSEOptions ⭐

StreamOptions (for streamJsonLines, streamSSE, downloadStream)

StreamResponse<T>

🧠 How It Works / Cómo Funciona

fetchSSE (fetch + ReadableStream) ⭐

streamSSE (EventSource — legacy)

💡 Examples / Ejemplos

⭐ fetchSSE: POST con body + cancellation + multi-event types

⭐ fetchSSE: React hook for AI streaming

⭐ fetchSSE: simple GET (same as EventSource, but composable)

Stream NDJSON / Transmitir NDJSON

Legacy streamSSE (EventSource — GET only)

Download with Progress / Descarga con Progreso

🔗 Véase También / See Also

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️