River TS Streaming

import { RiverEvents } from 'river.ts';

const events = new RiverEvents()
  .defineEvent('message', { message: 'Hello' })
  .defineEvent('data', { data: {} as { id: number; name: string } })
  .defineEvent('stream', { data: [] as string[], stream: true, chunkSize: 100 })
  // Request/response pattern with explicit response type
  .defineEvent('rpc.call', {
    data: {} as { method: string; params: unknown },
    response: {} as { result: unknown; error?: string }
  })
  // Runtime validation: `data` is inferred from the schema's output
  .defineEvent('job.run', {
    schema: z.object({ id: z.string(), priority: z.number() }),
    responseSchema: z.object({ ok: z.boolean() }) // optional, for request()
  })
  .build();

import { RiverEmitter } from 'river.ts/server';

const emitter = RiverEmitter.init(events);

// Create SSE stream for HTTP response
const stream = emitter.stream({
  callback: async (emit, clientId) => {
    await emit('message', { message: 'Connected' });
    await emit('data', { data: { id: 1, name: 'test' } });
  },
  clientId: 'optional-custom-id',
  ondisconnect: (clientId) => console.log(`${clientId} disconnected`)
});

return new Response(stream, { headers: emitter.headers() });

// Broadcast to all clients
await emitter.broadcast('message', { message: 'Update' });

// Send to specific client
await emitter.sendToClient('client-id', 'data', { data: { id: 2, name: 'specific' } });

const stream = emitter.stream({
  retry: 3000,            // ms; written once as `retry:` when the stream opens
  keepAlive: 15_000,      // ms; writes a `: keep-alive` comment so proxies keep the connection
  lastEventId: request.headers.get('Last-Event-ID'), // read it from the request yourself
  signal: request.signal,
  callback: async (emit, clientId, lastEventId) => {
    // lastEventId is undefined on a first connection; replay everything after it
    for (const entry of log.after(lastEventId)) {
      await emit('message', { message: entry.text }, entry.id); // 3rd arg = event id
    }
  }
});

await emitter.broadcast('message', { message: 'Update' }, 42);          // optional id
await emitter.sendToClient('client-id', 'message', { message: 'x' }, 43); // optional id

import { RiverClient } from 'river.ts/client';

const client = RiverClient.init(events, { reconnect: true });

client
  .prepare('http://localhost:3000/events', { method: 'GET' })
  .on('message', (data) => console.log(data.message))
  .on('data', (data) => console.log(data.id, data.name))
  .stream();

// Close connection
client.close();
// stream() can be called again after close()

const client = RiverClient.init(events, {
  reconnect: true,                       // or { initialDelay: 1000, maxDelay: 30_000 }
  lastEventId: savedId,                  // sent as Last-Event-ID on the first connection
  onInvalid: (type, issues, raw) => {},  // events that fail their schema; never dispatched
  fetchFn: fetch,
  headers: { Authorization: 'Bearer ...' }
});

client.lastEventId; // read-only: id of the last event received; persist it to resume later
client.addEventListener('open', () => {});       // each successful connection
client.addEventListener('reconnect', (e) => {}); // (e as CustomEvent).detail = { attempt, delay, error }
client.addEventListener('close', () => {});      // stopped for good