Streaming

Stream stages (SSE / AsyncIterable)

A stage whose stream function returns an AsyncIterable<T>. The orchestrator collects all emitted chunks into an array (the stage result). onChunk is called for each chunk in real time.

const orchestrator = createPipeline([
  { key: 'auth', request: async () => getToken() },
  {
    key: 'liveData',
    stream: async function* ({ prev }) {
      const source = new EventSource(`/api/stream?token=${prev}`)
      yield* eventSourceToAsyncIterable(source)
    },
    onChunk: (chunk, sharedData) => {
      sharedData.partial = (sharedData.partial ?? '') + chunk
      updateUI(sharedData.partial)
    },
  },
  {
    key: 'finalize',
    // allResults.liveData.data is the full array of chunks
    request: async ({ allResults }) => allResults.liveData.data.join(''),
  },
])

• Respects abort() — checks the abort signal between each chunk.
• Supports continueOnError — failed stream stages can be skipped like any other step.
• Emits standard step events: step:start, step:success, step:error.

Pagination

paginate() iterates a paginated API's pages as an AsyncGenerator<T[]>, hiding the difference between cursor-based and offset/limit-based APIs:

import { paginate, paginateAll, flattenPages } from 'rest-pipeline-js'

// Cursor-based (default strategy)
for await (const page of paginate({
  fetchPage: (cursor) => client.get('/items', { params: { cursor } }).then((r) => r.data),
})) {
  console.log(page.length, 'items')
}

// Offset/limit-based
for await (const page of paginate({
  strategy: 'offset',
  limit: 50,
  fetchPage: (offset, limit) =>
    client.get('/items', { params: { offset, limit } }).then((r) => r.data),
})) {
  console.log(page.length, 'items')
}

fetchPage returns { items, nextCursor } (cursor strategy — stop when nextCursor is null/undefined) or { items, total? } (offset strategy — stops when a page is shorter than limit, or offset reaches total if the API reports one).

• paginateAll(options) — collects every page into one flat array; simplest option when the total dataset is small enough to hold in memory at once.
• flattenPages(pages) — turns a stream of pages into a stream of individual items; useful as a StreamStageConfig.stream source when onChunk should fire per item rather than per page (see examples/pagination-stream.ts).
• Both paginate() and its fetchPage callback accept an optional signal for abort() support.

See examples/pagination-fanout.ts instead if you know the page count upfront and want to fetch them concurrently rather than sequentially.

WebSocket stages

A stage that runs over a persistent WebSocket connection instead of a single request/response — chat/presence feeds, live order books, collaborative editing events, etc. Messages returned by onMessage are collected into the stage's data array (same pattern as stream stages' chunks); onChunk fires per message in real time.

const orchestrator = pipe()
  .step({ key: 'auth', request: async () => getToken() })
  .websocket({
    key: 'chatFeed',
    url: ({ prev }) => `wss://chat.example.com/rooms/general?token=${prev}`,
    onOpen: () => console.log('connected'),
    onMessage: (data) => JSON.parse(data),
    onChunk: (message, sharedData) => updateUI(message),
    closeOn: (message) => message.text === '__end__',
    onClose: ({ wasClean }) => console.log('closed, clean:', wasClean),
    onError: (error) => console.error(error),
    timeoutMs: 5 * 60_000,
  })
  .build()

• url — string or a function of { prev, allResults, sharedData, signal }, same params request gets.
• createWebSocket — factory for the underlying implementation. Defaults to globalThis.WebSocket (browsers, Deno, Node ≥22). For Node <22, pass one backed by the ws package: createWebSocket: (url, protocols) => new WS(url, protocols).
• onMessage (required) — called per message with event.data; can be async. A non-undefined return value is collected into the stage's result array and passed to onChunk.
• Success/error is decided by the close event, not the error event: most WebSocket implementations fire error immediately before close, so onError alone doesn't fail the stage — a clean close (wasClean: true) resolves the stage successfully with everything collected so far; an unclean close rejects it, going through continueOnError like any other stage.
• closeOn(data) — return true to close the connection and end the stage successfully once you've seen what you need, instead of waiting for the server to close it.
• timeoutMs — overall connection timeout (not reset by messages); closes the socket and fails the stage if it fires before a clean close happens on its own.
• Respects abort() — closes the underlying connection and rejects the stage.

See examples/websocket-stage.ts for the full annotated version.