orderEventStream()

Signature

async function* orderEventStream<T>(
  source: AsyncIterable<EventEnvelope<T>>,
  options?: StreamOrderOptions<T>,
): AsyncIterable<StreamOrderBatch<T>>

This is the streaming entry point for watermark-aware ordering with explicit late-arrival and correction behavior.

Usage

import { orderEventStream } from "causal-order"

for await (const batch of orderEventStream(source(), {
  batchSize: 100,
  maxLateArrivalMs: 30_000n,
  lateArrivalPolicy: "flag",
  strict: false,
})) {
  console.log(batch.events)
  console.log(batch.anomalies)
  console.log(batch.watermark, batch.isFinal)
}

Details

• Consumes an async iterable of event envelopes.
• Emits watermark-aware ordered batches.
• Supports explicit late-arrival policies and correction-capable output.