Read

A read starts after an offset and gives batches of decoded items.

interface Batch<A> {
  readonly items: ReadonlyArray<A>;
  readonly nextOffset: string;
  readonly upToDate: boolean;
  readonly closed: boolean;
}

Reads do not include the message at the given offset. Omit the offset to read from the start. Keep nextOffset to continue later.

head does not restart a TTL.

ReadCopy Anchor Link

import { Effect, Stream } from "effect";
import { Notes } from "./streams.ts";

const sizes = Effect.flatMap(Notes, (notes) =>
  notes.read({}).pipe(
    Stream.map((batch) => batch.items.length),
    Stream.runCollect,
  ),
);

const everything = Effect.flatMap(Notes, (notes) => notes.collect({}));

read is an Effect Stream of batches. It does nothing until a sink runs it. limit sets the maximum number of stored messages in one batch.

collect reads to the current tail and returns { items, offset }. offset is the tail it read up to. Pass it as expectedOffset for a conditional write, or as offset to read on from there.

FollowCopy Anchor Link

import { Effect, Stream } from "effect";
import { Notes } from "./streams.ts";

const live = Effect.flatMap(Notes, (notes) =>
  notes.follow({}).pipe(
    Stream.flatMap((batch) => Stream.fromIterable(batch.items)),
    Stream.runForEach((note) => Effect.log(note.text)),
  ),
);

const consumer = Effect.gen(function* () {
  const fiber = yield* live.pipe(Effect.forkScoped);
  return fiber;
});

follow catches up, then waits. It does not give empty batches. A closed batch ends the stream. Interrupt the fiber to release a parked wait. Use Stream.take to stop after a number of batches.

SessionCopy Anchor Link

import { Effect } from "effect";
import { Notes } from "./streams.ts";

const once = (offset: string) => Effect.flatMap(Notes, (notes) => notes.session({}, { offset }));

session is one long-poll wait, for an application with its own poll loop. The result has raw messages, nextOffset, cursor, upToDate, closed, and timedOut. The messages are not decoded.

FoldCopy Anchor Link

import { Effect } from "effect";
import { Notes } from "./streams.ts";

const count = Effect.flatMap(Notes, (notes) =>
  notes.fold(
    {},
    () => 0,
    (total) => total + 1,
  ),
);

fold reads to the current tail and returns { state, offset }, like collect. The initial value is a function, so each fold starts from a new value and a mutable accumulator is safe. To fold only new items, pass the earlier offset and start from the earlier state.

To keep folding while the stream grows, fold over follow:

import { Effect, Stream } from "effect";
import { Notes } from "./streams.ts";

const live = Effect.flatMap(Notes, (notes) =>
  notes.follow({}).pipe(
    Stream.flatMap((batch) => Stream.fromIterable(batch.items)),
    Stream.runFold(
      () => 0,
      (total) => total + 1,
    ),
  ),
);

A fold over follow completes when the stream closes.

ErrorsCopy Anchor Link

read, fold, and collect use ReadFailure. follow uses FollowFailure, which also includes NotSupported. head uses HeadFailure. Import these types from @streamsy/core.

follow retries only a StreamsFault with retryable: true, from the same offset. A failed transport request, HTTP 429, or HTTP 5xx is retryable. Other failures end the stream. The default retry is exponential from 200 ms, jittered and capped at 30 s; retry sets another schedule.

A DecodeFault stops the read. The application selects the offset to continue from.

A Durable State read gives insert and update operations from other producers as upsert.

A Durable State read also gives the control messages that the stream holds. A control has headers.control set to snapshot-start, snapshot-end or reset, and it has no type or key. A reader that folds rows should clear them at a reset and apply the items after it. StreamAddress.isControl and StreamAddress.isChange narrow an item.

Over HTTPCopy Anchor Link

GET on the stream path with offset reads one batch. Add live=long-poll or live=sse to follow. All access rules allow reads. See Protocol routes.