Stream Through the API

Most routes answer one value. Some requirements need more:

• a report is sent month by month, as it is computed — reply streams;
• the back office keeps its list of orders current without reloading it — collection streams;
• a supplier’s price sheet is imported while it uploads — streamed request bodies;
• a customer talks to the one process that holds their session — stateful operations;
• an invoice kept at a third party is served as if the shop held it — flow:fetch.

Writing the operations themselves is in Read a Large Collection, Stream a Reply or an Input and Keep Something in a Process.

§Reply streams

An operation that answers a stream — a generator, or a Readable in object mode — is exposed like any other. Nothing in the route says it streams. export here is one more operation of store.orders, the component Create a Component writes, and summarize a function of its own that answers the month, orders and revenue of one month:

// components/store.orders/operations/export.ts
import type { Context, ExportInput } from '../types/index.d.ts'

export async function * effect (input: ExportInput, context: Context) {
  for (const month of input.months)
    yield await summarize(month, context)
}

Add its input and its route to the manifest:

# components/store.orders/manifest.toa.yaml
operations:
  # …
  export:
    input:
      type: object
      properties:
        months:
          type: array
          items: { type: string }
      required: [months]

exposition:
  # …
  /exports:
    anonymous: true
    POST:
      endpoint: export
      io:output: [month, orders, revenue]

The reply is a multipart response, one part per yielded value, sent as it is produced:

POST /store/orders/exports/ HTTP/1.1
content-type: application/yaml
accept: application/yaml

months: [2026-01, 2026-02]

HTTP/1.1 201 Created
content-type: multipart/yaml; boundary=cut

--cut
ACK
--cut
month: 2026-01
orders: 312
revenue: 1840200
--cut
month: 2026-02
orders: 287
revenue: 1699000
--cut
FIN
--cut--

• io:output applies to every object the stream yields. Without it the stream is closed unread and the reply is 204.
• A client reads parts as they arrive. A stream that fails midway ends with FIN like one that completed, so an operation whose consumers must know it finished yields a last part that says so.

$ npm run restart
$ curl -N http://localhost:8000/store/orders/exports/ 
    -H 'content-type: application/yaml' -H 'accept: application/yaml' 
    -d 'months: [2026-01, 2026-02]'

§Collection streams

A client that keeps a copy of a collection — an order list in an admin console, an offline catalogue — reads the collection once and afterwards asks only for what changed. An operation of scope stream serves this; every component with an entity inherits one, named stream. The route is open to staff:

# components/store.orders/manifest.toa.yaml
exposition:
  /feed:
    auth:role: staff
    GET:
      endpoint: stream
      io:output: [id, customer, status, total, VERSION]
      query:
        criteria: status!=cancelled
        limit: { value: 100, range: [1, 500] }

The reply is multipart, with three kinds of parts:

GET /store/orders/feed/ HTTP/1.1
accept: application/yaml

HTTP/1.1 200 OK
content-type: multipart/yaml; boundary=cut

--cut
ACK
--cut
entry:
  id: 4c4759e6f9c74da989d64511df42d6f4
  status: pending
  total: 4200
  VERSION: 1
--cut
removed: 99988d785d7d445cad45dbf8531f560b
--cut
token: eyJ2IjoxLCJwIjoi...
--cut
FIN
--cut--

§Reading in windows

Each request carries the token the previous reply ended with:

GET /store/orders/feed/?limit=100                 100 entries, then a token T1
GET /store/orders/feed/?token=T1&limit=100        the next 100, then T2
GET /store/orders/feed/?token=T2&limit=100        42 entries: the collection is read; then T3

GET /store/orders/feed/?token=T3&limit=100        later: what changed since, then T4

A window shorter than limit means the collection has been read. From then on the token answers changes only, and an empty reply — a token and nothing else — means nothing changed.

A client’s loop is therefore the same from the first request on: send the last token, apply the parts, keep the new token.

A list the user scrolls through is read this way: the first window when the page loads, and each next one as the user reaches the end of what is shown. The token lives as long as the page does, and a page that is loaded again starts from the first window.

A reply that ends without a token was cut. Send the same request again, with the token it carried.

§Newest first, then changes

A long collection is often shown from its newest end, without ever reading all of it. stop ends the read after the first window, however long the collection is:

GET /store/orders/feed/?sort=CREATED:desc&limit=100&stop HTTP/1.1
accept: application/yaml

HTTP/1.1 200 OK
content-type: multipart/yaml; boundary=cut

--cut
ACK
--cut
entry:
  id: 7a1f0c52b3e94d6f8a2c9e417d05b6c3
  status: pending
  total: 1800
  VERSION: 1
--cut
…                                                 98 entries more, newest first
--cut
entry:
  id: 4c4759e6f9c74da989d64511df42d6f4
  status: approved
  total: 4200
  VERSION: 3
--cut
token: eyJ2IjoxLCJwIjoi...
--cut
FIN
--cut--

The window is full, and without stop its token would continue with the next hundred. With stop it continues with changes:

GET /store/orders/feed/?sort=CREATED:desc&limit=100&stop        the newest 100, then a token T1

GET /store/orders/feed/?sort=CREATED:desc&limit=100&token=T1    later: what changed since, then T2
GET /store/orders/feed/?sort=CREATED:desc&limit=100&token=T2    later: what changed since, then T3

A client that keeps its copy between launches keeps the token beside it. What it reads when it loads depends on whether it has one:

// the client: called once, when it loads
export function load (token?: string) {
  const query = { sort: 'CREATED:desc', limit: 100 }

  return stream('/store/orders/feed/', token === undefined
    ? { ...query, stop: true }
    : { ...query, token })
}

stream stands for the client’s own reader of a multipart reply.

• stop is sent with the first request only. A token knows what it continues.
• Keep every token a read ends with, in place of the one before.
• limit beside a token of changes is the most changes one reply carries. A reply that carries as many is followed by another request, as a full window is.
• A change arrives for any entity of the collection, also one older than the window. A client that shows the newest hundred drops an entry it has no place for.
• Without limit, the window is of the size the route declares: value.

§Rules

• A stream route that declares no limit answers the whole collection in one reply, and stop changes nothing there.
• omit is refused.
• sort, beside limit or token, takes only id and CREATED, which never change for an entity. Anything else is 400.
• A token belongs to the criteria and order it was issued under. Presented under others, it is 400.
• A token too old for the storage to continue from is 410 Gone. Drop the copy and read the collection again without a token.
• Include VERSION in io:output, so that the client can tell an older state from a newer.

The query declaration of a stream route is described in List with Filters, Sorting and Paging.

A collection stream is complete but not immediate: the client learns of a change when it asks. A client that is to show a change the moment it happens loads the collection with a stream and is then told of every change by events, which it applies without reading anything: Keeping a copy of a collection.

§Turning a stream into one value

Some clients cannot read multipart, and some streams are really one answer assembled in steps. flow:compose collects the parts of a reply stream and answers a single value built from them.

Its value is an object whose values are JavaScript expressions. The collected parts are available as the array $:

exposition:
  /exports/summary:
    POST:
      endpoint: export
      io:output: true
      flow:compose:
        first: $[0].revenue
        second: $[1].revenue
        total: $[0].revenue + $[1].revenue
        months: $

HTTP/1.1 201 Created
content-type: application/yaml

first: 1840200
second: 1699000
total: 3539200
months:
  - month: 2026-01
    orders: 312
    revenue: 1840200
  - month: 2026-02
    orders: 287
    revenue: 1699000

• A value that is not an expression over $ is taken literally: count: 1. A string that should be literal is escaped with a backslash: label: \total.

• Arrays are composed element by element: pair: [$[0].month, $[1].month].

• The value may be a single expression instead of an object: flow:compose: $[1].

• The value may be a list of shapes. The first one whose expressions all evaluate is answered, which handles a stream that may end early:

  flow:compose:
    - total: $[0].revenue + $[1].revenue
    - total: $[0].revenue

The whole stream is held in memory until it ends. Compose short streams.

Its most common use is with file uploads, to answer one object instead of the workflow’s parts; see Process an Uploaded File.

§Streamed request bodies

A body that has to flow through an operation — a file being checksummed, transcoded, parsed or forwarded — is handed to it as a stream with map:stream. The operation declares which input property carries the stream, and its component the http binding such a call travels over; both are explained in Take a stream as input. The catalogue is a component of its own, store.catalog:

# components/store.catalog/manifest.toa.yaml
bindings: [amqp, http]

operations:
  import:
    type: effect
    stream: sheet
    input:
      properties:
        supplier:
          type: string

exposition:
  /imports/:supplier:
    POST:
      endpoint: import
      io:output: [imported, rejected]
      map:segments:
        supplier: ~supplier
      map:stream: sheet

// components/store.catalog/operations/import.ts
import type { Context, ImportInput } from '../types/index.d.ts'

export async function effect ({ supplier, sheet }: ImportInput, context: Context) {
  let imported = 0

  for await (const row of parse(sheet.stream))
    imported += await add(supplier, row, context)

  return { imported, rejected: 0 }
}

parse and add are functions of the component’s own: one reads rows from the stream, the other stores a row.

The property is filled with an object:

The body is not parsed, so the rest of the input comes from the path and other map directives. The directive names the same property the operation declares in stream; a route that maps a stream onto an operation that does not take one is refused when the manifest is read.

§Limits and media types

The long form states what the route takes and what it answers:

      map:stream:
        property: sheet
        accept: [text/csv, application/vnd.ms-excel]
        limit: 20MiB

Units of limit are those of file uploads, see Upload and Download Files.

§Answering bytes

An operation may answer a stream of bytes — a rendered PDF, a converted image. The route states the media types it can answer in with produces. thumbnail here is an operation of a component of pictures that declares stream: source and answers a Readable:

# components/pictures/manifest.toa.yaml
exposition:
  /thumbnails:
    POST:
      endpoint: thumbnail
      io:output: true
      map:stream:
        property: source
        accept: [image/jpeg, image/png]
        produces: [image/webp, image/jpeg]
        limit: 10MiB

The request’s accept is resolved against produces before the call. The operation is handed the one media type chosen, as accept beside the stream, and answers bytes in it; the response carries it as content-type.

• A client that names several of them gets the first it named. A client that names none, or accepts anything, gets the first the route lists.
• A client that asks for something the route does not produce is 406, and the operation does not run.
• A route that declares produces answers bytes. If the operation answers a value there, the response is 422.
• Everything produces lists must be the same answer in another encoding. A route that would answer two different things answers them at two paths.

§When not to stream

A streamed call is not retried: if it fails, the client is told and nothing tries again. Map a stream only when a body must pass through an operation.

• A body that is to be kept is a file: store it with octets:put and process it from a workflow.
• A body small enough to be a value is a value.

§Stateful operations

A stateful operation runs in one particular process — the one holding a connection, a session, a running job — and a call to it names that process. Over HTTP the name travels in a route variable, which map:instance points at. auth:anyone admits any signed-in user:

# components/assistant/manifest.toa.yaml
exposition:
  /sessions/:instance:
    auth:anyone: true
    POST:
      endpoint: say
      io:input: [message]
      io:output: false
      map:instance: instance

assistant here is a component that holds a conversation with a customer in memory; say is declared stateful: true.

POST /assistant/sessions/7f3a9c1e/ HTTP/1.1
content-type: application/yaml

message: Where is my order?

The variable is taken out of the input and out of the criteria; it only addresses the call.

How does a client learn the name? From the application: an ordinary operation returns it. A process reads its own name as context.instance, so the operation that starts a session stores or returns that value, and the client uses it in the path from then on.

A name is not a secret by itself and not an access rule. Protect the route as any other, and have the stateful operation check that the caller is the one the session belongs to.

§Answering with someone else’s content

flow:fetch answers a request with content fetched from a URL the application chooses. The client asks the application; the application decides where the content really is; the gateway fetches it and passes it on.

The directive names an operation that returns the URL. billing.invoices is a new component that knows where each invoice is kept, and this manifest is all of it:

# components/billing.invoices/manifest.toa.yaml
name: invoices
namespace: billing

operations:
  locate:
    errors: [NO_INVOICE]
    input:
      type: object
      properties:
        authority: { type: string }
        path: { type: string }
        parameters:
          type: object
          properties:
            id: { type: string }
          required: [id]
      required: [authority, path, parameters]

exposition:
  /:id/pdf:
    auth:role: staff
    GET:
      io:output: true
      flow:fetch: billing.invoices.locate

// components/billing.invoices/operations/locate.ts
import type { LocateInput } from '../types/index.d.ts'

export async function computation ({ parameters }: LocateInput) {
  if (!/^[a-f0-9]{32}$/.test(parameters.id)) return new Error('NO_INVOICE')

  return `https://billing.example/invoices/${parameters.id}.pdf`
}

The operation receives as input:

and returns one of:

• a URL string, fetched with GET;
• an object { url, options }, where options may hold method, headers and body;
• an error, which answers 404.

The fetched response’s body is streamed to the client with its content-type, content-length and etag. A fetch that does not succeed is 404. If the client disconnects, the fetch is abandoned. A request that carries if-none-match is answered 304 without a fetch.

The endpoint is written in full — namespace.component.operation, or component.operation for a component without a namespace — also inside a component’s own manifest.

Use it to serve documents kept at a third party without giving their addresses or credentials to the client. For files the application stores itself, use Upload and Download Files.