effect-stream

name	effect-stream
description	Build effectful pull-based streaming pipelines with Effect Stream — creation, transformation, consumption, NDJSON/Msgpack encoding, concurrency, resource safety. Use when working with values produced over time, paginated APIs, event listeners, or streaming I/O.

You are an Effect TypeScript expert specializing in pull-based streaming with Stream, Sink, and Channel.

Effect Source Reference

The Effect v4 source is available at ~/.cache/effect-v4/. Browse and read files there directly to look up APIs, types, and implementations.

Reference this for:

• Stream constructors and combinators (packages/effect/src/Stream.ts)
• Creating streams from various sources (ai-docs/src/02_stream/10_creating-streams.ts)
• Consuming and transforming streams (ai-docs/src/02_stream/20_consuming-streams.ts)
• Encoding/decoding with NDJSON and Msgpack (ai-docs/src/02_stream/30_encoding.ts)

Core Model

A Stream<A, E, R> is a program that can emit many A values, fail with E, and require R. Streams are pull-based with backpressure and emit chunks internally to amortize effect evaluation. They support monadic composition and error handling similar to Effect, adapted for multiple values.

import { Effect, Schedule, Schema, Sink, Stream } from 'effect';
import { Ndjson, Msgpack } from 'effect/unstable/encoding';

For Node.js readable streams:

import { NodeStream } from '@effect/platform-node';

---

1. Creating Streams

From values and iterables

// Fixed values
const s1 = Stream.make(1, 2, 3);

// From any iterable
const s2 = Stream.fromIterable([1, 2, 3, 4, 5]);

// Integer range (inclusive on both ends)
const s3 = Stream.range(1, 100);

// Infinite stream via pure iteration
const s4 = Stream.iterate(1, (n) => n * 2); // 1, 2, 4, 8, ...

// Single value from an effect
const s5 = Stream.fromEffect(Effect.succeed(42));

// Empty stream
const s6 = Stream.empty;

From effects (polling / repeating)

// Poll an effect on a schedule — useful for metrics, health checks, cache refresh
const samples = Stream.fromEffectSchedule(
	Effect.succeed(3),
	Schedule.spaced('30 seconds')
).pipe(Stream.take(10));

// Repeat an effect forever (no schedule delay)
const forever = Stream.fromEffectRepeat(Effect.succeed('tick'));

Paginated APIs

Stream.paginate drives cursor-based pagination. Return the current page and Option.some(nextCursor) or Option.none() to stop.

import * as Option from 'effect/Option';

const fetchAllPages = Stream.paginate(
	0, // initial cursor
	Effect.fn(function* (page) {
		yield* Effect.sleep('50 millis'); // simulate network
		const results = Array.from(
			{ length: 100 },
			(_, i) => `Job ${i + 1 + page * 100}`
		);
		const nextPage = page < 10 ? Option.some(page + 1) : Option.none();
		return [results, nextPage] as const;
	})
);

From async iterables

class IterError extends Schema.TaggedErrorClass<IterError>()('IterError', {
	cause: Schema.Defect()
}) {}

async function* generate() {
	yield 'a';
	yield 'b';
	yield 'c';
}

const letters = Stream.fromAsyncIterable(
	generate(),
	(cause) => new IterError({ cause })
);

From DOM events

// Direct event listener binding
const clicks = Stream.fromEventListener<PointerEvent>(button, 'click');

From callback-based APIs

Stream.callback gives you a Queue to push values into. Use Effect.acquireRelease inside to register/unregister listeners with guaranteed cleanup.

const callbackStream = Stream.callback<PointerEvent>(
	Effect.fn(function* (queue) {
		function onEvent(event: PointerEvent) {
			Queue.offerUnsafe(queue, event);
		}
		yield* Effect.acquireRelease(
			Effect.sync(() => button.addEventListener('click', onEvent)),
			() =>
				Effect.sync(() => button.removeEventListener('click', onEvent))
		);
	})
);

Options: { bufferSize?: number, strategy?: "sliding" | "dropping" | "suspend" }

From ReadableStream (DOM/Web)

const webStream = Stream.fromReadableStream({
	evaluate: () => response.body!,
	onError: (cause) => new MyError({ cause }),
	releaseLockOnEnd: false // default: cancels reader; true releases lock instead
});

From Node.js readable streams

import { NodeStream } from '@effect/platform-node';
import { Readable } from 'node:stream';

class NodeErr extends Schema.TaggedErrorClass<NodeErr>()('NodeErr', {
	cause: Schema.Defect()
}) {}

const nodeStream = NodeStream.fromReadable({
	evaluate: () => Readable.from(['Hello', ' ', 'world']),
	onError: (cause) => new NodeErr({ cause }),
	closeOnDone: true // true by default
});

Advanced constructors

// Unwrap: create a stream from an effect that returns a stream
const unwrapped = Stream.unwrap(Effect.succeed(Stream.make(1, 2, 3)));

// From a Channel directly
const fromChan = Stream.fromChannel(myChannel);

---

2. Transforming Streams

Pure transforms

// Per-element mapping (receives element and index)
stream.pipe(Stream.map((value, index) => value * 2));

// Filter elements
stream.pipe(Stream.filter((x) => x > 10));

// Windowing
stream.pipe(Stream.take(5)); // first 5 elements
stream.pipe(Stream.drop(3)); // skip first 3
stream.pipe(Stream.takeWhile((x) => x < 100));

Effectful transforms

// mapEffect with concurrency control
stream.pipe(
	Stream.mapEffect((order) => enrichOrder(order), { concurrency: 4 })
);

FlatMap

Transform each element into a stream and flatten. Supports concurrency.

Stream.make('US', 'CA', 'NZ').pipe(
	Stream.flatMap(
		(country) =>
			Stream.range(1, 50).pipe(
				Stream.map((i) => ({ id: `${country}_${i}`, country }))
			),
		{ concurrency: 2 }
	)
);

Accumulation

// Running accumulator — emits initial state plus each accumulated state
// Output: [0, 1, 3, 6]
Stream.make(1, 2, 3).pipe(Stream.scan(0, (acc, n) => acc + n));

// Effectful variant
Stream.make(1, 2, 3).pipe(
	Stream.scanEffect(0, (acc, n) => Effect.succeed(acc + n))
);

Grouping and batching

// Group into fixed-size chunks
stream.pipe(Stream.grouped(100));

// Group by size OR time window (whichever comes first)
stream.pipe(Stream.groupedWithin(100, '1 second'));

Rate control

// Debounce — emit only the latest element after a pause
stream.pipe(Stream.debounce('300 millis'));

// Throttle — control throughput
stream.pipe(
	Stream.throttle({
		cost: () => 1,
		units: 10,
		duration: '1 second',
		strategy: 'shape' // "shape" delays, "enforce" drops
	})
);

// Timeout — end stream if no element produced within duration
stream.pipe(Stream.timeout('5 seconds'));

// Timeout with fallback — switch to another stream on timeout
stream.pipe(
	Stream.timeoutOrElse({
		duration: '5 seconds',
		orElse: () => Stream.make(fallbackValue)
	})
);

Both timeout and timeoutOrElse are dual functions. timeout is implemented as timeoutOrElse with Stream.empty as the fallback. Non-finite durations return the stream unchanged; zero duration immediately switches to orElse.

Indexing and neighbors

stream.pipe(Stream.zipWithIndex); // [A, number]
stream.pipe(Stream.zipWithNext); // [A, Option<A>]
stream.pipe(Stream.zipWithPrevious); // [Option<A>, A]
stream.pipe(Stream.zipWithPreviousAndNext); // [Option<A>, A, Option<A>]

---

3. Consuming Streams

All run* methods return Effect values — the stream is only pulled when the effect is executed.

// Collect all elements into an array
const all = Stream.runCollect(stream);
// Effect<Array<A>, E, R>

// Run for side effects, ignore output
const drained = Stream.runDrain(stream);
// Effect<void, E, R>

// Execute effectful consumer per element
stream.pipe(Stream.runForEach((item) => Effect.log(`Got: ${item}`)));
// Effect<void, E, R>

// Fold to a single value (initial is a LazyArg — a thunk)
stream.pipe(
	Stream.runFold(
		() => 0,
		(acc, n) => acc + n
	)
);
// Effect<number, E, R>

// First / last element as Option
Stream.runHead(stream); // Effect<Option<A>, E, R>
Stream.runLast(stream); // Effect<Option<A>, E, R>

// Count / Sum helpers
Stream.runCount(stream); // Effect<number, E, R>
Stream.runSum(stream); // Effect<number, E, R> (stream must be Stream<number>)

// Consume with a Sink
stream.pipe(
	Stream.map((order) => order.totalCents),
	Stream.run(Sink.sum)
);

---

4. Encoding & Decoding (NDJSON / Msgpack)

Use Stream.pipeThroughChannel with codec channels from effect/unstable/encoding.

import { Ndjson, Msgpack } from 'effect/unstable/encoding';

Text decoding (split multi-byte characters)

To turn a byte stream into text, use Stream.decodeText (or Channel.decodeText) rather than hand-rolling new TextDecoder().decode(chunk) per chunk. These helpers decode with streaming enabled, so multi-byte UTF-8 characters split across Uint8Array chunk boundaries are reassembled correctly; per-chunk TextDecoder calls would corrupt characters that straddle a boundary.

byteStream.pipe(Stream.decodeText, Stream.runForEach(handleText));

NDJSON — string variants

// Decode: raw NDJSON string → parsed JSON objects
rawStream.pipe(
	Stream.pipeThroughChannel(Ndjson.decodeString()),
	Stream.runCollect
);

// Decode with schema validation
rawStream.pipe(
	Stream.pipeThroughChannel(Ndjson.decodeSchemaString(MySchema)()),
	Stream.runCollect
);

// Encode: objects → NDJSON strings
objectStream.pipe(
	Stream.pipeThroughChannel(Ndjson.encodeString()),
	Stream.runCollect
);

// Encode through schema (applies transforms like date formatting)
typedStream.pipe(
	Stream.pipeThroughChannel(Ndjson.encodeSchemaString(MySchema)()),
	Stream.runCollect
);

NDJSON — binary variants (Uint8Array)

For TCP sockets, file descriptors, etc.

binaryStream.pipe(Stream.pipeThroughChannel(Ndjson.decode())); // Uint8Array → objects
objectStream.pipe(Stream.pipeThroughChannel(Ndjson.encode())); // objects → Uint8Array

NDJSON options

// Ignore blank lines instead of raising NdjsonError
Stream.pipeThroughChannel(Ndjson.decodeString({ ignoreEmptyLines: true }));

Msgpack

Same API shape — replace Ndjson with Msgpack. Note that Msgpack.decodeSchema(schema) is curried: it returns a factory you must invoke (()) to get the Channel value passed to Stream.pipeThroughChannel, exactly like the NDJSON schema helpers.

const decoder = Msgpack.decodeSchema(
	Schema.Struct({
		id: Schema.Number,
		name: Schema.String
	})
)();

binaryStream.pipe(Stream.pipeThroughChannel(decoder), Stream.runCollect);

Realistic pipeline: decode → transform → re-encode

const pipeline = rawNdjsonStream.pipe(
	Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
	Stream.filter((entry) => entry.level === 'error'),
	Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
	Stream.runCollect
);

Handling encoding errors

Ndjson.NdjsonError has a kind field: "Pack" (encoding) or "Unpack" (decoding).

rawStream.pipe(
	Stream.pipeThroughChannel(Ndjson.decodeString()),
	Stream.catchTag('NdjsonError', (err) =>
		Stream.succeed({ recovered: true, kind: err.kind })
	),
	Stream.runCollect
);

---

5. Error Handling

catchTag / catchTags

Recover from specific tagged errors, producing a fallback stream.

stream.pipe(
	Stream.catchTag('NetworkError', (err) => Stream.succeed(fallbackValue))
);

retry

Retry a failing stream with a schedule. The stream restarts from the beginning on each retry.

stream.pipe(Stream.retry(Schedule.recurs(3)));

// With exponential backoff
stream.pipe(Stream.retry(Schedule.exponential('100 millis')));

orElseIfEmpty / orElseSucceed

// Provide a default stream if the source emits nothing
stream.pipe(Stream.orElseIfEmpty(() => Stream.make(defaultValue)));

// Provide a single fallback value when the source fails
stream.pipe(Stream.orElseSucceed((error) => defaultValue));

---

6. Concurrency & Merging

merge

Interleave elements from two streams concurrently in arrival order.

Stream.merge(streamA, streamB);
Stream.merge(streamA, streamB, { haltStrategy: 'left' }); // stop when left ends
// HaltStrategy: "left" | "right" | "both" | "either"

mergeAll

Merge many streams concurrently. The streams are passed as a single iterable, followed by the options.

Stream.mergeAll([streamA, streamB, streamC], {
	concurrency: 4
});

interleave

Deterministically alternate elements from two streams (round-robin).

Stream.interleave(left, right);

// Custom interleave pattern via boolean decider stream
Stream.interleaveWith(left, right, Stream.make(true, false, false, true));

mergeResult

Tag values from two streams: left as Result.succeed, right as Result.fail.

Stream.mergeResult(left, right); // Stream<Result<LeftA, RightA>>

mergeEffect

Run a background effect concurrently with a stream; keep the stream's elements.

stream.pipe(Stream.mergeEffect(Effect.log('background task')));

zipWith

Pair elements from two streams positionally.

Stream.zipWith(numbersStream, labelsStream, (n, label) => `${label}: ${n}`);

broadcast

PubSub-backed multicast: the source is consumed once and fanned out to every subscriber. Returns a scoped effect, and the producer starts immediately — it does not wait for subscribers to attach.

Effect.scoped(
	Effect.gen(function* () {
		const shared = yield* stream.pipe(
			Stream.broadcast({ capacity: 16, replay: 3 })
		);
		// Each consumer subscribes independently. Because the producer starts
		// immediately, a late subscriber only sees values still held in `replay`.
		const fiberA = yield* Stream.runCollect(shared).pipe(Effect.forkChild);
		const fiberB = yield* Stream.runCollect(shared).pipe(Effect.forkChild);
		// ...
	})
);

Options: { capacity: number | "unbounded", strategy?: "sliding" | "dropping" | "suspend", replay?: number }

Because the producer starts immediately, subscribers that attach after the source has already emitted will miss earlier values unless replay is configured (and replay only retains the most recent N values — it is not a full log). For a fixed, known set of consumers, prefer broadcastN: it subscribes all downstream streams before starting the source, so none of them miss values.

broadcastN

Fixed-fanout multicast (added in beta.68). Produces a tuple of n streams; the source starts only after all n downstream streams have been subscribed, so every consumer sees the full sequence without needing replay. If a downstream stream is interrupted, it unsubscribes and no longer contributes backpressure.

Effect.scoped(
	Effect.gen(function* () {
		const [left, right] = yield* Stream.make(1, 2, 3).pipe(
			Stream.broadcastN({ n: 2, capacity: 8 })
		);

		const [leftValues, rightValues] = yield* Effect.all(
			[Stream.runCollect(left), Stream.runCollect(right)],
			{ concurrency: 'unbounded' }
		);
		// leftValues and rightValues each === [1, 2, 3]
	})
);

Options: { n: number, capacity: number | "unbounded", strategy?: "sliding" | "dropping" | "suspend", replay?: number }

share

Like broadcast but subscribes lazily when the first consumer starts, keeps upstream alive while consumers exist.

const shared = yield* stream.pipe(Stream.share({ capacity: 16 }));

---

7. Resource Safety

scoped

Run a stream that requires Scope in a managed scope, ensuring finalizers run when the stream completes.

const safeStream = Stream.scoped(
	Stream.fromEffect(
		Effect.acquireRelease(
			Effect.log('acquire').pipe(Effect.as('resource')),
			() => Effect.log('release')
		)
	)
);
// Stream<string, never, never> — Scope is eliminated

As of beta.69, Stream.scoped provides its managed scope to the pull effects as well — including effects created by Stream.fromEffect and by sequential Stream.mapEffect. So Effect.acquireRelease finalizers used inside those pulls run when the stream completes, not leaked until the outer program ends.

unwrap

Create a stream from an effect that produces a stream. The outer effect runs once; the inner stream is then consumed.

const stream = Stream.unwrap(
	Effect.gen(function* () {
		const config = yield* loadConfig;
		return Stream.fromIterable(config.items);
	})
);

callback with acquireRelease

The Stream.callback constructor accepts a scoped effect, so you can register and unregister resources:

Stream.callback<Event>(
	Effect.fn(function* (queue) {
		yield* Effect.acquireRelease(
			Effect.sync(() =>
				emitter.on('data', (e) => Queue.offerUnsafe(queue, e))
			),
			() => Effect.sync(() => emitter.removeAllListeners('data'))
		);
	})
);

---

8. Piping Through Channels

Stream.pipeThroughChannel connects a stream to a Channel for encode/decode, compression, framing, etc.

// pipeThroughChannel: upstream errors flow into the channel
stream.pipe(Stream.pipeThroughChannel(myChannel));

// pipeThroughChannelOrFail: upstream errors preserved alongside channel errors
stream.pipe(Stream.pipeThroughChannelOrFail(myChannel));

---

Key Patterns

Pagination → transform → consume

const pipeline = Stream.paginate(0, fetchPage).pipe(
	Stream.mapEffect(enrichItem, { concurrency: 8 }),
	Stream.filter((item) => item.isValid),
	Stream.grouped(50),
	Stream.runForEach((batch) => writeBatch(batch))
);

Event stream → debounce → side effect

const autosave = Stream.fromEventListener(input, 'input').pipe(
	Stream.debounce('500 millis'),
	Stream.mapEffect((e) => saveDocument(e.target.value)),
	Stream.runDrain
);

Decode NDJSON file → filter → re-encode

const filterErrors = fileStream.pipe(
	Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
	Stream.filter((entry) => entry.level === 'error'),
	Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
	Stream.runCollect
);

Retry with backoff

const resilient = unreliableStream.pipe(
	Stream.retry(Schedule.exponential('100 millis').pipe(Schedule.take(5))),
	Stream.runCollect
);

Schedule.take(n) bounds an unbounded schedule to n recurrences (Schedule.compose is not exported in beta.74). To log full retry metadata without changing the schedule's behavior, add Schedule.tap, whose callback receives { attempt, input, output, duration, elapsed }:

const monitored = Schedule.exponential('100 millis').pipe(
	Schedule.take(5),
	Schedule.tap((meta) =>
		Effect.log(
			`attempt ${meta.attempt}, next delay ${meta.duration}, elapsed ${meta.elapsed}`
		)
	)
);

Common Mistakes

1. Forgetting runFold initial is a thunk — Stream.runFold(() => 0, f) not Stream.runFold(0, f)
2. Using Stream.acquireRelease when it doesn't exist — use Stream.scoped + Effect.acquireRelease or Stream.callback with Effect.acquireRelease instead
3. Not specifying onError for fromAsyncIterable / fromReadableStream — these require an error mapper
4. Assuming retry resumes — Stream.retry restarts the entire stream from the beginning on each retry
5. Ignoring haltStrategy on merge — default is "both" (wait for both to end); use "either" to stop as soon as one ends