Read a Large Collection

The shop exports all its orders, and keeps a search index of them up to date. The orders are those of store.orders, the component Create a Component writes. Its enumerate answers one page, and a page of everything does not fit in memory. Read a stream instead: the entities arrive one at a time, and a later read can ask only for what changed since.

§Read the whole collection

A component that takes the generic prototype, as every component with an entity does, has a stream operation. Call it like any other and iterate over the reply. store.reports here is a component of its own, with nothing in its manifest but its name:

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

// components/store.reports/operations/count.ts
import type { Context } from '../types/index.d.ts'

export async function computation (input: null, context: Context) {
  const stream = await context.remote.store.orders.stream({
    query: { criteria: 'status==approved' }
  })

  let total = 0

  for await (const part of stream)
    if (part.entry !== undefined) total += part.entry.total

  return total
}

However many orders there are, neither side holds more than one at a time.

Try it from the command line; the parts are printed as they arrive:

$ npx toa call store.orders.stream "{ query: { criteria: 'status==approved' } }"

§Parts

A stream yields parts, each one of three:

{ "entry": { "id": "a1", "customer": "c7", "total": 1200, "VERSION": 4 } }
{ "removed": "a2" }
{ "token": "eyJ2IjoxLCJwIjoi..." }

“The collection” is what the query selects. An order that stops matching status==pending left the collection of pending orders: its reader is told removed, though the order still exists.

§Export the orders

To work on the entities inside the component that owns them, write an operation whose second parameter is named stream:

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

export async function* observation (input: null, stream: AsyncIterable<{ entry?: Entity }>) {
  for await (const part of stream)
    if (part.entry !== undefined) yield toRow(part.entry)
}

function toRow (order: Entity) {
  return `${order.id},${order.customer},${order.total}\n`
}

The operation is handed the parts and answers what it makes of them, here a streamed reply of CSV rows:

$ npm run restart
$ npx toa call store.orders.export "{ query: { criteria: 'status==approved' } }"

The stream scope is available to an observation and to an effect. Where the parameter name cannot be read from the source, declare it:

# components/store.orders/manifest.toa.yaml
operations:
  export:
    type: observation
    scope: stream

An operation that yields its own values, as export does, hides the token from its caller. One that is to be read again from a token passes the token part through, or returns the stream as it is.

To serve the export or the stream over HTTP, see Streams and Flow.

§Read only what changed

A complete stream ends with a token. Present it on the next read, with the same query, and the answer is what happened in the collection since, followed by a new token.

store.search keeps a search index of pending orders. It is a component of its own; its manifest states its name and asks for process memory, where sync keeps the token:

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

state: ~

// components/store.search/operations/sync.ts
import type { Query } from '@toa.io/core/types'
import type { Context } from '../types/index.d.ts'

export async function effect (input: null, context: Context) {
  const query: Query = { criteria: 'status==pending' }

  if (context.state.token !== undefined) query.token = context.state.token

  const stream = await context.remote.store.orders.stream({ query })

  for await (const part of stream) {
    if (part.entry !== undefined) await index.put(part.entry)
    else if (part.removed !== undefined) await index.drop(part.removed)
    else if (part.token !== null) context.state.token = part.token
  }
}

The first run, without a token, transfers the collection. Every later run transfers the difference, and nothing if nothing changed. Called by a pulse, sync keeps the index a few minutes behind at most.

The token is kept in process memory for brevity; index stands for the client of the search engine. A reader whose copy outlives its process keeps the token beside the copy.

Tokens of changes need two things from the storage; see Turn changes on.

§Read in windows

A reader may take a bounded number of entities at a time by stating limit:

const stream = await context.remote.store.orders.stream({
  query: { criteria: 'status==pending', limit: 100, token }
})

Every window ends with a token, and the next read carries the last one received:

Windows are ordered by id ascending. sort may order them by id or by CREATED, in either direction; entities created at the same moment are ordered by id:

query: { criteria: 'status==pending', sort: ['CREATED:desc'], limit: 100 }

sort by any other property beside limit or token is refused with RequestContract. Changes arrive in the order they happened.

stop: true reads the first window and nothing deeper; its token continues with changes. That is a feed: the newest twenty, then whatever changes.

query: { criteria: 'customer==c7', sort: ['CREATED:desc'], limit: 20, stop: true }

A change then arrives for any entity of the collection, including those deeper than the window.

§What the reader handles

Keep the higher VERSION. An entity may arrive twice, once in a window and once as a change. Where the query states a projection, include VERSION in it.

A stream that ends without a token was cut. A stream that failed ends too, so the token is what says a stream is whole. Read again with the token already held.

A token can expire. A read from a token that is too old throws the StateHistory exception. Drop the copy and read the collection again, without a token.

A token belongs to its query. Presented with other criteria or another order, it is refused with QuerySyntax.

A token may be null. Where the storage keeps no history of changes, a complete stream ends with { "token": null }: the collection is whole, and what changed later is learned only by reading it again. A full window still ends with a token that continues the collection. sync above handles a null by not keeping it.

§Turn changes on

With MongoDB, a stream ends with a token of changes when two things hold. Otherwise it ends with token: null.

MongoDB runs as a replica set. The one npm run dock starts, from the docker-compose.yaml toa create wrote, is a replica set of one member.

The collection keeps images of entities before each change. Turn them on with a migration:

# components/store.orders/migrations/0003-images.yaml
- images: true

Before turning them on:

• They cost a copy of every changed entity for as long as the replica set’s oplog holds it.
• A token lasts as long as the oplog holds the point it names.
• A read from a token costs what the whole replica set wrote since that point. Read often enough to keep that small.
• - images: false in a later migration stops keeping them. Tokens issued before are then refused as expired.

§The query

omit is not available: a window is continued by its token, not by an offset.

A deleted entity is left out of the collection, and its deletion arrives as removed. With deleted: true it stays in the collection, and a deletion arrives as an entry with DELETED set.

A component that reacts to single changes subscribes to an event; a reader that keeps a copy of a collection reads a stream. The idea is described in Collection Streams.