Stream a Reply or an Input

The shop’s assistant answers a customer’s question with text generated by a language model, and the customer should see it as it is written. A catalog manager imports a price list too large to send as one value. The first is a streamed reply, which costs nothing and needs no declaration. The second is a streamed input, which is a different kind of call.

§Answer with a stream

Write the operation as a generator:

// components/assistant/operations/answer.ts
import type { AnswerInput, Context } from '../types/index.d.ts'

export async function* computation (input: AnswerInput, context: Context) {
  const { url, key } = context.configuration

  const response = await context.fetch(`${url}/completions`, {
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: `Bearer ${key.unwrap()}` },
    body: JSON.stringify({ prompt: input.question, stream: true })
  })

  if (!response.ok) throw new Error(`Model answered ${response.status}`)

  for await (const chunk of response.body.pipeThrough(new TextDecoderStream())) yield chunk
}

assistant is a component of its own. Its manifest declares the input, and the address and the key of the model as configuration; the request is made with context.fetch:

# components/assistant/manifest.toa.yaml
name: assistant

operations:
  answer:
    input:
      type: object
      properties:
        question: { type: string }
      required: [question]

configuration:
  schema:
    properties:
      url: { type: string, format: uri }
      key: { type: string, format: secret }
    required: [url, key]

or return a Readable, as thumbnail of a component that keeps pictures does; render stands for what draws one:

// components/pictures/operations/thumbnail.ts
import type { Context, ThumbnailInput } from '../types/index.d.ts'
import { Readable } from 'node:stream'

export async function computation (input: ThumbnailInput, context: Context) {
  return Readable.from(render(input.picture))
}

Nothing is declared. An operation may decide call by call whether it answers a value or a stream. The call itself is an ordinary one.

A reply is a value or a stream, never a value with a stream inside it.

§Try it

With the address and the key given as Call an External Service gives those of a mail provider:

$ npm run restart
$ npx toa call assistant.answer "{ input: { question: 'Where is my order?' } }"

The chunks are printed as they are yielded. Over HTTP the gateway serves the reply as a streamed response; see Streams and Flow.

§Read it in another operation

The caller receives a Readable:

const answer = await context.remote.assistant.answer({ input: { question } })

for await (const chunk of answer) {
  // ...
}

The stream carries bytes where the operation yielded Buffers, and values otherwise.

§When it fails halfway

A stream that failed never looks like a complete one. An exception thrown by the operation after its first value is thrown inside the caller’s for await, and so is the failure of a stream whose process stopped. Wrap the reading, not only the call, where the caller has something to undo:

try {
  for await (const chunk of answer) await write(chunk)
} catch (exception) {
  await discard()

  throw exception
}

§Take a stream as input

store.catalog, a component that keeps the products, imports a price list as it is uploaded, line by line, without holding the file. Name the input property that carries the stream, and declare the http binding beside the component’s usual one:

# components/store.catalog/manifest.toa.yaml
name: catalog
namespace: store

bindings: [amqp, http]

entity:
  properties:
    name: { type: string }
    price: { type: number }
    currency: { type: string }

operations:
  transit:
    query: false
    input:
      type: object
      properties:
        name: { type: string }
        price: { type: number }
        currency: { type: string }
  import:
    type: effect
    stream: file
    input:
      type: object
      properties:
        currency: { type: string }
      required: [currency]

The operation finds a Readable there:

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

export async function effect (input: ImportInput, context: Context) {
  const stream = input.file.stream ?? input.file

  let count = 0

  for await (const line of createInterface({ input: stream })) {
    const [name, price] = line.split(',')

    await context.local.transit({
      input: { name, price: Number(price), currency: input.currency }
    })

    count++
  }

  return { count }
}

It stores each line with the inherited transit, which a component has once its manifest declares it; see Use the Inherited Operations.

A caller puts a stream there:

import { createReadStream } from 'node:fs'

const { count } = await context.remote.store.catalog.import({
  input: { currency: 'EUR', file: createReadStream(path) }
})

To hand the body of an HTTP request to such an operation, see Streams and Flow.

§What to declare

stream names one input property. The rest of input is declared and validated as usual. The named property is not validated, and it is required: a call that omits it is refused.

stream may be declared on an effect, a computation and an unmanaged operation, not on a transition.

The operation needs the http binding, declared for the whole component as above, or for the operation alone under its own bindings. Without it the component does not start:

Operation 'import' takes a stream, which none of its bindings carries

Ordinary calls, tasks and events of the component keep using its other binding. Components run in one process need no binding for it.

§How to call

The property holds the stream, or an object whose stream member holds it, beside whatever the caller wants to say about it:

await context.remote.store.catalog.import({ input: { currency: 'EUR', file: stream } })

await context.remote.store.catalog.import({
  input: { currency: 'EUR', file: { type: 'text/csv', stream } }
})

The second form is what the gateway hands over for an upload: type is what arrived and accept is what the client asked to be answered with. That is why import reads input.file.stream ?? input.file.

These calls throw RequestContract before anything is sent:

• a call to an operation that takes a stream and carries none;
• a call that carries a stream to an operation that takes none;
• a value where the stream goes;
• a task or a delayed call that carries a stream.

§What a streamed call lacks

So:

• Handle the failure in the caller. Nothing has tried again, and a retry needs a fresh stream.
• A component scaled to zero or between deployments fails the call.
• A hundred calls in flight are a hundred sockets at each end.

§When not to use it

A streamed input is for bytes that flow through an operation and are not kept: a parse, a checksum, a transcode, a forward.

• An input that is a value is an ordinary call, however large the value.
• An answer that arrives in pieces is a streamed reply.
• An upload that is to be kept is stored by the gateway as it arrives, and components are called afterwards with a reference to it; see Files and File Workflows. Where bytes are stored and processed, store them first.

§Addresses for a local run

In a deployment there is nothing to state: a component is reached at its own service, on port 8005. Where processes run side by side on one machine, state the ports in the Context:

# context.toa.yaml
http@local:
  store.catalog: http://127.0.0.1:8005
  pictures: http://127.0.0.1:8006

A key is a component, a namespace, or . for every component. A process listens on the port of each component it serves, and callers use the same map to reach it.

§Limits

• Declaring stream on an operation that used to be ordinary changes how it is called. Every caller has to run a runtime version that knows the declaration before it is deployed; see Upgrades.
• A component that runs in a network of its own cannot be reached by a call that carries a stream.
• An effect under continuity answers values, and cannot take a streamed reply as a step.

The trade-offs are discussed in Streamed Calls.