# 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.

```ts
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:

```ts
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:

```ts
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](https://effect.plants.sh/streaming/sink/) and run it with `Stream.run`:

```ts
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](https://effect.plants.sh/streaming/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](https://effect.plants.sh/streaming/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](https://effect.plants.sh/streaming/sink/).
All other runners are specialisations of this one.

```ts
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.

```ts
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.

```ts
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).

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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).

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream).
Returns the `ReadableStream` directly, not an `Effect`.

```ts
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`.

```ts
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`.

```ts
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.

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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).

```ts
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.