From beefdfe665bf0ddea246e7764d8c2e729ce5c3a6 Mon Sep 17 00:00:00 2001 From: Alek Merani Date: Sat, 25 Apr 2026 17:53:56 -0700 Subject: [PATCH 1/4] HttpEventStreamStore --- CONTRIBUTING.md | 70 +++++++++ README.md | 18 --- demo/client/src/main.tsx | 49 ++++++- demo/server/src/index.ts | 65 +++++++++ src/EventStore.ts | 45 +----- src/HttpEventStreamStore.ts | 176 +++++++++++++++++++++++ src/SeverSentEventStore.ts | 44 ++++++ src/components/ReactTextStream/index.tsx | 16 ++- src/hooks/useTextStream.ts | 74 +++++++--- src/index.ts | 1 + 10 files changed, 473 insertions(+), 85 deletions(-) create mode 100644 CONTRIBUTING.md create mode 100644 src/HttpEventStreamStore.ts create mode 100644 src/SeverSentEventStore.ts diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..17c9e25 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,70 @@ +# Contributing + +Thanks for contributing! + +## Prerequisites + +- Node.js (this repo’s demo uses Vite; Node 20.19+ recommended) +- npm + +## Install + +From the repo root: + +```bash +npm install +``` + +## Running the demo (manual verification) + +The demo is split into two apps: + +- `demo/server`: Express server that exposes streaming endpoints +- `demo/client`: Vite + React client that consumes the streams + +### 1) Start the server + +```bash +cd demo/server +npm install +npm run dev +``` + +By default it runs on `http://localhost:3001`. + +Endpoints: + +- `GET /sse`: Server-Sent Events stream +- `POST /http-stream`: HTTP POST that responds with `text/event-stream` +- `GET /health`: health check + +### 2) Start the client + +In a second terminal: + +```bash +cd demo/client +npm install +npm run dev +``` + +The page will open automatically. It renders: + +- `ReactTextStream` component using SSE +- `useTextStream(url, onEvent)` using SSE +- `useTextStream({ store: "http", data, ... })` using HTTP POST + event-stream + +### 3) What to verify + +- **SSE path**: streaming words appear and ends with a “completed” event +- **HTTP POST path**: streaming words appear; changing the “HTTP message” input changes the request body sent to the server +- **Reconnect button**: restarts the stream + +## Local build + +From the repo root: + +```bash +npm run build +``` + diff --git a/README.md b/README.md index 087d5e4..0101618 100644 --- a/README.md +++ b/README.md @@ -6,24 +6,6 @@ A React library for streaming text using Server-Sent Events (SSE) with `text/eve ![Demo of rendering text event stream](./demo.gif) -### Run the demo locally - -From the repository root: - -```bash -npm install -npm run demo -``` - -This installs dependencies for [`demo/server`](./demo/server) and [`demo/client`](./demo/client), builds the library into `dist/`, then starts both in parallel: the SSE test server on [http://localhost:3001](http://localhost:3001) (endpoint `/sse`) and the Vite app (by default [http://localhost:5173](http://localhost:5173); Vite may open a browser automatically). The client loads `react-text-stream` via `file:../..`, matching how the published package resolves to `dist`. - -Other scripts: - -- **`npm run demo:server`** — SSE server only (`demo/server`, `npm run dev`). -- **`npm run demo:client`** — Vite dev server only (`demo/client`). Run **`npm run build`** at the repo root first so `dist/` exists for the linked package. -- **`npm run demo:install`** — install dependencies in both demo packages without starting servers. -- **`npm run demo:verify`** — build the library and run a production build of the demo client (useful to confirm the package layout before publish). - ## Installation ```bash diff --git a/demo/client/src/main.tsx b/demo/client/src/main.tsx index 39574e7..43fa350 100644 --- a/demo/client/src/main.tsx +++ b/demo/client/src/main.tsx @@ -5,6 +5,7 @@ import './index.css' const config = { url: 'http://localhost:3001/sse', + httpUrl: 'http://localhost:3001/http-stream', onEvent: (event: { type: string; word: string }) => event.type === 'completed' ? undefined : `${event.word ?? ''} `, } @@ -15,6 +16,7 @@ root.render() function App() { const [eventType, setEventType] = useState('') const [id, setId] = useState(Number.MAX_SAFE_INTEGER) + const [message, setMessage] = useState('hello from client') const onEvent = (event: { type: string; word: string }) => { setEventType(event.type) return config.onEvent(event) @@ -28,16 +30,28 @@ function App() {
({eventType === 'chunk' ? 'streaming' : eventType}) +
+ +
- + ) } function Streams({ onEvent, + httpMessage, }: { onEvent: (event: { type: string; word: string }) => string | undefined + httpMessage: string }) { return ( <> @@ -53,6 +67,10 @@ function Streams({

useTextStream() Hook

+
+

useTextStream() Hook (HTTP POST + event-stream)

+ +
) } @@ -67,3 +85,32 @@ function HookTextStream() { ) } + +function HookHttpTextStream({ message }: { message: string }) { + const [submittedMessage, setSubmittedMessage] = useState(undefined) + const stream = useTextStream({ + url: config.httpUrl, + store: 'http', + data: submittedMessage === undefined ? undefined : { message: submittedMessage }, + onEvent: config.onEvent, + }) + return ( + <> +
+ + +
+
+ {stream && stream.length > 0 ? String(stream) : ( + + {submittedMessage === undefined ? 'Waiting for input…' : 'Generating...'} + + )} +
+ + ) +} diff --git a/demo/server/src/index.ts b/demo/server/src/index.ts index e69647a..8015010 100644 --- a/demo/server/src/index.ts +++ b/demo/server/src/index.ts @@ -93,6 +93,70 @@ app.get('/sse', (req, res) => { }); }); +// HTTP POST -> SSE response endpoint +app.post('/http-stream', (req, res) => { + const { message } = (req.body ?? {}) as { message?: string }; + console.log('New HTTP stream request', { message }); + + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'Access-Control-Allow-Origin': '*', + 'Access-Control-Allow-Headers': 'Content-Type, Cache-Control', + }); + + res.write( + `data: ${JSON.stringify({ + type: 'connected', + message: 'HTTP stream established', + requestMessage: message ?? '', + })}\n\n`, + ); + + let wordIndex = 0; + const sendInterval = setInterval(() => { + if (wordIndex >= words.length) { + const completionChunk = { + type: 'completed', + message: 'All words have been sent', + totalWords: words.length, + timestamp: new Date().toISOString(), + }; + + res.write(`data: ${JSON.stringify(completionChunk)}\n\n`); + console.log(`HTTP stream completed: sent all ${words.length} words`); + + clearInterval(sendInterval); + res.end(); + return; + } + + const word = words[wordIndex]; + const chunk = { + type: 'chunk', + word: word, + wordLength: word.length, + index: wordIndex, + totalWords: words.length, + timestamp: new Date().toISOString(), + }; + + res.write(`data: ${JSON.stringify(chunk)}\n\n`); + wordIndex++; + }, 100); + + req.on('close', () => { + console.log('HTTP stream connection closed'); + clearInterval(sendInterval); + }); + + req.on('error', (error) => { + console.error('HTTP stream connection error:', error); + clearInterval(sendInterval); + }); +}); + // Health check endpoint app.get('/health', (req, res) => { res.json({ @@ -109,6 +173,7 @@ app.get('/', (req, res) => { message: 'SSE Test Server', endpoints: { '/sse': 'Server-Sent Events stream with lorem ipsum words', + '/http-stream': 'HTTP POST that responds with an event-stream', '/health': 'Health check endpoint' }, stats: { diff --git a/src/EventStore.ts b/src/EventStore.ts index e9898dc..847064d 100644 --- a/src/EventStore.ts +++ b/src/EventStore.ts @@ -1,44 +1,5 @@ -export function EventStore(url: string, onEvent: (event: P) => R|undefined) { - let retryCount = 0; - let currentData: R|undefined; - const listeners = new Set(); +import { SeverSentEventStore } from './SeverSentEventStore'; - function subscribe(callback: () => void) { - listeners.add(callback); - return () => listeners.delete(callback); - } +// Backwards-compat alias: the project historically exported EventStore. +export const EventStore = SeverSentEventStore; - function getSnapshot() { - return currentData; - } - - const eventSource = new EventSource(url) - - eventSource.onmessage = (event: MessageEvent) => { - const newData = JSON.parse(event.data) as P; - const parsedData = onEvent(newData); - if (parsedData === undefined) { - currentData = undefined; - retryCount = 0; - eventSource.close(); - return; - } - else { - currentData = parsedData; - listeners.forEach((listener: any) => listener()); - } - }; - - eventSource.onerror = () => { - retryCount++; - if (retryCount >= 3) { - retryCount = 0; - eventSource.close(); - } - }; - - return { - subscribe, - getSnapshot, - } -} \ No newline at end of file diff --git a/src/HttpEventStreamStore.ts b/src/HttpEventStreamStore.ts new file mode 100644 index 0000000..b4fdf5e --- /dev/null +++ b/src/HttpEventStreamStore.ts @@ -0,0 +1,176 @@ +type Listener = () => void; + +function toBodyInit(data: unknown): { body?: BodyInit; headers?: Record } { + if (data === undefined) return {}; + + // If caller already provided a BodyInit-ish value, pass through. + if ( + typeof data === 'string' || + data instanceof Blob || + data instanceof ArrayBuffer || + data instanceof FormData || + data instanceof URLSearchParams || + data instanceof ReadableStream + ) { + return { body: data as BodyInit }; + } + + return { + body: JSON.stringify(data), + headers: { 'Content-Type': 'application/json' }, + }; +} + +function parseSseEventDataFrames(sseFrame: string): string[] { + // SSE frames are separated by "\n\n". Within a frame, there can be multiple + // "data:" lines that should be concatenated with "\n". + const lines = sseFrame.split(/\r?\n/); + const dataLines: string[] = []; + for (const line of lines) { + if (line.startsWith('data:')) dataLines.push(line.slice('data:'.length).trimStart()); + } + if (dataLines.length === 0) return []; + return [dataLines.join('\n')]; +} + +export function HttpEventStreamStore( + url: string, + data: D | undefined, + onEvent: (event: P) => R | undefined, + fetchInit?: Omit, +) { + let retryCount = 0; + let currentData: R | undefined; + const listeners = new Set(); + + let controller: AbortController | undefined; + let stopped = false; + let currentInput: D | undefined = data; + + function notify() { + listeners.forEach((l) => l()); + } + + function abortCurrentRequest() { + controller?.abort(); + controller = undefined; + } + + async function start() { + // Defer network work until we both have an input payload and a consumer. + if (controller || stopped) return; + if (listeners.size === 0) return; + if (currentInput === undefined) return; + + controller = new AbortController(); + const { body, headers } = toBodyInit(currentInput); + + try { + const res = await fetch(url, { + ...fetchInit, + method: 'POST', + body, + headers: { + Accept: 'text/event-stream', + ...(headers ?? {}), + ...(fetchInit?.headers ?? {}), + }, + signal: controller.signal, + }); + + if (!res.ok) throw new Error(`HTTP ${res.status} ${res.statusText}`); + if (!res.body) throw new Error('Response body is not readable'); + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + + // Process complete SSE frames. + while (true) { + const idx = buffer.indexOf('\n\n'); + if (idx === -1) break; + + const frame = buffer.slice(0, idx); + buffer = buffer.slice(idx + 2); + + for (const dataStr of parseSseEventDataFrames(frame)) { + const newData = JSON.parse(dataStr) as P; + const parsedData = onEvent(newData); + if (parsedData === undefined) { + currentData = undefined; + retryCount = 0; + stop(); + return; + } + + currentData = parsedData; + notify(); + } + } + } + } catch (err) { + if (controller?.signal.aborted) return; + + retryCount++; + if (retryCount >= 3) { + retryCount = 0; + stop(); + return; + } + + // allow a subsequent start() attempt + controller = undefined; + if (!stopped) start(); + } + } + + function stop() { + stopped = true; + abortCurrentRequest(); + } + + function subscribe(callback: Listener) { + listeners.add(callback); + start(); + return () => { + listeners.delete(callback); + if (listeners.size === 0) stop(); + }; + } + + function getSnapshot() { + return currentData; + } + + function setData(nextData: D | undefined) { + currentInput = nextData; + + // If we already have an active request, restart with the new input. + if (controller) { + abortCurrentRequest(); + } + + // Clear out previous output when input is cleared. + if (currentInput === undefined) { + currentData = undefined; + retryCount = 0; + notify(); + return; + } + + start(); + } + + return { + subscribe, + getSnapshot, + setData, + }; +} + diff --git a/src/SeverSentEventStore.ts b/src/SeverSentEventStore.ts new file mode 100644 index 0000000..4227f10 --- /dev/null +++ b/src/SeverSentEventStore.ts @@ -0,0 +1,44 @@ +export function SeverSentEventStore(url: string, onEvent: (event: P) => R|undefined) { + let retryCount = 0; + let currentData: R|undefined; + const listeners = new Set(); + + function subscribe(callback: () => void) { + listeners.add(callback); + return () => listeners.delete(callback); + } + + function getSnapshot() { + return currentData; + } + + const eventSource = new EventSource(url) + + eventSource.onmessage = (event: MessageEvent) => { + const newData = JSON.parse(event.data) as P; + const parsedData = onEvent(newData); + if (parsedData === undefined) { + currentData = undefined; + retryCount = 0; + eventSource.close(); + return; + } + else { + currentData = parsedData; + listeners.forEach((listener: any) => listener()); + } + }; + + eventSource.onerror = () => { + retryCount++; + if (retryCount >= 3) { + retryCount = 0; + eventSource.close(); + } + }; + + return { + subscribe, + getSnapshot, + } +} \ No newline at end of file diff --git a/src/components/ReactTextStream/index.tsx b/src/components/ReactTextStream/index.tsx index 5689c81..8d285b7 100644 --- a/src/components/ReactTextStream/index.tsx +++ b/src/components/ReactTextStream/index.tsx @@ -5,17 +5,23 @@ interface ReactTextStreamProps { url: string; onEvent: (event: T) => string|undefined; render: (stream: string) => React.ReactNode; + store?: 'sse' | 'http'; + data?: unknown; + fetchInit?: Omit; } function ReactTextStream({ url, onEvent, - render + render, + store, + data, + fetchInit, }: ReactTextStreamProps) { - const stream = useTextStream( - url, - onEvent - )!; + const stream = + store !== undefined || data !== undefined || fetchInit !== undefined + ? useTextStream({ url, onEvent, store, data, fetchInit })! + : useTextStream(url, onEvent)!; return render(stream); } diff --git a/src/hooks/useTextStream.ts b/src/hooks/useTextStream.ts index 94a0b63..4966e3b 100644 --- a/src/hooks/useTextStream.ts +++ b/src/hooks/useTextStream.ts @@ -1,25 +1,61 @@ import { useState, useSyncExternalStore, useEffect, useMemo } from 'react'; import { EventStore } from '../EventStore'; +import { HttpEventStreamStore } from '../HttpEventStreamStore'; -function useTextStream

(url: string, onEvent: (event: P) => string|undefined) { - const eventStore = useMemo(() => EventStore(url, onEvent), [url]); - - const [stream, setStream] = useState(); - - const sseChunk = useSyncExternalStore(eventStore.subscribe, eventStore.getSnapshot); - - useEffect(() => { - if (sseChunk !== undefined) { - setStream((curStream) => { - if (!curStream) { - return sseChunk; - } - return curStream?.concat(sseChunk); - }) - } - }, [sseChunk]); - - return stream; +export type UseTextStreamOptions = { + url: string; + onEvent: (event: P) => string | undefined; + store?: 'sse' | 'http'; + data?: D; + fetchInit?: Omit; +}; + +function useTextStream

(url: string, onEvent: (event: P) => string | undefined): string | undefined; +function useTextStream(options: UseTextStreamOptions): string | undefined; +function useTextStream( + urlOrOptions: string | UseTextStreamOptions, + onEventArg?: (event: P) => string | undefined, +) { + const options: UseTextStreamOptions = + typeof urlOrOptions === 'string' + ? { url: urlOrOptions, onEvent: onEventArg as (event: P) => string | undefined, store: 'sse' } + : urlOrOptions; + + const storeType = options.store ?? 'sse'; + const url = options.url; + const onEvent = options.onEvent; + const data = options.data as D; + const fetchInit = options.fetchInit; + + const eventStore = useMemo(() => { + if (storeType === 'http') { + return HttpEventStreamStore(url, undefined, onEvent, fetchInit); + } + + // Preserve prior behavior for SSE: a stable store keyed by URL. + return EventStore(url, onEvent); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, storeType === 'http' ? [storeType, url, onEvent, fetchInit] : [storeType, url]); + + useEffect(() => { + if (storeType !== 'http') return; + (eventStore as ReturnType>).setData(data); + }, [storeType, eventStore, data]); + + const [stream, setStream] = useState(); + + const chunk = useSyncExternalStore(eventStore.subscribe, eventStore.getSnapshot); + + useEffect(() => { + if (chunk !== undefined) { + setStream((curStream) => { + if (!curStream) return chunk; + return curStream.concat(chunk); + }); + } + }, [chunk]); + + return stream; } export default useTextStream; \ No newline at end of file diff --git a/src/index.ts b/src/index.ts index 5f89502..02ec36e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,5 @@ import ReactTextStream from './components/ReactTextStream' import useTextStream from './hooks/useTextStream' +export type { UseTextStreamOptions } from './hooks/useTextStream' export { ReactTextStream, useTextStream } \ No newline at end of file From 6bff0a59e8f46a0dee49596430f92f3fc39be707 Mon Sep 17 00:00:00 2001 From: Alek Merani Date: Sat, 25 Apr 2026 18:05:49 -0700 Subject: [PATCH 2/4] fix example --- CONTRIBUTING.md | 21 ++++ demo/client/src/main.tsx | 36 +++---- demo/server/src/index.ts | 188 +++++++++++++++++++++++----------- test/demo-http-stream.test.ts | 147 ++++++++++++++++++++++++++ 4 files changed, 314 insertions(+), 78 deletions(-) create mode 100644 test/demo-http-stream.test.ts diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 17c9e25..a708277 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,6 +15,27 @@ From the repo root: npm install ``` +## Useful npm scripts (recommended) + +Run these from the repo root (they’re defined in the root `package.json`): + +```bash +# one command: install demo deps, build, then run server+client in parallel +npm run demo + +# if you only need pieces +npm run demo:install +npm run demo:server +npm run demo:client + +# library build + tests +npm run build +npm run test +npm run test:ci +npm run test:coverage +npm run test:ui +``` + ## Running the demo (manual verification) The demo is split into two apps: diff --git a/demo/client/src/main.tsx b/demo/client/src/main.tsx index 43fa350..decd3c7 100644 --- a/demo/client/src/main.tsx +++ b/demo/client/src/main.tsx @@ -16,7 +16,6 @@ root.render() function App() { const [eventType, setEventType] = useState('') const [id, setId] = useState(Number.MAX_SAFE_INTEGER) - const [message, setMessage] = useState('hello from client') const onEvent = (event: { type: string; word: string }) => { setEventType(event.type) return config.onEvent(event) @@ -30,29 +29,13 @@ function App() {
({eventType === 'chunk' ? 'streaming' : eventType}) -

- -
- + ) } -function Streams({ - onEvent, - httpMessage, -}: { - onEvent: (event: { type: string; word: string }) => string | undefined - httpMessage: string -}) { +function Streams({ onEvent }: { onEvent: (event: { type: string; word: string }) => string | undefined }) { return ( <>
@@ -69,7 +52,7 @@ function Streams({

useTextStream() Hook (HTTP POST + event-stream)

- +
) @@ -86,7 +69,8 @@ function HookTextStream() { ) } -function HookHttpTextStream({ message }: { message: string }) { +function HookHttpTextStream() { + const [message, setMessage] = useState('hello from client') const [submittedMessage, setSubmittedMessage] = useState(undefined) const stream = useTextStream({ url: config.httpUrl, @@ -96,6 +80,16 @@ function HookHttpTextStream({ message }: { message: string }) { }) return ( <> +
+ +