Effect Stream

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 ~/.local/share/opencode/repos/github.com/Effect-TS/effect@main/. 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.

Use a stream when values are naturally many-valued and ordered over time. For one effect repeated only for its side effects, prefer Effect.repeat with Schedule; see the effect-scheduling skill.

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

Source chooser:

• Values/tests: Stream.make or Stream.fromIterable.
• Callback boundary consumed by one worker: private Queue + Stream.fromQueue.
• Broadcast events: private PubSub + Stream.fromPubSub.
• Current value plus changes: SubscriptionRef.
• Schedule outputs/ticks: Stream.fromSchedule.
• Paginated pull API: Stream.paginate.
• Effect that first reads services/config: Stream.unwrap.
• Async iterable/platform source: Stream.fromAsyncIterable when no native Effect source exists.

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.TaggedError<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.TaggedError<NodeErr>()('NodeErr', {
	cause: Schema.Defect()
}) {}

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

For bounded collection of a Node readable, use NodeStream.toString, NodeStream.toArrayBuffer, or NodeStream.toUint8Array with maxBytes. The limit is inclusive: maxBytes: 0 permits an empty stream but fails through onError as soon as any byte is received, and the consumer destroys the readable on interruption or failure.

const text = NodeStream.toString(() => readable, {
	maxBytes: 0,
	onError: (cause) => new NodeErr({ cause })
});

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

Choose map for pure work, mapEffect for effectful work, and bounded mapEffect(..., { concurrency }) for parallel work. Add unordered: true only when output order is irrelevant. Use flatMap for zero/many outputs, filter/filterEffect for selection, and mapAccum/mapAccumEffect for stateful transforms.

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.

Use runForEach for side-effecting consumers, runDrain when values are irrelevant, and runFold for bounded aggregation. Reserve runCollect for tests and known-finite, memory-bounded streams; never collect an unbounded production event stream. In tests, prefer take(n) + runCollect.

For stream tests, use fromIterable for finite fixtures, empty for no events, and a test-owned Queue plus fromQueue when the test must drive events interactively. Coordinate with Deferred, Queue, Latch, or TestClock, never real sleeps.

// 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 / SchemaBinary)

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

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

Schema-derived binary frames (rc.112)

import { Stream } from 'effect';
import * as Schema from 'effect/Schema';
import { SchemaBinary } from 'effect/unstable/encoding';

class Reading extends Schema.Class<Reading>('Reading')({
	id: Schema.String,
	value: Schema.Number
}) {}

const roundTrip = Stream.make(new Reading({ id: 'sensor-1', value: 21 })).pipe(
	Stream.pipeThroughChannel(SchemaBinary.encode(Reading)()),
	Stream.pipeThroughChannel(SchemaBinary.decode(Reading, { maxFrameSize: 1024 })()),
	Stream.runCollect
);

encode(schema)() and decode(schema)() are channel factories; use them with Stream.pipeThroughChannel. Input chunks can split or concatenate frames. The decoder retains completed values before a later failure and fails with Schema.SchemaError on an incomplete final frame. Decoding/encoding services from schema transformations remain in the channel requirements. maxFrameSize limits decoding only. Keep paired schemas/options compatible, and copy any arena-backed encoded bytes that must survive later writes. For synchronous framing/dictionaries use SchemaBinary.encoder and parser; duplex adapts a bidirectional channel. See effect-schema-composition for codec layout, fingerprints, and ownership. Use RpcSerialization.layerSchemaBinary for RPC.

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')));

Schedules can be effectful and fail. Stream.retry includes the schedule's error in the resulting stream error channel; Effect.schedule and Effect.scheduleFrom likewise union schedule errors into their error channels. Recover or map that error explicitly rather than assuming only the repeated operation can fail. For sequential schedule composition, use Schedule.concat / Schedule.concatResult; the former andThen names were removed.

Execution-plan attempt events

Stream.withExecutionPlan accepts an onEvent observer for attempt-level logs and metrics:

const planned = stream.pipe(
	Stream.withExecutionPlan(plan, {
		onEvent: (event) => Effect.log('execution plan event', event)
	})
);

Events are AttemptStart, AttemptSuccess, or AttemptFailure. Every start has one terminal event, failures carry the full Cause, and attempt is cumulative while stepAttempt is 1-based within a step. The observer must have a never error channel; an observer defect is isolated from the attempt outcome and does not leave events unpaired. If a downstream consumer intentionally stops pulling early, the truncated attempt is reported as successful.

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

Buffer policy

Prefer natural backpressure. Add Stream.buffer only to deliberately decouple producer and consumer: "suspend" backpressures when full, "dropping" drops new values, and "sliding" drops old values to retain the latest. Avoid capacity: "unbounded" unless growth is bounded elsewhere and documented.

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

Long-lived service consumers

Expose Stream values from service interfaces while keeping producer Queue, PubSub, and mutable state private. Own long-lived consumers in a layer and ordinarily run them with stream.pipe(Stream.runForEach(handle), Effect.forkScoped) so layer shutdown interrupts the consumer.

forkScoped provides lifetime supervision, not failure recovery or restart supervision. A failed child fiber does not automatically fail its parent or restart itself. If consumer failure must stop the application, restart with policy, or be reported, explicitly join/monitor the fiber or install a supervisor at the owning runtime boundary. Preserve interruption as shutdown; do not blanket-catch causes and turn interruption into a retry loop.

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.upTo({ times: 5 }))
	),
	Stream.runCollect
);

Schedule.upTo({ times: n }) bounds an unbounded schedule to n recurrences. 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.upTo({ times: 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
6. Assuming forkScoped supervises failures — it scopes lifetime only. Explicitly monitor/restart/report long-lived consumers according to the owning service policy.
7. Collecting open streams — use runForEach/runDrain in production and take(n) + runCollect for finite tests.