Streaming Nodes

A node declared with streaming: true processes a stream of packets — it runs once per packet instead of once per run, and the runtime owns the loop. Streaming comes in two forms, distinguished by the fn you write.

The two forms are driven by the same feed: a batch of packets delivered one at a time (see Streams). A streaming: true node connected downstream of an => feed runs once per delivered element.

Body-form (the runtime owns the loop)

The modern form removes the scaffold. With streaming: true, a bare function body is compiled as:

async (packet, state, emit) => { <body> }

and the runtime drives it:

for await (const packet of queue) await fn.call(this, packet, state, emit)

So the body has three variables in scope:

Variable	Description
`packet`	The current input packet; `packet.read()` unwraps the data
`state`	A fresh `{}` per run, persisted across that run's packets (a re-run gets a fresh object)
`emit`	Sends an output packet — the body-form replacement for `yield`

title: Stream Echo
ns: stream
name: echo
ports:
  input:
    in:
      type: any
  output:
    out:
      type: any
streaming: true
fn: |
  emit(packet.read())

`emit` routes output like `yield`

A bare value goes to the default (first non-virtual) output port: emit(packet.read()).
An object map routes to named ports: emit({text: head}) sends head to the text port. Raw values are wrapped into packets automatically, so emit({text: 'hello'}) works without Packet.create.
No output()/$.create() ceremony is needed — that is the injected-namespace style of the non-streaming fn.

`state` replaces the closure

State that must survive across packets lives in state, not in a closure:

streaming: true
fn: |
  if (packet.read() === '```') state.inCode = !state.inCode

Because state is fresh per run, a re-run starts clean — the same behavior the legacy form got by re-creating its generator.

Bodies compile as async functions, so await works inside them for free.

Legacy form (async generator)

The original form is an async generator. fn is the generator body, input is the per-packet queue, and output is expressed with yield:

streaming: true
fn: |
  async function* (input, $) {
    for await (const p of input) {
      yield p
    }
  }

yield p sends to the default output port.
yield {port: p} routes to a named port.
Per-node state is trapped in the generator closure, so every author hand-writes the same scaffold.

The generator source is detected by its signature — a string fn whose text matches async function* / function* is treated as legacy. Anything else is treated as a body. A plain (non-string) function with streaming: true is also a valid body-form fn.

When a stream ends

Streaming nodes are fed by batches, and the end of the batch matters. The rule is the same as everywhere else in the language:

A finite stream requires a batch boundary — a cyclic feed (=>, or sendIIP({cyclic: true}, arr) from code) opens a batch, delivers each element, and closes it. A plain, non-cyclic IIP writes one packet and never closes a batch, so a stream fed that way never sees an end-of-stream.

When the batch closes, the runtime relays the close downstream (its finally runs), so a streaming output can chain into a >= collector and materialize the stream as one array. This is what makes a streaming node a legitimate source for a plain, non-streaming consumer: the collector terminates the stream.

Streaming