Streaming

A Stream<A, E, R> is an effectful, pull-based sequence of values. Where an Effect<A, E, R> produces a single result (or fails with E), a Stream produces zero, one, or many A values over time — and may also fail with E or require services R. Streams can be finite (the lines of a file) or infinite (a polling loop, a socket, a clock tick), and they evaluate lazily: nothing runs until you run the stream with a destructor like Stream.runForEach or a Sink.

Because streaming is pull-based, a Stream only does as much work as its consumer demands. This gives you backpressure for free, bounded memory over huge or unbounded sources, and the same composability you get from Effect — typed errors, resource safety via Scope, and structured concurrency.

import { Effect, Stream } from "effect"

// A pipeline reads numbers, keeps the even ones, doubles them, and logs each.
// Nothing executes until `runForEach` pulls values through the pipeline.
const program = Stream.range(1, 10).pipe(
  Stream.filter((n) => n % 2 === 0),
  Stream.map((n) => n * 2),
  Stream.runForEach((n) => Effect.logInfo(`value: ${n}`))
)

Effect.runFork(program)

The streaming triad

Effect’s streaming model is built from three cooperating modules. You will spend almost all of your time with Stream; Sink shows up when you want a reusable consumer, and Channel is the low-level engine you rarely touch directly.

• Stream<A, E, R> — the producer. A pull-based source of A values that you create, transform, and eventually run. This is the module you reach for first.
• Sink<A, In, L, E, R> — the consumer. A composable description of how a stream is folded into a single result A (a sum, a count, the first element, a collected array), consuming In elements and possibly leaving L leftovers. Run one with Stream.run.
• Channel<…> — the transformer. The lower-level primitive that both Stream and Sink are implemented on top of. Channels read and write chunks of values and are what make operators like Stream.pipeThroughChannel (used for encoding) possible.

Put together, they form a pipeline: a Stream produces values, channels transform them, and a Sink consumes them into a result.

import { Effect, Sink, Stream } from "effect"

// create -> transform -> run (into a Sink)
const total = Stream.range(1, 5).pipe(
  Stream.map((n) => n * n), // 1, 4, 9, 16, 25
  Stream.run(Sink.sum) // fold all values into a single number
)

Effect.runPromise(total).then(console.log)
// => 55

Tip

Stream.runForEach, Stream.runCollect, Stream.runDrain, and friends are convenience destructors built on top of Stream.run with a built-in Sink. Reach for an explicit Sink only when you want to reuse or compose the consuming logic.

Where to go next

Creating streams

Build streams from iterables, effects, paginated APIs, callbacks, and Node readables. Read more

Transforming streams

map, flatMap, filter, mapEffect, scan, grouping, and concurrency. Read more

Consuming streams

Run streams with runForEach, runCollect, runDrain, and runFold. Read more

Error handling

Recover, retry, and time out within streams. Read more

Encoding

Decode and encode structured data with Ndjson and Msgpack. Read more

Sink

Describe how a stream is consumed into a single result. Read more

When to reach for a Stream

Use a Stream when a source naturally yields many values over time and you want to process them incrementally instead of materialising everything in memory: log lines, paginated HTTP responses, websocket messages, file chunks, database cursors, or a polling loop. For a single asynchronous result, a plain Effect is the right tool.