Streaming Parsers

Parsers

Lower-level parse* helpers if you need to feed the engine yourself.

The parse* helpers sit one layer below connect*. They accept an AsyncIterable<string> (raw JSON chunks from fetch, Response.body, SSE, etc.) and yield typed values. Pipe their output into the connect helpers.

The grid below builds its AsyncIterable<ChatRow> with a hand-written generator, because its source is a typed Responses API event stream. parseElementStream solves the same problem for a plainer source: raw streaming JSON text, with no event envelope to unwrap.

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}
    />
  );
}

`parseElementStream<TRow>(stream)`

Source emits a streaming JSON array. The parser yields each element as it parses — you don't wait for the array to close.

import { parseElementStream } from "@pretable/stream-adapter";
 
const res = await fetch("/api/events");
const decoded = res.body!.pipeThrough(new TextDecoderStream());
for await (const row of parseElementStream<Row>(decoded)) {
  console.log("got row", row);
}

In the connect-element pattern:

import {
  connectElementStream,
  parseElementStream,
} from "@pretable/stream-adapter";
 
const res = await fetch("/api/events");
const decoded = res.body!.pipeThrough(new TextDecoderStream());
connectElementStream(rowModel, parseElementStream<Row>(decoded));

`parsePartialStream<TRow>(stream)`

Source emits a streaming JSON object. The parser yields successive Partial<TRow> snapshots as keys complete.

import { parsePartialStream } from "@pretable/stream-adapter";
 
const res = await fetch("/api/single-row-events");
const decoded = res.body!.pipeThrough(new TextDecoderStream());
for await (const partial of parsePartialStream<Row>(decoded)) {
  console.log("snapshot so far", partial);
}

In the connect-partial pattern:

import {
  connectPartialStream,
  parsePartialStream,
} from "@pretable/stream-adapter";
 
const res = await fetch("/api/single-row-events");
const decoded = res.body!.pipeThrough(new TextDecoderStream());
connectPartialStream(rowModel, parsePartialStream<Row>(decoded), {
  rowId: "row-001",
});

Errors

Both parsers throw on malformed JSON. Wrap the consumer (or the connect call) in a try/catch and inspect via the done promise:

const decoded = res.body!.pipeThrough(new TextDecoderStream());
const conn = connectElementStream(rowModel, parseElementStream(decoded));
conn.done.catch((err) => console.error("stream failed:", err));