Streaming Element streams

Element streams

Add one complete row per async-iterable element.

Use connectElementStream when every yielded value is a complete row. The chat grid below starts empty; each element the async generator yields is appended as a new row:

Streaming chat grid

Turn a streaming LLM response into rows with connectElementStream and append them to the grid as they arrive.

.md

"use client";

import { connectElementStream } from "@pretable/stream-adapter";
import { PretableSurface, useDisposeOnUnmount } from "@pretable/react";
import { createLocalRowModel } from "@pretable/core";
import { useEffect, useMemo } from "react";

import { columns, type ChatRow } from "./columns";
import {
  responseEventsToChatRows,
  type ChatResponseEvent,
} from "./response-events-to-chat-rows";

export type OpenChatResponseEvents = (input: {
  readonly model: string;
  readonly prompt: string;
}) =>
  AsyncIterable<ChatResponseEvent> | Promise<AsyncIterable<ChatResponseEvent>>;

export function ChatGrid({
  prompt,
  openResponseEvents,
}: {
  prompt: string;
  openResponseEvents: OpenChatResponseEvents;
}) {
  const rowModel = useMemo(
    () => createLocalRowModel({ rows: [], columns, getRowId: (row) => row.id }),
    [],
  );

  useEffect(() => {
    let disposed = false;
    let connection: ReturnType<typeof connectElementStream> | undefined;
    void (async () => {
      const stream = await openResponseEvents({
        model: "gpt-5",
        prompt,
      });
      const rows: AsyncIterable<ChatRow> = responseEventsToChatRows(stream);
      connection = connectElementStream(rowModel, rows);
      if (disposed) connection.dispose();
    })();
    return () => {
      disposed = true;
      connection?.dispose();
    };
  }, [openResponseEvents, prompt, rowModel]);

  // NOT `useEffect(() => () => rowModel.dispose())`: StrictMode rehearses an
  // unmount in dev, `useMemo` hands the same model back to the remount, and the
  // grid then renders nothing at all. `useDisposeOnUnmount` defers the disposal
  // by a microtask so a remount can cancel it.
  useDisposeOnUnmount(rowModel);

  return (
    <PretableSurface
      ariaLabel="Streaming chat"
      model={rowModel}
      viewportHeight={320}
    />
  );
}

Signature

function connectElementStream<
  TRow extends object,
  TRowId extends string | number,
>(
  rowModel: RowModelLike<TRow, TRowId>,
  stream: AsyncIterable<TRow>,
): StreamConnection;

The structural RowModelLike contract accepts atomic { add, update, remove } transactions. A Pretable local row model satisfies it directly.

React recipe

The grid above owns its rowModel with useMemo, opens the connection inside useEffect, and disposes both the connection and the row model on unmount — see its ChatGrid.tsx in the Code tab. It passes prompt through to a responseEventsToChatRows async generator, since Responses API streams yield lifecycle and text-delta events, not rows: the generator accumulates text deltas and yields a complete { id, role, content, tokens, latencyMs } value only when a response.completed event arrives (response-events-to-chat-rows.ts, same tab).

Two required props are easy to drop when hand-rolling this: PretableSurface needs both ariaLabel and viewportHeight, as the grid above passes them — neither is optional.

SSE recipe

connectElementStream(rowModel, fromEventSource("/api/events"));

Raw JSON

const response = await fetch("/api/events");
const strings = response.body!.pipeThrough(new TextDecoderStream());
connectElementStream(rowModel, parseElementStream<Row>(strings));

See Parsers →.