effect-stream

Use Stream.paginate with an initial cursor and an effectful function returning the current page plus Option.some(nextCursor) or Option.none() to stop. Combine it with mapEffect for concurrent enrichment and runForEach or runCollect to consume results.

Pipe the stream through Ndjson.decodeString() or Ndjson.decodeSchemaString(MySchema)() using Stream.pipeThroughChannel. For binary Uint8Array sources use Ndjson.decode(), and catch NdjsonError with Stream.catchTag to handle malformed lines.

Yes, use NodeStream.fromReadable from @effect/platform-node with an evaluate function and an onError mapper. The closeOnDone option defaults to true, closing the underlying Node stream when consumption finishes.

Stream.broadcast starts the producer immediately, so late subscribers miss earlier values unless replay is configured. Stream.broadcastN waits until all n downstream streams subscribe before starting, guaranteeing every consumer sees the full sequence.

Stream.retry restarts the entire stream from the beginning on each retry rather than resuming mid-stream. Combine it with Schedule.exponential and Schedule.take to bound attempts with backoff.

Wrap acquisition in Effect.acquireRelease inside Stream.scoped or Stream.callback so finalizers run when the stream completes. Since beta.69, Stream.scoped also extends its scope to pull effects created by fromEffect and mapEffect.