Consuming Streams

A Stream is a description — it does nothing until you run it. The run* family are the destructors that pull values through the pipeline and hand you back an Effect. Which one you choose depends on what you want out the other end: every element, all elements collected, an accumulated total, or just the side effects.

import { Effect, Stream } from "effect"

interface Order {
  readonly id: string
  readonly grandTotalCents: number
  readonly priority: "normal" | "high"
}

const orders: Stream.Stream<Order> = Stream.make(
  { id: "ord_1", grandTotalCents: 4_500, priority: "normal" },
  { id: "ord_2", grandTotalCents: 28_000, priority: "high" },
  { id: "ord_3", grandTotalCents: 9_900, priority: "normal" }
)

// `runForEach` runs an effectful consumer for *every* element and discards the
// elements. Use it when the value of running is the side effect (logging,
// writing, publishing). It returns Effect<void, E, R>.
export const logEach = orders.pipe(
  Stream.runForEach((order) =>
    Effect.logInfo(`order ${order.id}: $${(order.grandTotalCents / 100).toFixed(2)}`)
  )
)

runCollect, runDrain, runFold

These three cover the bulk of consumption. Pick based on the shape of the result you need:

import { Effect, Stream } from "effect"

const totals: Stream.Stream<number> = Stream.make(4_500, 28_000, 9_900)

// `runCollect` materialises every element into an immutable array.
// Only use it when the stream is bounded and fits comfortably in memory.
export const collected: Effect.Effect<Array<number>> = Stream.runCollect(totals)

// `runDrain` runs the stream purely for its effects and throws the output away.
// It is the right choice for infinite or very large streams whose work happens
// in upstream `tap`/`mapEffect` operators.
export const drained: Effect.Effect<void> = Stream.runDrain(totals)

// `runFold` reduces the stream to a single accumulated value, pulling one
// element at a time — constant memory regardless of stream length. The initial
// state is a thunk (`() => 0`) so it can be re-evaluated safely.
export const sum: Effect.Effect<number> = totals.pipe(
  Stream.runFold(() => 0, (acc, cents) => acc + cents)
)

runFold is the streaming equivalent of Array.reduce, but it never holds the whole stream in memory — it threads the accumulator through as each element arrives. For an effectful reducer (e.g. one that writes to a database per step), use Stream.runFoldEffect.

Edge elements and counting

When you only care about the first or last element, or a simple count, there are dedicated destructors that short-circuit appropriately:

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

const orders = Stream.make(
  { id: "ord_1", priority: "normal" as const },
  { id: "ord_2", priority: "high" as const },
  { id: "ord_3", priority: "high" as const }
)

// `runHead` pulls only the first element (then stops the stream) as an Option.
export const first: Effect.Effect<Option.Option<{ id: string }>> =
  Stream.runHead(orders)

// `runLast` returns the final element as an Option.
export const last = Stream.runLast(orders)

// `runCount` counts elements without retaining them.
export const howMany: Effect.Effect<number> = Stream.runCount(orders)

// `runForEachWhile` consumes until the predicate returns `false`, letting you
// stop early based on the data itself.
export const untilHigh = orders.pipe(
  Stream.runForEachWhile((order) =>
    Effect.succeed(order.priority !== "high")
  )
)

Consuming through a Sink

For anything more elaborate than the built-in destructors — collecting the first N elements, summing while leaving leftovers, fanning out to a queue — describe the consumption as a Sink and run it with Stream.run:

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

const cents = Stream.make(4_500, 28_000, 9_900)

// `Stream.run` consumes the stream with any Sink. `Sink.sum` adds up the
// numbers; the result is an Effect<number>.
export const total: Effect.Effect<number> = cents.pipe(Stream.run(Sink.sum))

run is the most general destructor — runCollect, runFold, and the rest are really specialised sinks. Reach for an explicit Sink when you want to reuse a consumption strategy or combine several. The full vocabulary lives on the Sink page.

Every run* is a Sink underneath

The run* destructors are convenience wrappers over Stream.run(someSink). runCollect is run with a collecting sink, runFold with a folding sink, and so on. If you find yourself reaching for a destructor that does not exist, it almost certainly exists as a Sink you can pass to Stream.run.

Core runners

Every destructor below returns an Effect that drives the stream to completion (or to an early stop). Run that effect with Effect.runPromise, by yield*-ing it inside Effect.gen, or by composing it into a larger program.

run

The most general destructor: consume the stream with any Sink. All other runners are specialisations of this one.

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

const program = Stream.make(1, 2, 3).pipe(Stream.run(Sink.sum))

Effect.runPromise(program)
// => 6

runCollect

Runs the stream and collects every element into an immutable array. Only safe for bounded streams that fit in memory.

import { Effect, Stream } from "effect"

Effect.runPromise(Stream.runCollect(Stream.make(1, 2, 3, 4, 5)))
// => [1, 2, 3, 4, 5]

runDrain

Runs the stream purely for its effects, discarding every emitted element. The right choice for infinite or very large streams whose work happens in upstream operators.

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

const program = Stream.make(1, 2, 3).pipe(
  Stream.mapEffect((n) => Console.log(`Processing: ${n}`)),
  Stream.runDrain
)

Effect.runPromise(program)
// => logs Processing: 1 / Processing: 2 / Processing: 3, result is void

runForEach

Runs an effectful consumer for every element and discards the elements. Use it when the value is the side effect (logging, writing, publishing).

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

const program = Stream.make(1, 2, 3).pipe(
  Stream.runForEach((n) => Console.log(`Processing: ${n}`))
)

Effect.runPromise(program)
// => logs Processing: 1 / Processing: 2 / Processing: 3

runForEachWhile

Like runForEach, but the callback returns an Effect<boolean>; consumption stops as soon as it yields false. Lets you stop early based on the data.

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

const program = Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.runForEachWhile((n) =>
    Console.log(`Processing: ${n}`).pipe(Effect.as(n < 3))
  )
)

Effect.runPromise(program)
// => logs Processing: 1 / Processing: 2 / Processing: 3 (then stops)

runForEachArray

Consumes the stream chunk-by-chunk, passing each non-empty array to the callback. Avoids per-element overhead when you can process a whole batch at once.

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

const program = Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.runForEachArray((chunk) =>
    Console.log(`Processing chunk: ${chunk.join(", ")}`)
  )
)

Effect.runPromise(program)
// => logs Processing chunk: 1, 2, 3, 4, 5

runCount

Runs the stream and returns the number of elements emitted, without retaining them.

import { Effect, Stream } from "effect"

Effect.runPromise(Stream.runCount(Stream.make(1, 2, 3, 4, 5)))
// => 5

runSum

Runs a Stream<number> and returns the numeric sum of its elements.

import { Effect, Stream } from "effect"

Effect.runPromise(Stream.runSum(Stream.make(1, 2, 3)))
// => 6

runFold

Reduces the stream to a single value with a pure reducer, pulling one element at a time (constant memory). The initial state is a thunk so it can be re-evaluated safely.

import { Effect, Stream } from "effect"

const program = Stream.runFold(
  Stream.make(1, 2, 3),
  () => 0,
  (acc, n) => acc + n
)

Effect.runPromise(program)
// => 6

runFoldEffect

Like runFold, but the reducer returns an Effect — use it when each step performs work (e.g. a database write or validation).

import { Effect, Stream } from "effect"

const program = Stream.runFoldEffect(
  Stream.make(1, 2, 3),
  () => 0,
  (acc, n) => Effect.succeed(acc + n)
)

Effect.runPromise(program)
// => 6

runHead

Pulls only the first element (then stops the stream) as an Option.

import { Effect, Stream } from "effect"

Effect.runPromise(Stream.runHead(Stream.make(1, 2, 3)))
// => Option.some(1)

Effect.runPromise(Stream.runHead(Stream.empty))
// => Option.none()

runLast

Returns the final element as an Option. The effect waits for the stream to complete before producing a value.

import { Effect, Stream } from "effect"

Effect.runPromise(Stream.runLast(Stream.make(1, 2, 3)))
// => Option.some(3)

String and binary collection

When a stream produces text or bytes, these destructors concatenate the chunks into a single value rather than an array.

mkString

Concatenates all emitted strings into a single string.

import { Effect, Stream } from "effect"

Effect.runPromise(Stream.mkString(Stream.make("Hello", " ", "World", "!")))
// => "Hello World!"

mkUint8Array

Concatenates the stream’s Uint8Array chunks into one contiguous Uint8Array. Handy when collecting the body of a binary stream.

import { Effect, Stream } from "effect"

const program = Stream.make(
  new Uint8Array([1, 2]),
  new Uint8Array([3, 4])
).pipe(Stream.mkUint8Array)

Effect.runPromise(program).then((bytes) => console.log([...bytes]))
// => [1, 2, 3, 4]

Bridging to other sinks of data

Beyond running to a single value, a Stream can be handed off to the outside world — pulled manually, exposed as a Web ReadableStream, iterated with for await, or fanned out into a Queue/PubSub for concurrent consumers. Most of these are scoped: they require a Scope and the underlying fiber/queue is torn down when the scope closes.

toPull

Returns a scoped Pull you can invoke repeatedly to consume the stream’s output chunks by hand. The pull fails with Cause.Done when the stream ends and with the stream’s error on failure.

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

const program = Effect.scoped(
  Effect.gen(function* () {
    const pull = yield* Stream.toPull(Stream.make(1, 2, 3))
    const chunk = yield* pull // a NonEmptyReadonlyArray<number>
    yield* Console.log(chunk)
  })
)

Effect.runPromise(program)
// => [1, 2, 3]

Pulling manually is the lowest-level way to drive a stream. You typically loop on the pull until it fails with Cause.Done; everything higher up (runForEach, the queue/iterable bridges) is built on top of it.

toReadableStream

Converts a Stream<A, E> (with no service requirements) into a Web platform ReadableStream. Returns the ReadableStream directly, not an Effect.

import { Stream } from "effect"

const readable = Stream.toReadableStream(Stream.make(1, 2, 3))
const reader = readable.getReader()
// reader.read() => Promise<{ done: false, value: 1 }>, then 2, 3, then done

An optional { strategy } argument accepts a standard QueuingStrategy for backpressure control.

toReadableStreamWith

Like toReadableStream, but you supply a Context for streams that require services. Use this when the stream’s R is not never.

import { Context, Stream } from "effect"

const readable = Stream.toReadableStreamWith(
  Stream.make(1, 2, 3, 4, 5),
  Context.empty()
)
// => ReadableStream<number>

toReadableStreamEffect

The effectful counterpart: builds the ReadableStream inside an Effect, capturing the current services automatically. Reach for this when you are already in Effect.gen and the stream needs R.

import { Effect, Stream } from "effect"

const program = Effect.gen(function* () {
  const readable = yield* Stream.toReadableStreamEffect(Stream.make(1, 2, 3))
  return readable instanceof ReadableStream
})

Effect.runPromise(program)
// => true

toAsyncIterable

Converts a Stream<A, E> (no requirements) into an AsyncIterable<A> for for await...of consumption.

import { Stream } from "effect"

const collect = async () => {
  const values: Array<number> = []
  for await (const value of Stream.toAsyncIterable(Stream.make(1, 2, 3))) {
    values.push(value)
  }
  return values
}

collect()
// => [1, 2, 3]

toAsyncIterableWith

Like toAsyncIterable, but takes an explicit Context so streams with service requirements can be iterated.

import { Context, Stream } from "effect"

const iterable = Stream.toAsyncIterableWith(
  Stream.make(1, 2, 3),
  Context.empty()
)
// => AsyncIterable<number>

toAsyncIterableEffect

Builds the AsyncIterable inside an Effect, capturing the current services — the effectful variant for use inside Effect.gen.

import { Effect, Stream } from "effect"

const program = Effect.gen(function* () {
  const iterable = yield* Stream.toAsyncIterableEffect(Stream.make(1, 2, 3))
  return yield* Effect.promise(async () => {
    const collected: Array<number> = []
    for await (const value of iterable) collected.push(value)
    return collected
  })
})

Effect.runPromise(program)
// => [1, 2, 3]

runIntoQueue

Runs the stream, offering each element to a queue you already own, and ends the queue with Cause.Done when the stream completes. The queue’s error channel must accept E | Cause.Done. Typically forked so producers and consumers run concurrently.

import { Cause, Effect, Queue, Stream } from "effect"

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number, Cause.Done>(4)

  yield* Effect.forkChild(
    Stream.runIntoQueue(Stream.fromIterable([1, 2, 3]), queue)
  )

  const values = [
    yield* Queue.take(queue),
    yield* Queue.take(queue),
    yield* Queue.take(queue)
  ]
  const done = yield* Effect.flip(Queue.take(queue))

  return { values, done }
})
// => { values: [1, 2, 3], done: Cause.Done }

toQueue

Creates a scoped Queue.Dequeue that the stream feeds in the background. Elements are offered as the stream runs; completion signals Cause.Done, failures fail the queue, and the queue is shut down when the scope closes.

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

const program = Effect.scoped(
  Effect.gen(function* () {
    const queue = yield* Stream.toQueue(
      Stream.fromIterable([1, 2, 3]),
      { capacity: 8 }
    )
    return yield* Queue.takeBetween(queue, 1, 3)
  })
)
// => takes 1..3 elements from the queue

Pass { capacity: "unbounded" } for an unbounded buffer, or a numeric capacity with an optional strategy of "dropping" | "sliding" | "suspend" to control backpressure.

runIntoPubSub

Runs the stream, publishing every element into a PubSub you already own. The optional shutdownOnEnd (default off here) shuts the PubSub down when the stream ends.

import { Console, Effect, PubSub, Stream } from "effect"

const program = Effect.scoped(
  Effect.gen(function* () {
    const pubsub = yield* PubSub.unbounded<number>()
    const subscription = yield* PubSub.subscribe(pubsub)

    yield* Stream.runIntoPubSub(Stream.fromIterable([1, 2]), pubsub)

    yield* Console.log(yield* PubSub.take(subscription))
    yield* Console.log(yield* PubSub.take(subscription))
  })
)

Effect.runPromise(program)
// => 1
// => 2

toPubSub

Creates a scoped PubSub of emitted values that the stream feeds, for concurrent fan-out to multiple subscribers. shutdownOnEnd defaults to true.

import { Console, Effect, PubSub, Stream } from "effect"

const program = Effect.scoped(
  Effect.gen(function* () {
    const pubsub = yield* Stream.fromArray([1, 2]).pipe(
      Stream.toPubSub({ capacity: 8 })
    )
    const subscription = yield* PubSub.subscribe(pubsub)
    yield* Console.log(yield* PubSub.take(subscription))
  })
)
// => 1

Capacity options mirror toQueue: { capacity: "unbounded" } (with optional replay) or a numeric capacity with an optional strategy and replay.

toPubSubTake

Like toPubSub, but publishes Take values instead of raw elements, so subscribers can observe the stream’s end and failure signals (not just data).

import { Console, Effect, PubSub, Stream } from "effect"

const program = Effect.scoped(
  Effect.gen(function* () {
    const pubsub = yield* Stream.fromArray([1, 2, 3]).pipe(
      Stream.toPubSubTake({ capacity: 8 })
    )
    const subscription = yield* PubSub.subscribe(pubsub)
    const take = yield* PubSub.take(subscription) // Take<number, never>
    if (Array.isArray(take)) {
      yield* Console.log(take)
    }
  })
)
// => [1, 2, 3]

Scoped bridges need a Scope

toPull, toQueue, toPubSub, and toPubSubTake all return effects that require Scope.Scope. Wrap them in Effect.scoped (or run inside an existing scope) so the background fiber and the queue/pubsub are released when you are done. runIntoQueue and runIntoPubSub, by contrast, write into a resource you already manage and do not introduce a scope themselves.