Stream

Describes effectful sources that emit values over time.

A Stream<A, E, R> can emit many A values, fail with E, and require services R while it is being consumed. Streams are useful for data that is pulled in steps, such as values from collections, queues, pubsubs, schedules, callbacks, async iterables, or platform streams. The APIs here cover the full stream lifecycle: create a stream, transform or combine it, control buffering and timing, handle failures, and finally consume it.

249 exports Since v2.0.0 Source

Accessors

service

Since v4.0.0

Accesses a service from the context and emits it as a single element.

Signature

declare function service<I, S>(service: Key<I, S>): Stream<S, never, I>

Example Accessing a service as a stream

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

class Greeter extends Context.Service<Greeter, {
  readonly greet: (name: string) => string
}>()("Greeter") {}

const stream = Stream.service(Greeter).pipe(
  Stream.map((greeter) => greeter.greet("World"))
)

await Effect.runPromise(
  stream.pipe(
    Stream.provideService(Greeter, {
      greet: (name) => `Hello, ${name}!`
    }),
    Stream.runCollect
  )
) // => ["Hello, World!"]

serviceOption

Since v4.0.0

Optionally accesses a service from the context and emits the result as a single element.

When to use

Use when you need a stream that emits an optional service from the context without requiring that service to be present.

Signature

declare function serviceOption<I, S>(service: Key<I, S>): Stream<Option<S>>

Example Accessing an optional service as a stream

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

class Greeter extends Context.Service<Greeter, {
  readonly greet: (name: string) => string
}>()("Greeter") {}

const stream = Stream.serviceOption(Greeter).pipe(
  Stream.map((maybeGreeter) =>
    Option.match(maybeGreeter, {
      onNone: () => "No greeter",
      onSome: (greeter) => greeter.greet("World")
    })
  )
)

await Effect.runPromise(
  stream.pipe(
    Stream.provideService(Greeter, {
      greet: (name) => `Hello, ${name}!`
    }),
    Stream.runCollect
  )
) // => ["Hello, World!"]

Accumulation

accumulate

Since v2.0.0

Accumulates elements into a growing array, emitting the cumulative array for each input chunk.

Signature

declare function accumulate<A, E, R>(self: Stream<A, E, R>): Stream<[A, ...Array<A>], E, R>

Example Accumulating stream elements

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const accumulated = yield* Stream.runCollect(
    Stream.fromArray([1, 2, 3]).pipe(
      Stream.rechunk(1),
      Stream.accumulate
    )
  )
  accumulated // => [ [ 1 ], [ 1, 2 ], [ 1, 2, 3 ] ]
})

await Effect.runPromise(program)

collect

Since v4.0.0

Collects all elements into an array and emits it as a single element.

Signature

declare function collect<A, E, R>(self: Stream<A, E, R>): Stream<Array<A>, E, R>

Example Collecting values into a stream element

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3)

const program = Effect.gen(function*() {
  const collected = yield* stream.pipe(Stream.collect, Stream.runCollect)
  collected[0] // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

scan

Since v2.0.0

Accumulates state across the stream, emitting the initial state and each updated state.

Signature

declare const scan: {
  <S, A>(initial: LazyArg<S>, f: (s: S, a: A) => S): <E, R>(self: Stream<A, E, R>) => Stream<S, E, R>;
  <A, E, R, S>(self: Stream<A, E, R>, initial: LazyArg<S>, f: (s: S, a: A) => S): Stream<S, E, R>;
}

Example Scanning stream state

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.scan(() => 0, (acc, n) => acc + n),
    Stream.runCollect
  )
  values // => [ 0, 1, 3, 6 ]
})

await Effect.runPromise(program)

scanEffect

Since v2.0.0

Accumulates state effectfully and emits the initial state plus each accumulated state.

Signature

declare const scanEffect: {
  <S, A, E2, R2>(initial: LazyArg<S>, f: (s: S, a: A) => Effect<S, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<S, E2 | E, R2 | R>;
  <A, E, R, S, E2, R2>(self: Stream<A, E, R>, initial: LazyArg<S>, f: (s: S, a: A) => Effect<S, E2, R2>): Stream<S, E | E2, R | R2>;
}

Example Effectfully scanning stream state

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const states = yield* Stream.make(1, 2, 3).pipe(
    Stream.scanEffect(() => 0, (sum, n) => Effect.succeed(sum + n)),
    Stream.runCollect
  )
  states // => [ 0, 1, 3, 6 ]
})
await Effect.runPromise(program)

Aggregation

aggregate

Since v2.0.0

Aggregates elements using the provided sink and emits each sink result as a stream element.

Details

The stream runs the upstream and downstream in separate fibers, so the sink can keep consuming input while downstream is busy processing the previous output.

Signature

declare const aggregate: {
  <B, A, A2, E2, R2>(sink: Sink<B, A | A2, A2, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<B, E2 | E, R2 | R>;
  <A, E, R, B, A2, E2, R2>(self: Stream<A, E, R>, sink: Sink<B, A | A2, A2, E2, R2>): Stream<B, E | E2, R | R2>;
}

Example Aggregating with a sink

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

await Effect.runPromise(Effect.gen(function* () {
  const aggregated = yield* Stream.runCollect(
    Stream.make(1, 2, 3, 4, 5, 6).pipe(
      Stream.aggregate(
        Sink.foldUntil(() => 0, 3, (sum, n) => Effect.succeed(sum + n))
      )
    )
  )
  aggregated // => [ 6, 15 ]
}))

aggregateWithin

Since v2.0.0

Aggregates elements with a sink, emitting each result when the sink completes or the schedule triggers.

Details

The schedule can flush the current aggregation even if the sink has not finished. It is stepped at most once per aggregation, when the aggregation receives its first element, with the previous aggregation's output as input. An aggregation the sink completes immediately, such as one built from leftovers, may finish without stepping the schedule.

Time between aggregations is hidden from the schedule, so an idle stream does not run it. Time-based schedules such as Schedule.fixed and Schedule.upTo({ duration }) only count time spent aggregating; use Schedule.spaced for a flush timer.

When the schedule ends, the stream emits the current aggregation and any sink leftovers, then ends without pulling more from upstream.

Signature

declare const aggregateWithin: {
  <B, A, A2, E2, R2, C, E3, R3>(sink: Sink<B, A | A2, A2, E2, R2>, schedule: Schedule<C, Option<B>, E3, R3>): <E, R>(self: Stream<A, E, R>) => Stream<B, E2 | E3 | E, R2 | R3 | R>;
  <A, E, R, B, A2, E2, R2, C, E3, R3>(self: Stream<A, E, R>, sink: Sink<B, A | A2, A2, E2, R2>, schedule: Schedule<C, Option<B>, E3, R3>): Stream<B, E | E2 | E3, R | R2 | R3>;
}

Example Aggregating with a sink and schedule

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

await Effect.runPromise(Effect.gen(function* () {
  const aggregated = yield* Stream.runCollect(
    Stream.make(1, 2, 3, 4, 5, 6).pipe(
      Stream.aggregateWithin(
        Sink.foldUntil(() => 0, 3, (sum, n) => Effect.succeed(sum + n)),
        Schedule.forever
      )
    )
  )
  aggregated // => [ 6, 15 ]
}))

transduce

Since v2.0.0

Applies a sink transducer to the stream and emits each sink result.

Signature

declare const transduce: {
  <A2, A, E2, R2>(sink: Sink<A2, A, A, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, sink: Sink<A2, A, A, E2, R2>): Stream<A2, E | E2, R | R2>;
}

Example Transducing with a sink

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

const program = Effect.gen(function* () {
  const result = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.transduce(Sink.take(2)),
    Stream.runCollect
  )

  result // => [ [ 1, 2 ], [ 3, 4 ], [] ]
})
await Effect.runPromise(program)

Broadcasting

broadcast

Since v2.0.0

Creates a PubSub-backed stream that multicasts the source to all subscribers.

Details

The returned stream is scoped and uses the provided PubSub capacity and replay settings.

Signature

declare const broadcast: {
  (options: {
    readonly capacity: "unbounded";
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly replay?: number;
    readonly strategy?: "sliding" | "dropping" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Effect<Stream<A, E, never>, never, Scope | R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly replay?: number;
    readonly strategy?: "sliding" | "dropping" | "suspend";
  }): Effect<Stream<A, E, never>, never, Scope | R>;
}

Example Broadcasting a stream

import { Effect, Stream } from "effect"

const program = Effect.scoped(
  Effect.gen(function* () {
    const broadcasted = yield* Stream.broadcast(Stream.fromArray([1, 2, 3]), {
      capacity: 8,
      replay: 3
    })

    const [left, right] = yield* Effect.all([
      Stream.runCollect(broadcasted),
      Stream.runCollect(broadcasted)
    ], { concurrency: "unbounded" })

    const result = [left, right] // => [[1, 2, 3], [1, 2, 3]]
  })
)

await Effect.runPromise(program)

broadcastN

Since v4.0.0

Creates a fixed-size tuple of streams that each emit the same elements as the source stream.

Details

The source stream starts after all downstream streams have been subscribed. With the default suspend strategy, the source can only advance capacity chunks ahead of the slowest downstream stream. If a downstream stream is interrupted, it unsubscribes from the broadcast so it no longer contributes backpressure. Finite fractional values of n are rounded down. NaN and non-positive values create no downstream streams.

Signature

declare const broadcastN: {
  <N extends number>(options: {
    readonly capacity: "unbounded";
    readonly n: N;
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly n: N;
    readonly replay?: number;
    readonly strategy?: "sliding" | "dropping" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Effect<TupleOf<N, Stream<A, E, never>>, never, Scope | R>;
  <A, E, R, N extends number>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
    readonly n: N;
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly n: N;
    readonly replay?: number;
    readonly strategy?: "sliding" | "dropping" | "suspend";
  }): Effect<TupleOf<N, Stream<A, E, never>>, never, Scope | R>;
}

Example Broadcasting to two consumers

import { Effect, Stream } from "effect"

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

    const values = yield* Effect.all([
      Stream.runCollect(left),
      Stream.runCollect(right)
    ], { concurrency: "unbounded" })

    values // => [ [ 1, 2, 3 ], [ 1, 2, 3 ] ]
  })
)

await Effect.runPromise(program)

Since v3.8.0

Returns a new Stream that multicasts the original stream, subscribing when the first consumer starts.

Details

The upstream continues running while there is at least one consumer and is finalized after the last one exits. If idleTimeToLive is set, the upstream is kept alive for that duration so a later subscriber can continue from the next element instead of restarting.

Signature

declare const share: {
  (options: {
    readonly capacity: "unbounded";
    readonly idleTimeToLive?: Duration.Input;
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly idleTimeToLive?: Duration.Input;
    readonly replay?: number;
    readonly strategy?: "sliding" | "dropping" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Effect<Stream<A, E, never>, never, Scope | R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
    readonly idleTimeToLive?: Duration.Input;
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly idleTimeToLive?: Duration.Input;
    readonly replay?: number;
    readonly strategy?: "sliding" | "dropping" | "suspend";
  }): Effect<Stream<A, E, never>, never, Scope | R>;
}

Example Sharing a stream

import { Deferred, Effect, Ref, Stream } from "effect"

const result = await Effect.runPromise(
  Effect.scoped(
    Effect.gen(function*() {
      const firstReady = yield* Deferred.make<void>()
      const secondReady = yield* Deferred.make<void>()
      const acquisitions = yield* Ref.make(0)
      const source = Stream.fromEffect(Ref.update(acquisitions, (n) => n + 1)).pipe(
        Stream.drain,
        Stream.concat(Stream.make(0)),
        Stream.concat(
          Stream.fromEffect(Effect.all([Deferred.await(firstReady), Deferred.await(secondReady)])).pipe(Stream.drain)
        ),
        Stream.concat(Stream.make(1, 2, 3))
      )
      const shared = yield* Stream.share(source, { capacity: 16, replay: 1 })
      const consume = (ready: Deferred.Deferred<void>) =>
        shared.pipe(
          Stream.tap((value) => value === 0 ? Deferred.succeed(ready, void 0) : Effect.void),
          Stream.filter((value) => value !== 0),
          Stream.runCollect
        )

      const values = yield* Effect.all([consume(firstReady), consume(secondReady)], { concurrency: "unbounded" })
      return { values, acquisitions: yield* Ref.get(acquisitions) }
    })
  )
)
result // => { values: [[1, 2, 3], [1, 2, 3]], acquisitions: 1 }

Buffering

buffer

Since v2.0.0

Buffers up to capacity elements so a faster producer can progress independently of a slower consumer.

Details

Finite buffers use the configured queue strategy: "suspend" applies backpressure, while "dropping" and "sliding" may discard elements when the buffer is full. This combinator destroys chunking; use Stream.rechunk afterward if you need fixed chunk sizes.

Signature

declare const buffer: {
  (options: {
    readonly capacity: "unbounded";
  } | {
    readonly capacity: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
  } | {
    readonly capacity: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): Stream<A, E, R>;
}

Example Buffering stream elements

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.buffer({ capacity: 1 }),
    Stream.runCollect
  )
  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

bufferArray

Since v4.0.0

Allows a faster producer to progress independently of a slower consumer by buffering up to capacity chunks in a queue.

Details

Finite buffers use the configured queue strategy: "suspend" applies backpressure, while "dropping" and "sliding" may discard chunks when the buffer is full. This combinator preserves chunking and is best with power-of-2 capacities.

Signature

declare const bufferArray: {
  (options: {
    readonly capacity: "unbounded";
  } | {
    readonly capacity: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
  } | {
    readonly capacity: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): Stream<A, E, R>;
}

Example Buffering stream chunks

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.fromArrays([1, 2], [3, 4]).pipe(
    Stream.bufferArray({ capacity: 2 }),
    Stream.runCollect
  )
  result // => [ 1, 2, 3, 4 ]
})

await Effect.runPromise(program)

Constants

DefaultChunkSize

Since v2.0.0

The default chunk size used by Stream constructors and combinators.

Signature

declare const DefaultChunkSize: number

Example Reading the default chunk size

import { Stream } from "effect"

Stream.DefaultChunkSize // => 4096

Constructors

callback

Since v4.0.0

Creates a stream from a callback that can emit values into a queue.

When to use

Use when you need callback-based code to emit stream values by offering to a Queue, or signal stream completion through the Queue module APIs.

By default it uses an "unbounded" buffer size. You can customize the buffer size and strategy by passing an object as the second argument with the bufferSize and strategy fields.

Signature

declare function callback<A, E = never, R = never>(f: (queue: Queue<A, Done<void> | E>) => Effect<unknown, E, Scope | R>, options?: {
  readonly bufferSize?: number;
  readonly strategy?: "sliding" | "dropping" | "suspend";
}): Stream<A, E, Exclude<R, Scope>>

Example Creating a stream from a callback that can emit values into a queue

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

const stream = Stream.callback<number>((queue) =>
  Effect.sync(() => {
    // Emit values to the stream
    Queue.offerUnsafe(queue, 1)
    Queue.offerUnsafe(queue, 2)
    Queue.offerUnsafe(queue, 3)
    // Signal completion
    Queue.endUnsafe(queue)
  })
)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]

die

Since v2.0.0

The stream that dies with the specified defect.

Signature

declare function die(defect: unknown): Stream<never>

Example Dying with a defect

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

const defect = new Error("Boom")
const stream = Stream.die(defect)

await Effect.runPromise(Effect.exit(Stream.runCollect(stream))) // => Exit.failCause(Cause.die(defect))

empty

Since v2.0.0

Creates an empty stream.

Signature

declare const empty: Stream<never>

Example Creating an empty stream

import { Effect, Stream } from "effect"

await Effect.runPromise(Stream.runCollect(Stream.empty)) // => []

fail

Since v2.0.0

Terminates with the specified error.

Signature

declare function fail<E>(error: E): Stream<never, E>

Example Failing a stream

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

await Effect.runPromise(Effect.exit(Stream.runCollect(Stream.fail("Uh oh!")))) // => Exit.fail("Uh oh!")

failCause

Since v2.0.0

Creates a stream that fails with the specified Cause.

Signature

declare function failCause<E>(cause: Cause<E>): Stream<never, E>

Example Failing with a cause

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

const stream = Stream.failCause(Cause.fail("Database connection failed")).pipe(
  Stream.catchCause(() => Stream.succeed("recovered"))
)

await Effect.runPromise(Stream.runCollect(stream)) // => ["recovered"]

failCauseSync

Since v2.0.0

The stream that always fails with the specified lazily evaluated Cause.

Signature

declare function failCauseSync<E>(evaluate: LazyArg<Cause<E>>): Stream<never, E>

Example Failing with a lazy cause

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

const stream = Stream.failCauseSync(() =>
  Cause.fail("Connection timeout after retries")
)

await Effect.runPromise(Stream.runCollect(stream).pipe(Effect.exit)) // => Exit.fail("Connection timeout after retries")

failSync

Since v2.0.0

Terminates with the specified lazily evaluated error.

Signature

declare function failSync<E>(evaluate: LazyArg<E>): Stream<never, E>

Example Failing a stream lazily

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

const stream = Stream.failSync(() => "Uh oh!")

await Effect.runPromise(Stream.runCollect(stream).pipe(Effect.exit)) // => Exit.fail("Uh oh!")

fromArray

Since v4.0.0

Creates a stream from an array of values.

Signature

declare function fromArray<A>(array: readonly Array<A>): Stream<A>

Example Creating a stream from an array of values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.fromArray([1, 2, 3])
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)

fromArrayEffect

Since v4.0.0

Creates a stream from an effect that produces an array of values.

When to use

Use when the array must be acquired from an Effect before the stream emits, and acquisition services or failures should be part of the stream.

Signature

declare function fromArrayEffect<A, E, R>(effect: Effect<readonly Array<A>, E, R>): Stream<A, Exclude<E, Done<any>>, R>

Example Creating a stream from an effect that produces an array of values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.fromArrayEffect(Effect.succeed(["Ada", "Grace"]))
  const values = yield* Stream.runCollect(stream)
  values // => ["Ada", "Grace"]
})

await Effect.runPromise(program)

fromArrays

Since v4.0.0

Creates a stream from an arbitrary number of arrays.

Signature

declare function fromArrays<Arr extends readonly Array<readonly Array<any>>>(...arrays: Arr): Stream<Arr[number][number]>

Example Creating a stream from an arbitrary number of arrays

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.fromArrays([1, 2], [3, 4])
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3, 4]
})

await Effect.runPromise(program)

fromAsyncIterable

Since v2.0.0

Creates a stream from an AsyncIterable.

Signature

declare function fromAsyncIterable<A, E>(iterable: AsyncIterable<A>, onError: (error: unknown) => E): Stream<A, E>

Example Creating a stream from an AsyncIterable

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

class StreamError extends Data.TaggedError("StreamError")<{ readonly cause: unknown }> {}

const iterable = (async function*() {
  yield 1
  yield 2
  yield 3
})()

await Effect.runPromise(Effect.gen(function*() {
  const stream = Stream.fromAsyncIterable(iterable, (cause) => new StreamError({ cause }))
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
}))

fromChannel

Since v2.0.0

Creates a stream from a array-emitting Channel.

Signature

declare const fromChannel: <Arr extends Arr.NonEmptyReadonlyArray<any>, E, R>(channel: Channel.Channel<Arr, E, void, unknown, unknown, unknown, R>) => Stream<Arr extends Arr.NonEmptyReadonlyArray<infer A> ? A : never, E, R>

Example Creating a stream from an array-emitting channel

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

const channel = Channel.succeed([1, 2, 3] as const)
const stream = Stream.fromChannel(channel)
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]

fromEffect

Since v2.0.0

Creates a stream from an effect.

Signature

declare function fromEffect<A, E, R>(effect: Effect<A, E, R>): Stream<A, E, R>

Example Creating a stream from an effect

import { Effect, Stream } from "effect"

const stream = Stream.fromEffect(Effect.succeed(42))
await Effect.runPromise(Stream.runCollect(stream)) // => [42]

fromEffectDrain

Since v4.0.0

Creates a stream that runs the effect and emits no elements.

Signature

declare function fromEffectDrain<A, E, R>(effect: Effect<A, E, R>): Stream<never, E, R>

Example Draining an effect into a stream

import { Effect, Stream } from "effect"

let drained = false
await Effect.runPromise(
  Stream.fromEffectDrain(Effect.sync(() => {
    drained = true
  })).pipe(Stream.runDrain)
)
drained // => true

fromEffectRepeat

Since v4.0.0

Creates a stream from an effect producing a value of type A which repeats forever.

Signature

declare function fromEffectRepeat<A, E, R>(effect: Effect<A, E, R>): Stream<A, Exclude<E, Done<any>>, R>

Example Repeating an effect forever

import { Effect, Stream } from "effect"

let n = 0
const stream = Stream.fromEffectRepeat(Effect.sync(() => ++n)).pipe(Stream.take(5))
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, 4, 5]

fromEffectSchedule

Since v4.0.0

Creates a stream from an effect producing a value of type A, which is repeated using the specified schedule.

Signature

declare function fromEffectSchedule<A, E, R, X, AS, ES, RS>(effect: Effect<A, E, R>, schedule: Schedule<X, AS, ES, RS>): Stream<A, E | ES, R | RS>

Example Repeating an effect with a schedule

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

const stream = Stream.fromEffectSchedule(Effect.succeed("ping"), Schedule.recurs(2))
await Effect.runPromise(Stream.runCollect(stream)) // => ["ping", "ping", "ping"]

fromEventListener

Since v3.1.0

Creates a stream from an event listener.

Signature

declare function fromEventListener<A = unknown>(target: EventListener, type: string, options?: boolean | {
  readonly bufferSize?: number;
  readonly capture?: boolean;
  readonly once?: boolean;
  readonly passive?: boolean;
}): Stream<A>

Example Creating a stream from an event listener

import { Effect, Stream } from "effect"

class NumberTarget implements Stream.EventListener<number> {
  addEventListener(event: string, f: (event: number) => void) {
    if (event === "data") {
      f(1)
      f(2)
      f(3)
    }
  }
  removeEventListener(_event: string, _f: (event: number) => void) {}
}

await Effect.runPromise(Effect.gen(function*() {
  const stream = Stream.fromEventListener(new NumberTarget(), "data").pipe(
    Stream.take(3)
  )
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
}))

fromIterable

Since v2.0.0

Creates a new Stream from an iterable collection of values.

Details

chunkSize: Maximum number of values emitted per chunk. Finite fractional values are rounded down, while NaN and non-positive values are treated as 1.

Signature

declare function fromIterable<A>(iterable: Iterable<A>, options?: {
  readonly chunkSize?: number;
}): Stream<A>

Example Creating a stream from an iterable

import { Effect, Stream } from "effect"

const numbers = [1, 2, 3]

const program = Effect.gen(function*() {
  const stream = Stream.fromIterable(numbers)
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)

fromIterableEffect

Since v2.0.0

Creates a stream from an effect producing an iterable of values.

When to use

Use when the iterable must be acquired from an Effect before the stream emits, and acquisition services or failures should be part of the stream.

Signature

declare function fromIterableEffect<A, E, R>(iterable: Effect<Iterable<A, any, any>, E, R>): Stream<A, E, R>

Example Creating a stream from an iterable effect

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

class UserRepo extends Context.Service<UserRepo, {
  readonly list: Effect.Effect<ReadonlyArray<string>>
}>()("UserRepo") {}

const listUsers = Effect.service(UserRepo).pipe(
  Effect.andThen((repo) => repo.list)
)

const stream = Stream.fromIterableEffect(listUsers)

const program = Effect.gen(function*() {
  const users = yield* stream.pipe(
    Stream.provideService(UserRepo, {
      list: Effect.succeed(["user1", "user2"])
    }),
    Stream.runCollect
  )
  users // => ["user1", "user2"]
})

await Effect.runPromise(program)

fromIterableEffectRepeat

Since v4.0.0

Creates a stream by repeatedly running an effect that yields an iterable of values.

Signature

declare function fromIterableEffectRepeat<A, E, R>(iterable: Effect<Iterable<A, any, any>, E, R>): Stream<A, Exclude<E, Done<any>>, R>

Example Repeating an iterable effect

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.fromIterableEffectRepeat(Effect.succeed([1, 2])).pipe(
    Stream.take(5)
  )
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 1, 2, 1]
})

await Effect.runPromise(program)

fromIteratorSucceed

Since v2.0.0

Creates a stream that consumes values from an iterator.

Details

The maxChunkSize parameter controls how many values are pulled per chunk. Finite fractional values are rounded down, while NaN and non-positive values are treated as 1.

Signature

declare function fromIteratorSucceed<A>(iterator: IterableIterator<A>, maxChunkSize?: number): Stream<A>

Example Consuming values from an iterator

import { Effect, Stream } from "effect"

function* numbers() {
  yield 1
  yield 2
  yield 3
}

const stream = Stream.fromIteratorSucceed(numbers())

const program = Effect.gen(function* () {
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)

fromPubSub

Since v2.0.0

Creates a stream from a subscription to a PubSub.

Signature

declare function fromPubSub<A>(pubsub: PubSub<A>): Stream<A>

Example Creating a stream from a subscription to a PubSub

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

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<number>({ replay: 3 })

  const fiber = yield* Stream.fromPubSub(pubsub).pipe(
    Stream.take(3),
    Stream.runCollect,
    Effect.forkChild
  )

  yield* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)
  yield* PubSub.publish(pubsub, 3)

  const values = yield* Fiber.join(fiber)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)

fromPubSubTake

Since v4.0.0

Creates a stream from a PubSub of Take values.

Details

Take values include end and failure signals.

Signature

declare function fromPubSubTake<A, E>(pubsub: PubSub<Take<A, E, void>>): Stream<A, E>

Example Creating a stream from PubSub takes

import { Effect, Exit, PubSub, Stream, Take } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<Take.Take<number, string>>({
    replay: 3
  })

  yield* PubSub.publish(pubsub, [1])
  yield* PubSub.publish(pubsub, [2])
  yield* PubSub.publish(pubsub, Exit.succeed<void>(undefined))

  const values = yield* Stream.fromPubSubTake(pubsub).pipe(Stream.runCollect)
  values // => [1, 2]
})

await Effect.runPromise(program)

fromPull

Since v2.0.0

Creates a stream from a pull effect, such as one produced by Stream.toPull.

Details

A pull effect yields chunks on demand and completes when the upstream stream ends. See Stream.toPull for a matching producer.

Signature

declare function fromPull<A, E, R, EX, RX>(pull: Effect<Pull<readonly [A, A], E, void, R>, EX, RX>): Stream<A, EX | Exclude<E, Done<any>>, R | RX>

Example Creating a stream from a pull effect

import { Effect, Stream } from "effect"

const program = Effect.scoped(
  Effect.gen(function*() {
    const source = Stream.make(1, 2, 3)
    const pull = yield* Stream.toPull(source)
    const stream = Stream.fromPull(Effect.succeed(pull))
    return yield* Stream.runCollect(stream)
  })
)

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

fromQueue

Since v2.0.0

Creates a stream that pulls values from a Queue.Dequeue.

Details

The stream emits non-empty batches of queued values and ends when the queue fails with Cause.Done; other queue failures are propagated.

Signature

declare function fromQueue<A, E>(queue: Dequeue<A, E>): Stream<A, Exclude<E, Done<void>>>

Example Creating a stream from a queue of values

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

const program = Effect.gen(function*() {
  const queue = yield* Queue.unbounded<number, Cause.Done>()
  yield* Queue.offer(queue, 1)
  yield* Queue.offer(queue, 2)
  yield* Queue.offer(queue, 3)
  yield* Queue.end(queue)

  const stream = Stream.fromQueue(queue)
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)

fromReadableStream

Since v2.0.0

Creates a stream from a lazily supplied Web ReadableStream.

Details

The stream reads from a ReadableStreamDefaultReader, maps read failures with onError, and closes the reader when the stream finalizes. By default the reader is canceled; set releaseLockOnEnd to release the lock instead.

Signature

declare function fromReadableStream<A, E>(options: {
  readonly evaluate: LazyArg<ReadableStream<A>>;
  readonly onError: (error: unknown) => E;
  readonly releaseLockOnEnd?: boolean;
}): Stream<A, E>

Example Creating a stream from a ReadableStream

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

class StreamError extends Data.TaggedError("StreamError")<{ readonly cause: unknown }> {}

const readableStream = new ReadableStream({
  start(controller) {
    controller.enqueue(1)
    controller.enqueue(2)
    controller.enqueue(3)
    controller.close()
  }
})

const program = Effect.gen(function*() {
  const stream = Stream.fromReadableStream({
    evaluate: () => readableStream,
    onError: (cause) => new StreamError({ cause })
  })
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)

fromSchedule

Since v2.0.0

Creates a stream that emits each output of a schedule that does not require input, for as long as the schedule continues.

Signature

declare function fromSchedule<O, E, R>(schedule: Schedule<O, unknown, E, R>): Stream<O, E, R>

Example Creating a stream from a schedule

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

const program = Effect.gen(function*() {
  const schedule = Schedule.recurs(3)
  const stream = Stream.fromSchedule(schedule)
  const values = yield* Stream.runCollect(stream)
  values // => [0, 1, 2]
})

await Effect.runPromise(program)

fromSubscription

Since v4.0.0

Creates a stream from a PubSub subscription.

When to use

Use when you already have a PubSub.Subscription and want to expose its messages as a Stream, with Stream.take or cancellation controlling how many values are consumed.

Signature

declare function fromSubscription<A>(pubsub: Subscription<A>): Stream<A>

Example Creating a stream from a PubSub subscription

import { 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* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)

  const stream = Stream.fromSubscription(subscription)
  const values = yield* stream.pipe(Stream.take(2), Stream.runCollect)
  values // => [1, 2]
}))

await Effect.runPromise(program)

iterate

Since v2.0.0

Creates an infinite stream by repeatedly applying a function to a seed value.

Signature

declare function iterate<A>(value: A, next: (value: A) => A): Stream<A>

Example Iterating from a seed value

import { Effect, Stream } from "effect"

const stream = Stream.iterate(1, (n) => n + 1).pipe(Stream.take(3))

const program = Effect.gen(function* () {
  const values = yield* Stream.runCollect(stream)
  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

make

Since v2.0.0

Creates a stream from a sequence of values.

Signature

declare function make<As extends readonly Array<any>>(...values: As): Stream<As[number]>

Example Creating a stream from a sequence of values

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]

never

Since v2.0.0

The stream that never produces any value or fails with any error.

Signature

declare const never: Stream<never>

Example Creating a never-ending stream

import { Effect, Stream } from "effect"

const program = Stream.never.pipe(
  Stream.take(0),
  Stream.runCollect
)

await Effect.runPromise(program) // => []

paginate

Since v2.0.0

Creates a stream by repeatedly evaluating an effectful page function.

When to use

Use to consume paginated APIs where each step returns a batch of values together with an optional next state.

Details

This is similar to unfold, but each step can emit zero or more values and independently decide whether another state should be requested.

Signature

declare function paginate<S, A, E = never, R = never>(s: S, f: (s: S) => Effect<readonly [readonly Array<A>, Option<S>], E, R>): Stream<A, E, R>

Example Paginating stream state

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

const stream = Stream.paginate(0, (n: number) =>
  Effect.succeed(
    [
      [n],
      n < 3 ? Option.some(n + 1) : Option.none<number>()
    ] as const
  ))

await Effect.runPromise(Stream.runCollect(stream)) // => [0, 1, 2, 3]

range

Since v2.0.0

Constructs a stream from a range of integers, including both endpoints.

Details

If the provided min is greater than max, the stream will not emit any values. Finite fractional chunkSize values are rounded down, while NaN and non-positive values are treated as 1.

Signature

declare function range(min: number, max: number, chunkSize: number): Stream<number>

Example Creating a numeric range

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.range(1, 5).pipe(Stream.runCollect)
  values // => [ 1, 2, 3, 4, 5 ]
})

await Effect.runPromise(program)

succeed

Since v2.0.0

Creates a single-valued pure stream.

Signature

declare function succeed<A>(value: A): Stream<A>

Example Creating a single-valued pure stream

import { Effect, Stream } from "effect"

await Effect.runPromise(Stream.runCollect(Stream.succeed(3))) // => [3]

suspend

Since v2.0.0

Creates a lazily constructed stream.

Details

The stream factory is evaluated each time the stream is run.

Signature

declare function suspend<A, E, R>(stream: LazyArg<Stream<A, E, R>>): Stream<A, E, R>

Example Creating a lazily constructed stream

import { Effect, Stream } from "effect"

await Effect.runPromise(Stream.suspend(() => Stream.make(1, 2, 3)).pipe(Stream.runCollect)) // => [1, 2, 3]

sync

Since v2.0.0

Creates a stream that synchronously evaluates a function and emits the result as a single value.

Details

The function is evaluated each time the stream is run.

Signature

declare function sync<A>(evaluate: LazyArg<A>): Stream<A>

Example Evaluating a value synchronously

import { Effect, Stream } from "effect"

await Effect.runPromise(Stream.sync(() => 2 + 1).pipe(Stream.runCollect)) // => [3]

tick

Since v2.0.0

Creates a stream that emits void immediately once, then emits another void after each specified interval.

Signature

declare function tick(interval: Input): Stream<void>

Example Emitting ticks on an interval

import { Effect, Stream } from "effect"

await Effect.runPromise(Stream.tick(0).pipe(Stream.take(3), Stream.runCollect)) // => [undefined, undefined, undefined]

transformPull

Since v4.0.0

Derives a stream by transforming its pull effect.

Signature

declare function transformPull<A, E, R, B, E2, R2, EX, RX>(self: Stream<A, E, R>, f: (pull: Pull<readonly [A, A], E, void>, scope: Scope) => Effect<Pull<readonly [B, B], E2, void, R2>, EX, RX>): Stream<B, EX | Exclude<E2, Done<any>>, R | R2 | RX>

Example Transforming a pull effect

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3)

const transformed = Stream.transformPull(stream, (pull) => Effect.succeed(pull))

await Effect.runPromise(Stream.runCollect(transformed)) // => [1, 2, 3]

transformPullBracket

Since v4.0.0

Transforms a stream by effectfully transforming its pull effect.

Details

A forked scope is also provided to the transformation function, which is closed once the resulting stream has finished processing.

Signature

declare function transformPullBracket<A, E, R, B, E2, R2, EX, RX>(self: Stream<A, E, R>, f: (pull: Pull<readonly [A, A], E, void, R>, scope: Scope, forkedScope: Scope) => Effect<Pull<readonly [B, B], E2, void, R2>, EX, RX>): Stream<B, EX | Exclude<E2, Done<any>>, R | R2 | RX>

Example Transforming a stream by effectfully transforming its pull effect

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

const finalized: Array<boolean> = []
const stream = Stream.make(1, 2, 3)

const transformed = Stream.transformPullBracket(
  stream,
  (pull, _scope, forkedScope) =>
    Effect.gen(function*() {
      yield* Scope.addFinalizer(forkedScope, Effect.sync(() => finalized.push(true)))
      return pull
    })
)

await Effect.runPromise(Stream.runCollect(transformed)) // => [1, 2, 3]
finalized // => [true]

unfold

Since v2.0.0

Creates a stream by repeatedly applying an effectful step function to a state.

Details

Each readonly [value, nextState] result emits value and continues with nextState; returning undefined ends the stream.

Signature

declare function unfold<S, A, E, R>(s: S, f: (s: S) => Effect<readonly [A, S] | undefined, E, R>): Stream<A, E, R>

Example Unfolding stream state

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.unfold(1, (n) => Effect.succeed([n, n + 1] as const))
  const values = yield* Stream.runCollect(stream.pipe(Stream.take(5)))
  values // => [ 1, 2, 3, 4, 5 ]
})

await Effect.runPromise(program)

unwrap

Since v2.0.0

Creates a stream produced from an Effect.

Signature

declare function unwrap<A, E2, R2, E, R>(effect: Effect<Stream<A, E2, R2>, E, R>): Stream<A, E2 | E, R2 | Exclude<R, Scope>>

Example Unwrapping a stream effect

import { Effect, Stream } from "effect"

const effect = Effect.succeed(Stream.make(1, 2, 3))

const stream = Stream.unwrap(effect)

const program = Effect.gen(function*() {
  const chunk = yield* Stream.runCollect(stream)
  chunk // => [ 1, 2, 3 ]
})
await Effect.runPromise(program)

Deduplication

changes

Since v2.0.0

Emits only elements that differ from the previous one.

Signature

declare function changes<A, E, R>(self: Stream<A, E, R>): Stream<A, E, R>

Example Emitting changed values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.fromIterable([1, 1, 2, 2, 3]).pipe(
    Stream.changes,
    Stream.runCollect
  )

  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

changesWith

Since v2.0.0

Returns a stream that only emits elements that are not equal to the previously emitted element, as determined by the specified predicate.

Signature

declare const changesWith: {
  <A>(f: (x: A, y: A) => boolean): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, f: (x: A, y: A) => boolean): Stream<A, E, R>;
}

Example Emitting values that changed by equivalence

import { Effect, Stream } from "effect"

const stream = Stream.make("A", "a", "B", "b", "b").pipe(
  Stream.changesWith((left, right) => left.toLowerCase() === right.toLowerCase())
)

await Effect.runPromise(
  Effect.gen(function*() {
    const values = yield* Stream.runCollect(stream)
    values // => [ 'A', 'B' ]
  })
)

changesWithEffect

Since v2.0.0

Emits only elements that differ from the previous element, using an effectful equality check.

Details

The predicate runs for each element after the first; returning true treats it as equal and skips it.

Signature

declare const changesWithEffect: {
  <A, E2, R2>(f: (x: A, y: A) => Effect<boolean, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, f: (x: A, y: A) => Effect<boolean, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Effectfully emitting changed values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.make(1, 1, 2, 2, 3, 3).pipe(
    Stream.changesWithEffect((a, b) => Effect.succeed(a === b))
  )
  const result = yield* Stream.runCollect(stream)
  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

Delays & Timeouts

timeout

Since v2.0.0

Ends the stream if it does not produce a value within the specified duration.

Signature

declare const timeout: {
  (duration: Input): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, duration: Input): Stream<A, E, R>;
}

Example Timing out a stream

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1).pipe(
    Stream.concat(Stream.never),
    Stream.timeout("1 second"),
    Stream.runCollect
  )
  values // => [ 1 ]
})

await Effect.runPromise(program)

timeoutOrElse

Since v4.0.0

Switches to a fallback stream if this stream does not emit a value within the specified duration.

When to use

Use when a stream should continue with another stream if an upstream pull waits longer than the allowed duration.

Details

The timeout is checked for each pull. A zero duration uses orElse immediately, while an infinite duration leaves the original stream unchanged.

Gotchas

The fallback stream is not timed after the switch.

See

timeout for ending the stream instead of switching to a fallback stream

Signature

declare const timeoutOrElse: {
  <B, E2, R2>(options: {
    readonly duration: Duration.Input;
    readonly orElse: () => Stream<B, E2, R2>;
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<B | A, E2 | E, R2 | R>;
  <A, E, R, B, E2, R2>(self: Stream<A, E, R>, options: {
    readonly duration: Duration.Input;
    readonly orElse: () => Stream<B, E2, R2>;
  }): Stream<A | B, E | E2, R | R2>;
}

Destructors

mkArrayBuffer

Since v4.0.0

Concatenates the stream's Uint8Array chunks into a single ArrayBuffer.

Gotchas

This materializes the full content in memory. The source stream must not reuse or mutate emitted buffers, which are retained until collection completes.

Signature

declare function mkArrayBuffer<E, R>(self: Stream<Uint8Array<ArrayBufferLike>, E, R>): Effect<ArrayBuffer, E, R>

Example Joining byte chunks into an ArrayBuffer

import { Effect, Stream } from "effect"

const program = Stream.make(
  new Uint8Array([1, 2]),
  new Uint8Array([3, 4])
).pipe(
  Stream.mkArrayBuffer,
  Effect.map((buffer) => [...new Uint8Array(buffer)])
)

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

mkString

Since v2.0.0

Concatenates all emitted strings into a single string.

Signature

declare function mkString<E, R>(self: Stream<string, E, R>): Effect<string, E, R>

Example Joining strings from a stream

import { Effect, Stream } from "effect"

const stream = Stream.make("Hello", " ", "World", "!")
const program = Effect.gen(function*() {
  const text = yield* Stream.mkString(stream)
  text // => "Hello World!"
})

await Effect.runPromise(program)

mkUint8Array

Since v4.0.0

Concatenates the stream's Uint8Array chunks into a single Uint8Array.

Gotchas

This materializes the full content in memory. The source stream must not reuse or mutate emitted buffers, which are retained until collection completes.

Signature

declare function mkUint8Array<E, R>(self: Stream<Uint8Array<ArrayBufferLike>, E, R>): Effect<Uint8Array<ArrayBufferLike>, E, R>

Example Joining Uint8Array chunks

import { Effect, Stream } from "effect"

const stream = Stream.make(new Uint8Array([1, 2]), new Uint8Array([3, 4]))
const program = Effect.gen(function*() {
  const bytes = yield* Stream.mkUint8Array(stream)
  const values = Array.from(bytes) // => [1, 2, 3, 4]
})

await Effect.runPromise(program)

peel

Since v2.0.0

Runs a sink to peel off enough elements to produce a value and returns that value with the remaining stream in a scope.

Details

The returned stream is only valid within the scope.

Signature

declare const peel: {
  <A2, A, E2, R2>(sink: Sink<A2, A, A, E2, R2>): <E, R>(self: Stream<A, E, R>) => Effect<[A2, Stream<A, E, never>], E2 | E, Scope | R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, sink: Sink<A2, A, A, E2, R2>): Effect<[A2, Stream<A, E, never>], E | E2, Scope | R | R2>;
}

Example Peeling a stream with a sink

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

const stream = Stream.fromArrays([1, 2, 3], [4, 5, 6])
const sink = Sink.take<number>(3)

const program = Effect.scoped(
  Effect.gen(function*() {
    const [peeled, rest] = yield* Stream.peel(stream, sink)
    const remaining = yield* Stream.runCollect(rest)
    const result = [peeled, remaining] // => [[1, 2, 3], [4, 5, 6]]
  })
)

await Effect.runPromise(program)

run

Since v2.0.0

Runs a stream with a sink and returns the sink result.

Signature

declare const run: {
  <A2, A, L, E2, R2>(sink: Sink<A2, A, L, E2, R2>): <E, R>(self: Stream<A, E, R>) => Effect<A2, E2 | E, R2 | R>;
  <A, E, R, L, A2, E2, R2>(self: Stream<A, E, R>, sink: Sink<A2, A, L, E2, R2>): Effect<A2, E | E2, R | R2>;
}

Example Running a stream with a sink

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

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

await Effect.runPromise(program) // => 6

runCollect

Since v2.0.0

Runs the stream and collects all elements into an array.

Signature

declare function runCollect<A, E, R>(self: Stream<A, E, R>): Effect<Array<A>, E, R>

Example Collecting stream values

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)

const program = Effect.gen(function*() {
  const collected = yield* Stream.runCollect(stream)
  collected // => [ 1, 2, 3, 4, 5 ]
})

await Effect.runPromise(program)

runCount

Since v2.0.0

Runs the stream and returns the number of elements emitted.

Signature

declare function runCount<A, E, R>(self: Stream<A, E, R>): Effect<number, E, R>

Example Counting stream values

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)

const program = Effect.gen(function* () {
  const count = yield* Stream.runCount(stream)
  count // => 5
})

await Effect.runPromise(program)

runDrain

Since v2.0.0

Runs the stream for its effects, discarding emitted elements.

Signature

declare function runDrain<A, E, R>(self: Stream<A, E, R>): Effect<void, E, R>

Example Draining a stream run

import { Effect, Stream } from "effect"

const values: Array<number> = []
const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2, 3).pipe(
    Stream.mapEffect((n) => Effect.sync(() => values.push(n)))
  )

  yield* Stream.runDrain(stream)
})

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

runFold

Since v2.0.0

Runs the stream and folds elements using a pure reducer.

Signature

declare const runFold: {
  <Z, A>(initial: LazyArg<Z>, f: (acc: Z, a: A) => Z): <E, R>(self: Stream<A, E, R>) => Effect<Z, E, R>;
  <A, E, R, Z>(self: Stream<A, E, R>, initial: LazyArg<Z>, f: (acc: Z, a: A) => Z): Effect<Z, E, R>;
}

Example Folding stream values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const total = yield* Stream.runFold(
    Stream.make(1, 2, 3),
    () => 0,
    (acc, n) => acc + n
  )
  total // => 6
})

await Effect.runPromise(program)

runFoldEffect

Since v2.0.0

Runs the stream and folds elements using an effectful reducer.

When to use

Use when reducing stream elements needs Effects, services, or failures in the reducer.

Signature

declare const runFoldEffect: {
  <Z, A, EX, RX>(initial: LazyArg<Z>, f: (acc: Z, a: A) => Effect<Z, EX, RX>): <E, R>(self: Stream<A, E, R>) => Effect<Z, EX | E, RX | R>;
  <A, E, R, Z, EX, RX>(self: Stream<A, E, R>, initial: LazyArg<Z>, f: (acc: Z, a: A) => Effect<Z, EX, RX>): Effect<Z, E | EX, R | RX>;
}

Example Effectfully folding stream values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const total = yield* Stream.runFoldEffect(
    Stream.make(1, 2, 3),
    () => 0,
    (acc, n) => Effect.succeed(acc + n)
  )
  total // => 6
})

await Effect.runPromise(program)

runForEach

Since v2.0.0

Runs the provided effectful callback for each element of the stream.

Signature

declare const runForEach: {
  <A, X, E2, R2>(f: (a: A) => Effect<X, E2, R2>): <E, R>(self: Stream<A, E, R>) => Effect<void, E2 | E, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, f: (a: A) => Effect<X, E2, R2>): Effect<void, E | E2, R | R2>;
}

Example Running an effect for each value

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3)
const values: Array<string> = []

const program = Effect.gen(function*() {
  yield* Stream.runForEach(stream, (n) => Effect.sync(() => values.push(`Processing: ${n}`)))
})

await Effect.runPromise(program)
values // => ["Processing: 1", "Processing: 2", "Processing: 3"]

runForEachArray

Since v4.0.0

Consumes the stream in chunks, passing each non-empty array to the callback.

Signature

declare const runForEachArray: {
  <A, X, E2, R2>(f: (a: readonly [A, A]) => Effect<X, E2, R2>): <E, R>(self: Stream<A, E, R>) => Effect<void, E2 | E, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, f: (a: readonly [A, A]) => Effect<X, E2, R2>): Effect<void, E | E2, R | R2>;
}

Example Consuming stream chunks

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)
const chunks: Array<string> = []
const program = Effect.gen(function*() {
  yield* Stream.runForEachArray(
    stream,
    (chunk) => Effect.sync(() => chunks.push(chunk.join(", ")))
  )
})

await Effect.runPromise(program)
chunks // => ["1, 2, 3, 4, 5"]

runForEachWhile

Since v2.0.0

Runs the stream, applying the effectful predicate to each element and stopping when it returns false.

Signature

declare const runForEachWhile: {
  <A, E2, R2>(f: (a: A) => Effect<boolean, E2, R2>): <E, R>(self: Stream<A, E, R>) => Effect<void, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, f: (a: A) => Effect<boolean, E2, R2>): Effect<void, E | E2, R | R2>;
}

Example Running effects while a predicate holds

import { Effect, Stream } from "effect"

const values: Array<number> = []
const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2, 3, 4, 5)

  yield* Stream.runForEachWhile(stream, (n) =>
    Effect.gen(function*() {
      yield* Effect.sync(() => values.push(n))
      return n < 3
    })
  )
})

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

runHead

Since v2.0.0

Runs the stream and returns the first element as an Option.

Signature

declare function runHead<A, E, R>(self: Stream<A, E, R>): Effect<Option<A>, E, R>

Example Getting the first stream value

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

const program = Effect.gen(function*() {
  const head = yield* Stream.runHead(Stream.make(1, 2, 3))
  Option.getOrThrow(head) // => 1
})

await Effect.runPromise(program)

runIntoPubSub

Since v2.0.0

Runs the stream, publishing elements into the provided PubSub.

Details

shutdownOnEnd controls whether the PubSub is shut down when the stream ends. It only shuts down when set to true.

Signature

declare const runIntoPubSub: {
  <A>(pubsub: PubSub<A>, options?: {
    readonly shutdownOnEnd?: boolean;
  }): <E, R>(self: Stream<A, E, R>) => Effect<void, E, R>;
  <A, E, R>(self: Stream<A, E, R>, pubsub: PubSub<A>, options?: {
    readonly shutdownOnEnd?: boolean;
  }): Effect<void, E, R>;
}

Example Running a stream into a PubSub

import { 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)

  const first = yield* PubSub.take(subscription)
  const second = yield* PubSub.take(subscription)

  first // => 1
  second // => 2
}))

await Effect.runPromise(program)

runIntoQueue

Since v2.0.0

Runs the stream, offering each element to the provided queue and ending it with Cause.Done when the stream completes.

Signature

declare const runIntoQueue: {
  <A, E>(queue: Queue<A, Done<void> | E>): <R>(self: Stream<A, E, R>) => Effect<void, never, R>;
  <A, E, R>(self: Stream<A, E, R>, queue: Queue<A, Done<void> | E>): Effect<void, never, R>;
}

Example Running a stream into a queue

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

  values // => [ 1, 2, 3 ]
  done._tag === "Done" // => true
})
await Effect.runPromise(program)

runLast

Since v2.0.0

Runs the stream and returns the last element as an Option.

When to use

Use to consume a finite stream when only the final emitted element matters.

Details

Option.some contains the last emitted element. Option.none means the stream completed without emitting.

Gotchas

The returned effect waits for the stream to complete before it can produce a value.

See

runHead for consuming only the first emitted element
runCollect for collecting every emitted element
runDrain for consuming the stream while discarding emitted elements

Signature

declare function runLast<A, E, R>(self: Stream<A, E, R>): Effect<Option<A>, E, R>

runSum

Since v2.0.0

Runs the stream and returns the numeric sum of its elements.

Signature

declare function runSum<E, R>(self: Stream<number, E, R>): Effect<number, E, R>

Example Summing stream values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const total = yield* Stream.runSum(Stream.make(1, 2, 3))
  total // => 6
})

await Effect.runPromise(program)

toAsyncIterable

Since v3.15.0

Converts a stream to an AsyncIterable for for await...of consumption.

Signature

declare function toAsyncIterable<A, E>(self: Stream<A, E>): AsyncIterable<A>

Example Converting to an async iterable

import { Stream } from "effect"

const stream = Stream.make(1, 2, 3)

await globalThis.Array.fromAsync(Stream.toAsyncIterable(stream)) // => [1, 2, 3]

toAsyncIterableEffect

Since v3.15.0

Creates an effect that yields an AsyncIterable using the current services.

When to use

Use when the AsyncIterable should be created inside Effect with the current context supplying the stream's services.

Signature

declare function toAsyncIterableEffect<A, E, R>(self: Stream<A, E, R>): Effect<AsyncIterable<A, any, any>, never, R>

Example Creating an AsyncIterable effect

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3)

const program = Effect.gen(function*() {
  const iterable = yield* Stream.toAsyncIterableEffect(stream)
  return yield* Effect.promise(() => globalThis.Array.fromAsync(iterable))
})

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

toAsyncIterableWith

Since v4.0.0

Converts the stream to an AsyncIterable using the provided services.

When to use

Use when converting outside an Effect and you already have the Context needed to run the stream.

Signature

declare const toAsyncIterableWith: {
  <XR>(context: Context<XR>): <A, E, R>(self: Stream<A, E, R>) => AsyncIterable<A>;
  <A, E, XR, R>(self: Stream<A, E, R>, context: Context<XR>): AsyncIterable<A>;
}

Example Converting to an AsyncIterable with services

import { Context, Stream } from "effect"

const stream = Stream.make(1, 2, 3)
const iterable = Stream.toAsyncIterableWith(stream, Context.empty())

await globalThis.Array.fromAsync(iterable) // => [1, 2, 3]

toChannel

Since v2.0.0

Creates a channel from a stream.

Signature

declare function toChannel<A, E, R>(stream: Stream<A, E, R>): Channel<readonly [A, A], E, void, unknown, unknown, unknown, R>

Example Converting a stream to a channel

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

const channel = Stream.toChannel(Stream.make(1, 2, 3))
const values = await Effect.runPromise(Channel.runCollect(channel))
values.flat() // => [1, 2, 3]

toPubSub

Since v2.0.0

Converts a stream to a PubSub of emitted values for concurrent consumption.

Details

shutdownOnEnd indicates whether the PubSub should be shut down when the stream ends. By default this is true.

Signature

declare const toPubSub: {
  (options: {
    readonly capacity: "unbounded";
    readonly replay?: number;
    readonly shutdownOnEnd?: boolean;
  } | {
    readonly capacity: number;
    readonly replay?: number;
    readonly shutdownOnEnd?: boolean;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Effect<PubSub<A>, never, Scope | R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
    readonly replay?: number;
    readonly shutdownOnEnd?: boolean;
  } | {
    readonly capacity: number;
    readonly replay?: number;
    readonly shutdownOnEnd?: boolean;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): Effect<PubSub<A>, never, Scope | R>;
}

Example Converting a stream to a PubSub for concurrent consumption

import { 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)
  const first = yield* PubSub.take(subscription)

  first // => 1
}))
await Effect.runPromise(program)

toPubSubTake

Since v4.0.0

Converts a stream to a PubSub of Take values for concurrent consumption.

Details

Chunks are published as Take values. When the stream ends, the PubSub is ended with the stream's Exit, so every subscriber, including one that subscribes later, observes completion or failure after its buffered chunks.

Signature

declare const toPubSubTake: {
  (options: {
    readonly capacity: "unbounded";
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly replay?: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Effect<PubSub<Take<A, E, void>>, never, Scope | R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
    readonly replay?: number;
  } | {
    readonly capacity: number;
    readonly replay?: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): Effect<PubSub<Take<A, E, void>>, never, Scope | R>;
}

Example Converting to a PubSub of takes

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

const program = 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)

  if (Array.isArray(take)) {
    take // => [ 1, 2, 3 ]
  }
})
await Effect.runPromise(Effect.scoped(program))

toPull

Since v2.0.0

Returns a scoped pull for manually consuming the stream's output chunks.

Details

The pull fails with Cause.Done when the stream ends and with the stream error on failure.

Signature

declare function toPull<A, E, R>(self: Stream<A, E, R>): Effect<Pull<readonly [A, A], E, void, never>, never, Scope | R>

Example Creating a scoped pull

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3)

const program = Effect.scoped(
  Effect.gen(function*() {
    const pull = yield* Stream.toPull(stream)
    const chunk = yield* pull
    chunk // => [ 1, 2, 3 ]
  })
)

await Effect.runPromise(program)

toQueue

Since v2.0.0

Creates a scoped dequeue that is fed by the stream for concurrent consumption.

Details

Elements are offered to the queue as the stream runs. Stream completion is signaled with Cause.Done, stream failures fail the queue, and the queue is shut down when the surrounding scope closes.

Signature

declare const toQueue: {
  (options: {
    readonly capacity: "unbounded";
  } | {
    readonly capacity: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): <A, E, R>(self: Stream<A, E, R>) => Effect<Dequeue<A, Done<void> | E>, never, Scope | R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly capacity: "unbounded";
  } | {
    readonly capacity: number;
    readonly strategy?: "dropping" | "sliding" | "suspend";
  }): Effect<Dequeue<A, Done<void> | E>, never, Scope | R>;
}

Example Converting a stream to a Queue for concurrent consumption

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

const program = Effect.gen(function* () {
  const queue = yield* Stream.toQueue(Stream.fromIterable([1, 2, 3]), { capacity: 8 })
  const chunk = yield* Queue.takeBetween(queue, 1, 3)
  chunk // => [ 1, 2, 3 ]
})
await Effect.runPromise(Effect.scoped(program))

toReadableStream

Since v2.0.0

Converts a stream to a ReadableStream.

Details

See https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream.

Signature

declare const toReadableStream: {
  <A>(options?: {
    readonly strategy?: QueuingStrategy<A>;
  }): <E>(self: Stream<A, E>) => ReadableStream<A>;
  <A, E>(self: Stream<A, E>, options?: {
    readonly strategy?: QueuingStrategy<A>;
  }): ReadableStream<A>;
}

Example Converting a stream to a ReadableStream

import { Stream } from "effect"

const readableStream = Stream.toReadableStream(Stream.make(1, 2, 3))
const values = await globalThis.Array.fromAsync(readableStream)
values // => [ 1, 2, 3 ]

toReadableStreamEffect

Since v2.0.0

Creates an Effect that builds a ReadableStream from the stream.

When to use

Use when bridging to Web Streams from inside an Effect so the required services can be captured from the current context.

Details

See https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream.

Signature

declare const toReadableStreamEffect: {
  <A>(options?: {
    readonly strategy?: QueuingStrategy<A>;
  }): <E, R>(self: Stream<A, E, R>) => Effect<ReadableStream<A>, never, R>;
  <A, E, R>(self: Stream<A, E, R>, options?: {
    readonly strategy?: QueuingStrategy<A>;
  }): Effect<ReadableStream<A>, never, R>;
}

Example Creating a ReadableStream effect

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)

const effect = Effect.gen(function*() {
  const readableStream = yield* Stream.toReadableStreamEffect(stream)
  readableStream instanceof ReadableStream // => true
})

await Effect.runPromise(effect)

toReadableStreamWith

Since v4.0.0

Converts the stream to a ReadableStream using the provided services.

When to use

Use when bridging to Web Streams and you already have the Context required to run the stream outside an Effect.

Details

See https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream.

When signal aborts, the stream is interrupted even if the reader has stopped pulling, and the ReadableStream closes.

Signature

declare const toReadableStreamWith: {
  <A, XR>(context: Context<XR>, options?: {
    readonly signal?: AbortSignal;
    readonly strategy?: QueuingStrategy<A>;
  }): <E, R>(self: Stream<A, E, R>) => ReadableStream<A>;
  <A, E, XR, R>(self: Stream<A, E, R>, context: Context<XR>, options?: {
    readonly signal?: AbortSignal;
    readonly strategy?: QueuingStrategy<A>;
  }): ReadableStream<A>;
}

Example Converting to a ReadableStream with services

import { Context, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)
const readableStream = Stream.toReadableStreamWith(stream, Context.empty())
const values = await globalThis.Array.fromAsync(readableStream)
values // => [ 1, 2, 3, 4, 5 ]

Do Notation

bind

Since v2.0.0

Binds the result of a stream to a field in the do-notation record.

Signature

declare const bind: {
  <N extends string, A extends Record<string, any>, B, E2, R2>(name: N, f: (a: NoInfer<A>) => Stream<B, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
  }): <E, R>(self: Stream<A, E, R>) => Stream<Simplify<Omit<A, N> & Record<N, B>>, E2 | E, R2 | R>;
  <A extends Record<string, any>, E, R, B, E2, R2, N extends string>(self: Stream<A, E, R>, name: N, f: (a: NoInfer<A>) => Stream<B, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
  }): Stream<Simplify<Omit<A, N> & Record<N, B>>, E | E2, R | R2>;
}

Example Binding a stream value

import { Effect, Stream } from "effect"

const program = Stream.Do.pipe(
  Stream.bind("a", () => Stream.make(1, 2)),
  Stream.bind("b", ({ a }) => Stream.succeed(a + 1))
)

const result = Stream.runCollect(program)

await Effect.runPromise(result) // => [{ a: 1, b: 2 }, { a: 2, b: 3 }]

bindEffect

Since v2.0.0

Binds an Effect-produced value into the do-notation record for each stream element.

Signature

declare const bindEffect: {
  <N extends string, A extends Record<string, any>, B, E2, R2>(name: N, f: (a: NoInfer<A>) => Effect<B, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
    readonly unordered?: boolean;
  }): <E, R>(self: Stream<A, E, R>) => Stream<Simplify<Omit<A, N> & Record<N, B>>, E2 | E, R2 | R>;
  <A extends Record<string, any>, E, R, B, E2, R2, N extends string>(self: Stream<A, E, R>, name: N, f: (a: NoInfer<A>) => Effect<B, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
    readonly unordered?: boolean;
  }): Stream<Simplify<Omit<A, N> & Record<N, B>>, E | E2, R | R2>;
}

Example Binding an effect value

import { Effect, Stream } from "effect"

const stream = Stream.Do.pipe(
  Stream.bind("value", () => Stream.make(1, 2)),
  Stream.bindEffect("double", ({ value }) => Effect.succeed(value * 2))
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [ { value: 1, double: 2 }, { value: 2, double: 4 } ]
})

await Effect.runPromise(program)

bindTo

Since v2.0.0

Maps each element into a record keyed by the provided name.

Signature

declare const bindTo: {
  <N extends string>(name: N): <A, E, R>(self: Stream<A, E, R>) => Stream<{ [K in string]: A }, E, R>;
  <A, E, R, N extends string>(self: Stream<A, E, R>, name: N): Stream<{ [K in string]: A }, E, R>;
}

Example Binding values to a record key

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3).pipe(Stream.bindTo("value"))

await Effect.runPromise(Stream.runCollect(stream)) // => [{ value: 1 }, { value: 2 }, { value: 3 }]

Do

Since v2.0.0

Provides the entry point for do-notation style stream composition.

Signature

declare const Do: Stream<{}>

Example Starting stream do notation

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

const program = pipe(
  Stream.Do,
  Stream.bind("value", () => Stream.fromArray([1, 2])),
  Stream.let("next", ({ value }) => value + 1)
)

const effect = Effect.gen(function*() {
  const collected = yield* Stream.runCollect(program)
  collected // => [ { value: 1, next: 2 }, { value: 2, next: 3 } ]
})

await Effect.runPromise(effect)

Error Handling

catchCause

Since v2.0.0

Switches over to the stream produced by the provided function in case this one fails. Allows recovery from all causes of failure, including interruption if the stream is uninterruptible.

Signature

declare const catchCause: {
  <E, A2, E2, R2>(f: (cause: Cause<E>) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (cause: Cause<E>) => Stream<A2, E2, R2>): Stream<A | A2, E2, R | R2>;
}

Example Catching stream causes

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("Oops!")),
  Stream.concat(Stream.make(3, 4))
)

const recovered = stream.pipe(
  Stream.catchCause(() => Stream.make(999))
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(recovered)
  values // => [ 1, 2, 999 ]
})

await Effect.runPromise(program)

catchCauseFilter

Since v4.0.0

Recovers from stream failures by filtering the Cause and switching to a recovery stream.

When to use

Use when you need to recover a stream only from causes selected by a Filter, while giving the recovery both the selected value and the original Cause.

Details

The filter is applied to the full Cause. A successful filter result is passed to f together with the original cause; a failed filter result re-fails with the residual cause.

See

catchCauseIf for predicate-based cause selection
catchFilter for filtering typed error values instead of full causes
catchCause for recovering from every cause without filtering

Signature

declare const catchCauseFilter: {
  <E, EB, A2, E2, R2, X extends Cause<any>>(filter: Filter<Cause<E>, EB, X>, f: (failure: EB, cause: Cause<E>) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2 | Error<X>, R2 | R>;
  <A, E, R, EB, A2, E2, R2, X extends Cause<any>>(self: Stream<A, E, R>, filter: Filter<Cause<E>, EB, X>, f: (failure: EB, cause: Cause<E>) => Stream<A2, E2, R2>): Stream<A | A2, E2 | Error<X>, R | R2>;
}

catchCauseIf

Since v4.0.0

Recovers from stream failures by filtering the Cause and switching to a recovery stream. Non-matching causes are re-emitted as failures.

Signature

declare const catchCauseIf: {
  <E, A2, E2, R2>(predicate: Predicate<Cause<E>>, f: (cause: Cause<E>) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A, E | E2, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, predicate: Predicate<Cause<E>>, f: (cause: Cause<E>) => Stream<A2, E2, R2>): Stream<A | A2, E | E2, R | R2>;
}

Example Catching matching causes

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

const program = Effect.gen(function*() {
  const failingStream = Stream.fail("NetworkError")
  const recovered = Stream.catchCauseIf(
    failingStream,
    (cause) => Cause.hasFails(cause),
    (cause) => Stream.make(`Recovered: ${Cause.squash(cause)}`)
  )

  const output = yield* Stream.runCollect(recovered)
  output // => [ 'Recovered: NetworkError' ]
})

await Effect.runPromise(program)

catchDefect

Since v4.0.0

Recovers from defects using the provided function.

Details

Typed failures and interruptions are not caught.

Signature

declare const catchDefect: {
  <A2, E2, R2>(f: (defect: unknown) => Stream<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (defect: unknown) => Stream<A2, E2, R2>): Stream<A | A2, E | E2, R | R2>;
}

Example Recovering from a defect

import { Effect, Stream } from "effect"

const stream = Stream.die("boom").pipe(
  Stream.catchDefect((defect) => Stream.succeed(`recovered: ${defect}`))
)

const result = Effect.runSync(Stream.runCollect(stream))
result // => ["recovered: boom"]

catchFilter

Since v4.0.0

Recovers from errors that match a Filter by switching to a recovery stream.

When to use

Use to recover from stream errors with a reusable Filter when matching can also narrow or transform the error before choosing the recovery stream.

Details

Successful filter results are passed to f. Failed filter results go to orElse when provided; otherwise the filter failure is re-failed.

See

catchIf for predicate or refinement based recovery
catchTag for _tag based recovery from one tagged error
catchTags for _tag based recovery from multiple tagged errors
catchCauseFilter for filtering full causes

Signature

declare const catchFilter: {
  <E, EB, A2, E2, R2, X, A3 = unassigned, E3 = never, R3 = never>(filter: Filter<NoInfer<E>, EB, X>, f: (failure: EB) => Stream<A2, E2, R2>, orElse?: (failure: X) => Stream<A3, E3, R3>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A | Exclude<A3, unassigned>, E2 | E3 | A3 extends unassigned ? X : never, R2 | R3 | R>;
  <A, E, R, EB, A2, E2, R2, X, A3 = unassigned, E3 = never, R3 = never>(self: Stream<A, E, R>, filter: Filter<NoInfer<E>, EB, X>, f: (failure: EB) => Stream<A2, E2, R2>, orElse?: (failure: X) => Stream<A3, E3, R3>): Stream<A | A2 | Exclude<A3, unassigned>, E2 | E3 | A3 extends unassigned ? X : never, R | R2 | R3>;
}

catchIf

Since v4.0.0

Recovers from errors that match a predicate by switching to a recovery stream.

Details

When a failure matches the filter, the stream switches to the recovery stream. Non-matching failures propagate downstream, so the error type is preserved unless the filter narrows it.

Signature

declare const catchIf: {
  <E, EB, A2, E2, R2, A3 = unassigned, E3 = never, R3 = never>(refinement: Refinement<NoInfer<E>, EB>, f: (e: EB) => Stream<A2, E2, R2>, orElse?: (e: Exclude<E, EB>) => Stream<A3, E3, R3>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A | Exclude<A3, unassigned>, E2 | E3 | A3 extends unassigned ? Exclude<E, EB> : never, R2 | R3 | R>;
  <E, A2, E2, R2, A3 = unassigned, E3 = never, R3 = never>(predicate: Predicate<NoInfer<E>>, f: (e: NoInfer<E>) => Stream<A2, E2, R2>, orElse?: (e: NoInfer<E>) => Stream<A3, E3, R3>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A | Exclude<A3, unassigned>, E2 | E3 | A3 extends unassigned ? E : never, R2 | R3 | R>;
  <A, E, R, EB, A2, E2, R2, A3 = unassigned, E3 = never, R3 = never>(self: Stream<A, E, R>, refinement: Refinement<E, EB>, f: (e: EB) => Stream<A2, E2, R2>, orElse?: (e: Exclude<E, EB>) => Stream<A3, E3, R3>): Stream<A | A2 | Exclude<A3, unassigned>, E2 | E3 | A3 extends unassigned ? Exclude<E, EB> : never, R | R2 | R3>;
  <A, E, R, A2, E2, R2, A3 = unassigned, E3 = never, R3 = never>(self: Stream<A, E, R>, predicate: Predicate<E>, f: (e: E) => Stream<A2, E2, R2>, orElse?: (e: E) => Stream<A3, E3, R3>): Stream<A | A2 | Exclude<A3, unassigned>, E2 | E3 | A3 extends unassigned ? E : never, R | R2 | R3>;
}

Example Catching matching failures

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail(42)),
  Stream.catchIf(
    (error): error is 42 => error === 42,
    () => Stream.make(999)
  )
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [ 1, 2, 999 ]
})

await Effect.runPromise(program)

catchReason

Since v4.0.0

Catches a specific reason within a tagged error.

When to use

Use to handle nested error causes without removing the parent error from the error channel.

Details

The handler receives the unwrapped reason.

Signature

declare const catchReason: {
  <K extends string, E, RK extends string, A2, E2, R2, A3 = unassigned, E3 = never, R3 = never>(errorTag: K, reasonTag: RK, f: (reason: ExtractReason<ExtractTag<NoInfer<E>, K>, RK>, error: NarrowReason<ExtractTag<NoInfer<E>, K>, RK>) => Stream<A2, E2, R2>, orElse?: (reason: ExcludeReason<ExtractTag<NoInfer<E>, K>, RK>, error: OmitReason<ExtractTag<NoInfer<E>, K>, RK>) => Stream<A3, E3, R3>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A | Exclude<A3, unassigned>, E2 | E3 | ExcludeTag<E, K> | A3 extends unassigned ? ExtractTag<E, K> : never, R2 | R3 | R>;
  <A, E, R, K extends string, RK extends string, A2, E2, R2, A3 = unassigned, E3 = never, R3 = never>(self: Stream<A, E, R>, errorTag: K, reasonTag: RK, f: (reason: ExtractReason<ExtractTag<E, K>, RK>, error: NarrowReason<ExtractTag<E, K>, RK>) => Stream<A2, E2, R2>, orElse?: (reason: ExcludeReason<ExtractTag<E, K>, RK>, error: OmitReason<ExtractTag<E, K>, RK>) => Stream<A3, E3, R3>): Stream<A | A2 | Exclude<A3, unassigned>, E2 | E3 | ExcludeTag<E, K> | A3 extends unassigned ? ExtractTag<E, K> : never, R | R2 | R3>;
}

Example Catching a tagged error reason

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

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}

class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
  limit: number
}> {}

class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError | QuotaExceededError
}> {}

const stream = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
)

const program = Effect.gen(function*() {
  const values = yield* stream.pipe(
    Stream.catchReason("AiError", "RateLimitError", (reason) =>
      Stream.succeed(`retry: ${reason.retryAfter}`)
    ),
    Stream.runCollect
  )
  values // => [ 'retry: 60' ]
})

await Effect.runPromise(program)

catchReasons

Since v4.0.0

Catches multiple reasons within a tagged error using an object of handlers.

Signature

declare const catchReasons: {
  <K extends string, E, Cases extends { [RK in string]: (reason: ExtractReason<ExtractTag<NoInfer<E>, K>, RK>, error: NarrowReason<ExtractTag<NoInfer<E>, K>, RK>) => Stream<any, any, any> }, A2 = unassigned, E2 = never, R2 = never>(errorTag: K, cases: Cases, orElse?: (reason: ExcludeReason<ExtractTag<NoInfer<E>, K>, Extract<keyof Cases, string>>, error: OmitReason<ExtractTag<NoInfer<E>, K>, Extract<keyof Cases, string>>) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A | Exclude<A2, unassigned> | { [RK in string | number | symbol]: Cases[RK] extends (...args: Array<any>) => Stream<A, any, any> ? A : never }[keyof Cases], E2 | ExcludeTag<E, K> | A2 extends unassigned ? ExtractTag<E, K> : never | { [RK in string | number | symbol]: Cases[RK] extends (...args: Array<any>) => Stream<any, E, any> ? E : never }[keyof Cases], R2 | R | { [RK in string | number | symbol]: Cases[RK] extends (...args: Array<any>) => Stream<any, any, R> ? R : never }[keyof Cases]>;
  <A, E, R, K extends string, Cases extends { [RK in string]: (reason: ExtractReason<ExtractTag<E, K>, RK>, error: NarrowReason<ExtractTag<E, K>, RK>) => Stream<any, any, any> }, A2 = unassigned, E2 = never, R2 = never>(self: Stream<A, E, R>, errorTag: K, cases: Cases, orElse?: (reason: ExcludeReason<ExtractTag<NoInfer<E>, K>, Extract<keyof Cases, string>>, error: OmitReason<ExtractTag<NoInfer<E>, K>, Extract<keyof Cases, string>>) => Stream<A2, E2, R2>): Stream<A | Exclude<A2, unassigned> | { [RK in string | number | symbol]: Cases[RK] extends (...args: Array<any>) => Stream<A, any, any> ? A : never }[keyof Cases], E2 | ExcludeTag<E, K> | A2 extends unassigned ? ExtractTag<E, K> : never | { [RK in string | number | symbol]: Cases[RK] extends (...args: Array<any>) => Stream<any, E, any> ? E : never }[keyof Cases], R | R2 | { [RK in string | number | symbol]: Cases[RK] extends (...args: Array<any>) => Stream<any, any, R> ? R : never }[keyof Cases]>;
}

Example Catching tagged error reasons

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

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}

class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
  limit: number
}> {}

class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError | QuotaExceededError
}> {}

const stream = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
)

const program = Effect.gen(function*() {
  const values = yield* stream.pipe(
    Stream.catchReasons("AiError", {
      RateLimitError: (reason) => Stream.succeed(`retry: ${reason.retryAfter}`),
      QuotaExceededError: (reason) => Stream.succeed(`quota: ${reason.limit}`)
    }),
    Stream.runCollect
  )
  values // => [ 'retry: 60' ]
})

await Effect.runPromise(program)

catchTag

Since v2.0.0

Recovers from failures whose _tag matches the provided value by switching to the stream returned by f.

When to use

Use when you need to handle a specific error case from a stream whose error type is a tagged union with a readonly _tag field.

Signature

declare const catchTag: {
  <K extends string | readonly [Tags<E>, Tags<E>], E, A1, E1, R1, A2 = unassigned, E2 = never, R2 = never>(k: K, f: (e: ExtractTag<NoInfer<E>, K extends readonly [string, string] ? K[number] : K>) => Stream<A1, E1, R1>, orElse?: (e: ExcludeTag<E, K extends readonly [string, string] ? K[number] : K>) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A1 | A | Exclude<A2, unassigned>, E1 | E2 | A2 extends unassigned ? ExcludeTag<E, K extends readonly [string, string] ? K[number] : K> : never, R1 | R2 | R>;
  <A, E, R, K extends string | readonly [Tags<E>, Tags<E>], R1, E1, A1, A2 = unassigned, E2 = never, R2 = never>(self: Stream<A, E, R>, k: K, f: (e: ExtractTag<E, K extends readonly [string, string] ? K[number] : K>) => Stream<A1, E1, R1>, orElse?: (e: ExcludeTag<E, K extends readonly [string, string] ? K[number] : K>) => Stream<A2, E2, R2>): Stream<A | A1 | Exclude<A2, unassigned>, E1 | E2 | A2 extends unassigned ? ExcludeTag<E, K extends readonly [string, string] ? K[number] : K> : never, R | R1 | R2>;
}

Example Catching tagged failures

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

class HttpError extends Data.TaggedError("HttpError")<{ message: string }> {}

const stream = Stream.fail(new HttpError({ message: "timeout" }))

const recovered = Stream.catchTag(stream, "HttpError", (error) =>
  Stream.make(`Recovered: ${error.message}`)
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(recovered)
  values // => [ 'Recovered: timeout' ]
})

await Effect.runPromise(program)

catchTags

Since v2.0.0

Switches to a recovery stream based on matching _tag handlers.

Signature

declare const catchTags: {
  <E, Cases extends { [K in string]: (error: Extract<E, {
    _tag: K;
  }>) => Stream<any, any, any> } | { [K in string]: (error: Extract<E, {
    _tag: K;
  }>) => Stream<any, any, any> } & { [K in number | symbol]: never }, A2 = unassigned, E2 = never, R2 = never>(cases: Cases, orElse?: (e: Exclude<E, {
    _tag: keyof Cases;
  }>) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A | Exclude<A2, unassigned> | { [K in string | number | symbol]: Cases[K] extends (...args: Array<any>) => Stream<A, any, any> ? A : never }[keyof Cases], E2 | A2 extends unassigned ? Exclude<E, {
    _tag: keyof Cases;
  }> : never | { [K in string | number | symbol]: Cases[K] extends (...args: Array<any>) => Stream<any, E, any> ? E : never }[keyof Cases], R2 | R | { [K in string | number | symbol]: Cases[K] extends (...args: Array<any>) => Stream<any, any, R> ? R : never }[keyof Cases]>;
  <R, E, A, Cases extends { [K in string]: (error: Extract<E, {
    _tag: K;
  }>) => Stream<any, any, any> } | { [K in string]: (error: Extract<E, {
    _tag: K;
  }>) => Stream<any, any, any> } & { [K in number | symbol]: never }, A2 = unassigned, E2 = never, R2 = never>(self: Stream<A, E, R>, cases: Cases, orElse?: (e: Exclude<E, {
    _tag: keyof Cases;
  }>) => Stream<A2, E2, R2>): Stream<A | Exclude<A2, unassigned> | { [K in string | number | symbol]: Cases[K] extends (...args: Array<any>) => Stream<A, any, any> ? A : never }[keyof Cases], E2 | A2 extends unassigned ? Exclude<E, {
    _tag: keyof Cases;
  }> : never | { [K in string | number | symbol]: Cases[K] extends (...args: Array<any>) => Stream<any, E, any> ? E : never }[keyof Cases], R | R2 | { [K in string | number | symbol]: Cases[K] extends (...args: Array<any>) => Stream<any, any, R> ? R : never }[keyof Cases]>;
}

Example Catching tagged failures with handlers

import { Effect, Stream } from "effect"

class NotFound {
  readonly _tag = "NotFound"
  constructor(readonly resource: string) {}
}

class Unauthorized {
  readonly _tag = "Unauthorized"
  constructor(readonly user: string) {}
}

const stream = Stream.fail(new NotFound("profile"))

const program = Effect.gen(function* () {
  const result = yield* stream.pipe(
    Stream.catchTags({
      NotFound: () => Stream.succeed("fallback"),
      Unauthorized: () => Stream.succeed("login")
    }),
    Stream.runCollect
  )
  result // => [ 'fallback' ]
})

await Effect.runPromise(program)

ignore

Since v4.0.0

Ignores failures and ends the stream on error.

When to use

Use when you want a failing stream to end gracefully rather than propagate the error.

Details

The log option controls whether the failure is logged before the stream terminates.

See

ignoreCause for a variant that also ignores defects, not just typed failures

Signature

declare const ignore: <Arg extends Stream<any, any, any> | {
  readonly log?: boolean | Severity;
} | undefined>(selfOrOptions: Arg, options?: {
  readonly log?: boolean | Severity;
}) => [Arg] extends [Stream<infer A, infer _E, infer R>] ? Stream<A, never, R> : <A, E, R>(self: Stream<A, E, R>) => Stream<A, never, R>

Example Ignoring stream failures

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.concat(Stream.fail("boom")),
    Stream.ignore,
    Stream.runCollect
  )
  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

Example Configuring ignore logging

import { Effect, Stream } from "effect"

await Effect.runPromise(Effect.gen(function*() {
  const values = yield* Stream.fail("boom").pipe(
    Stream.ignore({ log: false }),
    Stream.runCollect
  )
  values // => []
}))

ignoreCause

Since v4.0.0

Ignores the stream's failure cause, including defects, and ends the stream.

When to use

Use when you need to silently suppress a stream's entire failure cause, including both typed errors and defects, rather than propagate it downstream.

See

ignore to ignore only typed failures without suppressing defects

Signature

declare const ignoreCause: <Arg extends Stream<any, any, any> | {
  readonly log?: boolean | Severity;
} | undefined>(streamOrOptions: Arg, options?: {
  readonly log?: boolean | Severity;
}) => [Arg] extends [Stream<infer A, infer _E, infer R>] ? Stream<A, never, R> : <A, E, R>(self: Stream<A, E, R>) => Stream<A, never, R>

Example Ignoring stream failure causes

import { Effect, Stream } from "effect"

await Effect.runPromise(Effect.gen(function*() {
  const values = yield* Stream.make(1, 2).pipe(
    Stream.concat(Stream.die("boom")),
    Stream.ignoreCause({ log: false }),
    Stream.runCollect
  )
  values // => [1, 2]
}))

mapError

Since v2.0.0

Transforms the errors emitted by this stream using f.

Signature

declare const mapError: {
  <E, E2>(f: (error: E) => E2): <A, R>(self: Stream<A, E, R>) => Stream<A, E2, R>;
  <A, E, R, E2>(self: Stream<A, E, R>, f: (error: E) => E2): Stream<A, E2, R>;
}

Example Mapping stream errors

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.fail("bad").pipe(
    Stream.mapError((error) => `mapped: ${error}`),
    Stream.catch((error) => Stream.make(`recovered from ${error}`)),
    Stream.runCollect
  )
  result // => [ 'recovered from mapped: bad' ]
})

await Effect.runPromise(program)

orDie

Since v2.0.0

Turns typed failures into defects, making the stream infallible.

Signature

declare function orDie<A, E, R>(self: Stream<A, E, R>): Stream<A, never, R>

Example Turning failures into defects

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.orDie,
    Stream.runCollect
  )

  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

orElseIfEmpty

Since v2.0.0

Switches to a fallback stream if this stream is empty.

Signature

declare const orElseIfEmpty: {
  <E, A2, E2, R2>(orElse: LazyArg<Stream<A2, E2, R2>>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A, E | E2, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, orElse: LazyArg<Stream<A2, E2, R2>>): Stream<A | A2, E | E2, R | R2>;
}

Example Switching on empty streams

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.empty.pipe(
    Stream.orElseIfEmpty(() => Stream.make(1, 2)),
    Stream.runCollect
  )
  values // => [ 1, 2 ]
})

await Effect.runPromise(program)

orElseSucceed

Since v2.0.0

Returns a stream that emits a fallback value when this stream fails.

Signature

declare const orElseSucceed: {
  <E, A2>(f: (error: E) => A2): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A, never, R>;
  <A, E, R, A2>(self: Stream<A, E, R>, f: (error: E) => A2): Stream<A | A2, never, R>;
}

Example Recovering with a fallback value

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.fail("NetworkError").pipe(
    Stream.orElseSucceed((error) => `Recovered: ${error}`)
  )

  const values = yield* Stream.runCollect(stream)
  values // => [ 'Recovered: NetworkError' ]
})

await Effect.runPromise(program)

result

Since v4.0.0

Lifts failures and successes into a Result, yielding a stream that cannot fail.

Details

The stream ends after the first failure, emitting a Result.fail value.

Signature

declare function result<A, E, R>(self: Stream<A, E, R>): Stream<Result<A, E>, never, R>

Example Converting failures to results

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

const program = Effect.gen(function*() {
  const results = yield* Stream.make(1, 2).pipe(
    Stream.concat(Stream.fail("boom")),
    Stream.result,
    Stream.map(Result.match({
      onFailure: (error) => `failure: ${error}`,
      onSuccess: (value) => `success: ${value}`
    })),
    Stream.runCollect
  )
  results // => [ 'success: 1', 'success: 2', 'failure: boom' ]
})

await Effect.runPromise(program)

retry

Since v2.0.0

Retries the stream according to the given schedule when it fails.

Details

This retries the entire stream, so will re-execute all of the stream's acquire operations.

The schedule is reset as soon as the first element passes through the stream again.

Signature

declare const retry: {
  <E, X, E2, R2>(policy: Schedule<X, NoInfer<E>, E2, R2> | ($: <SO, SE, SR>(_: Schedule<SO, NoInfer<E>, SE, SR>) => Schedule<SO, E, SE, SR>) => Schedule<X, NoInfer<E>, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A, E | E2, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, policy: Schedule<X, NoInfer<E>, E2, R2> | ($: <SO, SE, SR>(_: Schedule<SO, NoInfer<E>, SE, SR>) => Schedule<SO, E, SE, SR>) => Schedule<X, NoInfer<E>, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Retrying stream failures

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

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1).pipe(
    Stream.concat(Stream.fail("boom")),
    Stream.retry(Schedule.recurs(1)),
    Stream.take(2),
    Stream.runCollect
  )

  values // => [ 1, 1 ]
})

await Effect.runPromise(program)

tapCause

Since v2.0.0

Runs an effect when the stream fails without changing its values or error, unless the tap effect itself fails.

Signature

declare const tapCause: {
  <E, A2, E2, R2>(f: (cause: Cause<E>) => Effect<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A, E | E2, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (cause: Cause<E>) => Effect<A2, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Tapping stream causes

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

const observations: Array<boolean> = []
const stream = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapCause((cause) => Effect.sync(() => observations.push(Cause.isReason(cause)))),
  Stream.catch(() => Stream.succeed(0))
)

const program = Effect.gen(function* () {
  const result = yield* Stream.runCollect(stream)
  result // => [1, 2, 0]
})

await Effect.runPromise(program)
observations // => [false]

tapDefect

Since v4.0.0

Peeks at defects effectfully without changing the stream unless the tap fails.

See

tapCause for peeking at the full failure cause
catchDefect for recovering from defects

Signature

declare const tapDefect: {
  <B, E2, R2>(f: (defect: unknown) => Effect<B, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, B, E2, R2>(self: Stream<A, E, R>, f: (defect: unknown) => Effect<B, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Effectfully peeking at defects

import { Effect, Stream } from "effect"

const defects: Array<unknown> = []
const stream = Stream.make(1, 2).pipe(
  Stream.concat(Stream.die("boom")),
  Stream.tapDefect((defect) => Effect.sync(() => defects.push(defect))),
  Stream.catchCause(() => Stream.make(3))
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)
defects // => [ 'boom' ]

tapError

Since v2.0.0

Peeks at errors effectfully without changing the stream unless the tap fails.

Signature

declare const tapError: {
  <E, A2, E2, R2>(f: (error: E) => Effect<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A, E | E2, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (error: E) => Effect<A2, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Effectfully peeking at errors

import { Effect, Stream } from "effect"

const errors: Array<string> = []
const stream = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapError((error) => Effect.sync(() => errors.push(error))),
  Stream.catch(() => Stream.make(999))
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 999]
})

await Effect.runPromise(program)
errors // => ["boom"]

tapErrorTag

Since v4.0.0

Peeks at errors with a matching _tag effectfully without changing the stream unless the tap fails.

See

tapError for peeking at every typed error
catchTag for recovering from a tagged error

Signature

declare const tapErrorTag: {
  <K extends string | readonly [Tags<E>, Tags<E>], E, A1, E1, R1>(k: K, f: (e: ExtractTag<NoInfer<E>, K extends readonly [string, string] ? K[number] : K>) => Effect<A1, E1, R1>): <A, R>(self: Stream<A, E, R>) => Stream<A, E | E1, R1 | R>;
  <A, E, R, K extends string | readonly [Tags<E>, Tags<E>], R1, E1, A1>(self: Stream<A, E, R>, k: K, f: (e: ExtractTag<E, K extends readonly [string, string] ? K[number] : K>) => Effect<A1, E1, R1>): Stream<A, E | E1, R | R1>;
}

Example Effectfully peeking at a tagged error

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

class NetworkError extends Data.TaggedError("NetworkError")<{
  statusCode: number
}> {}

class ValidationError extends Data.TaggedError("ValidationError")<{
  field: string
}> {}

const seen: Array<number> = []
const stream: Stream.Stream<number, NetworkError | ValidationError> = Stream.fail(
  new NetworkError({ statusCode: 504 })
)

const program = stream.pipe(
  Stream.tapErrorTag("NetworkError", (error) => Effect.sync(() => seen.push(error.statusCode))),
  Stream.catch(() => Stream.make(0)),
  Stream.runCollect
)

await Effect.runPromise(program) // => [ 0 ]
seen // => [ 504 ]

unwrapReason

Since v4.0.0

Promotes nested reason errors into the stream error channel, replacing the parent error.

See

catchReason for recovering from a specific reason
catchReasons for handling several reasons at once

Signature

declare const unwrapReason: {
  <K extends string, E>(errorTag: K): <A, R>(self: Stream<A, E, R>) => Stream<A, ExcludeTag<E, K> | ReasonOf<ExtractTag<E, K>>, R>;
  <A, E, R, K extends string>(self: Stream<A, E, R>, errorTag: K): Stream<A, ExcludeTag<E, K> | ReasonOf<ExtractTag<E, K>>, R>;
}

Example Extracting the reason from a tagged error

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

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}

class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
  limit: number
}> {}

class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError | QuotaExceededError
}> {}

const stream: Stream.Stream<string, AiError> = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 30 }) })
)

// Before: Stream<string, AiError>
// After:  Stream<string, RateLimitError | QuotaExceededError>
const unwrapped = stream.pipe(Stream.unwrapReason("AiError"))

const program = Effect.gen(function*() {
  const error = yield* Effect.flip(Stream.runCollect(unwrapped))
  error._tag // => "RateLimitError"
})

await Effect.runPromise(program)

withExecutionPlan

unstable

Since v3.16.0

Applies an ExecutionPlan to a stream, retrying with step-provided resources until it succeeds or the plan is exhausted.

Details

By default, a failing step can fallback even after emitting elements; set preventFallbackOnPartialStream to fail instead of mixing partial output with a later fallback.

Attempts can be observed from outside the stream by passing options.onEvent, which receives an ExecutionPlan.Event before each attempt and after it settles; see Effect.withExecutionPlan for the handler semantics. When a downstream consumer stops pulling early (for example Stream.take outside the plan), the truncated attempt reports AttemptSuccess: the consumer stopped, not the source.

Signature

declare const withExecutionPlan: {
  <Input, R2, Provides, PolicyE, RX = never>(policy: ExecutionPlan<{
    error: PolicyE;
    input: Input;
    provides: Provides;
    requirements: R2;
  }>, options?: {
    readonly onEvent?: (event: ExecutionPlan.Event<Input | PolicyE>) => Effect.Effect<void, never, RX>;
    readonly preventFallbackOnPartialStream?: boolean;
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<A, PolicyE | E, R2 | RX | Exclude<R, Provides>>;
  <A, E, R, R2, Input, Provides, PolicyE, RX = never>(self: Stream<A, E, R>, policy: ExecutionPlan<{
    error: PolicyE;
    input: Input;
    provides: Provides;
    requirements: R2;
  }>, options?: {
    readonly onEvent?: (event: ExecutionPlan.Event<E | PolicyE>) => Effect.Effect<void, never, RX>;
    readonly preventFallbackOnPartialStream?: boolean;
  }): Stream<A, E | PolicyE, R2 | RX | Exclude<R, Provides>>;
}

Example Applying an execution plan

import { Context, Effect, ExecutionPlan, Layer, Stream } from "effect"

class Service extends Context.Service<Service>()("Service", {
  make: Effect.succeed({
    stream: Stream.fail("A") as Stream.Stream<number, string>
  })
}) {
  static Bad = Layer.succeed(Service, Service.of({ stream: Stream.fail("A") }))
  static Good = Layer.succeed(Service, Service.of({ stream: Stream.make(1, 2, 3) }))
}

const plan = ExecutionPlan.make(
  { provide: Service.Bad },
  { provide: Service.Good }
)

const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))

const program = Effect.gen(function*() {
  const items = yield* stream.pipe(Stream.withExecutionPlan(plan), Stream.runCollect)
  items // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

Filtering

drop

Since v2.0.0

Drops the first n elements from this stream.

Details

Finite fractional values of n are rounded down. NaN and non-positive values return the source stream unchanged.

Signature

declare const drop: {
  (n: number): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, n: number): Stream<A, E, R>;
}

Example Dropping values from the left

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)
const result = Stream.drop(stream, 2)

const program = Effect.gen(function*() {
  const items = yield* Stream.runCollect(result)
  items // => [ 3, 4, 5 ]
})

await Effect.runPromise(program)

dropRight

Since v2.0.0

Drops the last specified number of elements from this stream.

Details

Keeps the last n elements in memory to drop them on completion. Finite fractional values of n are rounded down. NaN and non-positive values return the source stream unchanged.

Signature

declare const dropRight: {
  (n: number): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, n: number): Stream<A, E, R>;
}

Example Dropping values from the right

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.dropRight(2),
    Stream.runCollect
  )
  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

dropUntil

Since v2.0.0

Drops elements until the specified predicate evaluates to true, then drops that matching element.

Signature

declare const dropUntil: {
  <A>(predicate: (a: NoInfer<A>, index: number) => boolean): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, predicate: (a: NoInfer<A>, index: number) => boolean): Stream<A, E, R>;
}

Example Dropping until a predicate matches

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4, 5)
const result = Stream.dropUntil(stream, (n) => n >= 3)

await Effect.runPromise(Effect.gen(function*() {
  const output = yield* Stream.runCollect(result)
  output // => [ 4, 5 ]
}))

dropUntilEffect

Since v2.0.0

Drops all elements of the stream until the specified effectful predicate evaluates to true.

When to use

Use when dropping the leading prefix requires an Effect or service and the first matching element should also be dropped.

Signature

declare const dropUntilEffect: {
  <A, E2, R2>(predicate: (a: NoInfer<A>, index: number) => Effect<boolean, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, predicate: (a: NoInfer<A>, index: number) => Effect<boolean, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Dropping until an effectful predicate matches

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.range(1, 5).pipe(
    Stream.dropUntilEffect((n) => Effect.succeed(n % 3 === 0)),
    Stream.runCollect
  )
  result // => [ 4, 5 ]
})

await Effect.runPromise(program)

dropWhile

Since v2.0.0

Drops elements from the stream while the specified predicate evaluates to true.

Signature

declare const dropWhile: {
  <A>(predicate: (a: NoInfer<A>, index: number) => boolean): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, predicate: (a: NoInfer<A>, index: number) => boolean): Stream<A, E, R>;
}

Example Dropping while a predicate holds

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.dropWhile((n) => n < 3),
    Stream.runCollect
  )
  values // => [ 3, 4, 5 ]
})

await Effect.runPromise(program)

dropWhileEffect

Since v2.0.0

Drops elements while the specified effectful predicate evaluates to true.

Signature

declare const dropWhileEffect: {
  <A, E2, R2>(predicate: (a: NoInfer<A>, index: number) => Effect<boolean, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, predicate: (a: A, index: number) => Effect<boolean, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Effectfully dropping while a predicate holds

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.dropWhileEffect((n) => Effect.succeed(n < 3)),
    Stream.runCollect
  )
  result // => [ 3, 4, 5 ]
})

await Effect.runPromise(program)

dropWhileFilter

Since v4.0.0

Drops elements while the filter succeeds.

When to use

Use when you need to remove a leading stream prefix based on a synchronous Filter result while preserving the remaining original stream elements.

Details

Result.succeed drops the current element. The first Result.fail stops dropping, emits that original element, and the rest of the source stream is emitted without further filtering.

See

dropWhile for boolean predicate prefix dropping
takeWhileFilter for keeping the accepted prefix as filter success values
dropWhileEffect for effectful predicate prefix dropping

Signature

declare const dropWhileFilter: {
  <A, B, X>(filter: Filter<NoInfer<A>, B, X>): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R, B, X>(self: Stream<A, E, R>, filter: Filter<NoInfer<A>, B, X>): Stream<A, E, R>;
}

filter

Since v2.0.0

Filters a stream to the elements that satisfy a predicate.

Signature

declare const filter: {
  <A, B>(refinement: Refinement<NoInfer<A>, B>): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A>(predicate: Predicate<NoInfer<A>>): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, refinement: Refinement<A, B>): Stream<B, E, R>;
  <A, E, R>(self: Stream<A, E, R>, predicate: Predicate<A>): Stream<A, E, R>;
}

Example Filtering stream values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2, 3, 4).pipe(
    Stream.filter((n) => n % 2 === 0)
  )
  const values = yield* Stream.runCollect(stream)
  values // => [ 2, 4 ]
})

await Effect.runPromise(program)

filterEffect

Since v2.0.0

Filters elements in a single pass effectfully.

Signature

declare const filterEffect: {
  <A, EX, RX>(predicate: (a: NoInfer<A>, i: number) => Effect<boolean, EX, RX>): <E, R>(self: Stream<A, E, R>) => Stream<A, EX | E, RX | R>;
  <A, E, R, EX, RX>(self: Stream<A, E, R>, predicate: (a: NoInfer<A>, i: number) => Effect<boolean, EX, RX>): Stream<A, E | EX, R | RX>;
}

Example Effectfully filtering stream values

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2, 3, 4).pipe(Stream.filterEffect((n) => Effect.succeed(n > 2)))

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [ 3, 4 ]
})

await Effect.runPromise(program)

filterMap

Since v2.0.0

Filters and maps stream elements in one pass using a Filter.

When to use

Use to keep only stream elements accepted by a Filter and emit each filter success value.

Details

Result.succeed values are emitted and Result.fail values are skipped.

See

filter for keeping original elements with a boolean predicate or refinement
filterMapEffect for an effectful Filter
partition for consuming both filter success and failure values

Signature

declare const filterMap: {
  <A, B, X>(filter: Filter<NoInfer<A>, B, X>): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, B, X>(self: Stream<A, E, R>, filter: Filter<A, B, X>): Stream<B, E, R>;
}

filterMapEffect

Since v2.0.0

Filters and maps elements in one pass effectfully using a FilterEffect.

When to use

Use to apply effectful logic that can reject stream elements or emit transformed values before they continue downstream.

Details

Result.succeed values are emitted, Result.fail values are skipped, and effect failures fail the stream.

See

filterMap for the synchronous Filter variant
filterEffect for effectfully keeping original elements
mapEffect for effectfully transforming every element

Signature

declare const filterMapEffect: {
  <A, B, X, EX, RX>(filter: FilterEffect<NoInfer<A>, B, X, EX, RX>): <E, R>(self: Stream<A, E, R>) => Stream<B, EX | E, RX | R>;
  <A, E, R, B, X, EX, RX>(self: Stream<A, E, R>, filter: FilterEffect<A, B, X, EX, RX>): Stream<B, E | EX, R | RX>;
}

limitBytes

unstable

Since v4.0.0

Emits byte chunks until the configured limit would be exceeded, then drops the crossing chunk and switches to a fallback stream.

Signature

declare const limitBytes: {
  <E, R>(bytes: Input, onLimitReached: LazyArg<Stream<Uint8Array<ArrayBufferLike>, E, R>>): (self: Stream<Uint8Array<ArrayBufferLike>, E, R>) => Stream<Uint8Array<ArrayBufferLike>, E, R>;
  <E, R>(self: Stream<Uint8Array<ArrayBufferLike>, E, R>, bytes: Input, onLimitReached: LazyArg<Stream<Uint8Array<ArrayBufferLike>, E, R>>): Stream<Uint8Array<ArrayBufferLike>, E, R>;
}

Example Truncating at a byte limit

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

const program = Stream.make(
  new Uint8Array([1, 2]),
  new Uint8Array([3, 4, 5])
).pipe(
  Stream.limitBytes(ByteSize.bytes(4), () => Stream.empty),
  Stream.runCollect,
  Effect.map((chunks) => chunks.map((chunk) => [...chunk]))
)

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

partition

Since v2.0.0

Splits a stream into scoped passing and failing substreams using a Filter.

Details

The returned streams are backed by queues in the current scope and should be consumed while that scope remains open. The faster stream may advance up to capacity elements ahead of the slower one. The first stream emits the filter successes and the second emits the filter failures. The default capacity is 16.

Signature

declare const partition: {
  <A, Pass, Fail>(filter: Filter<NoInfer<A>, Pass, Fail>, options?: {
    readonly capacity?: number | "unbounded";
  }): <E, R>(self: Stream<A, E, R>) => Effect<[passes: Stream<Pass, E, never>, fails: Stream<Fail, E, never>], never, Scope | R>;
  <A, E, R, Pass, Fail>(self: Stream<A, E, R>, filter: Filter<NoInfer<A>, Pass, Fail>, options?: {
    readonly capacity?: number | "unbounded";
  }): Effect<[passes: Stream<Pass, E, never>, fails: Stream<Fail, E, never>], never, Scope | R>;
}

Example Partitioning a stream

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

const program = Effect.gen(function*() {
  const [passes, fails] = yield* Stream.partition(
    Stream.make(1, 2, 3, 4),
    (n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n)
  )
  const evens = yield* Stream.runCollect(passes)
  const odds = yield* Stream.runCollect(fails)
  evens // => [ 2, 4 ]
  odds // => [ 1, 3 ]
})
await Effect.runPromise(Effect.scoped(program))

partitionEffect

Since v4.0.0

Splits a stream with an effectful Filter, returning scoped streams for filter successes and failures.

When to use

Use when you need to classify each stream element with an effectful Filter and consume both passing and failing mapped values as streams.

Details

The returned streams are backed by queues in the current scope and should be consumed while that scope remains open. The first stream emits success values from the filter, and the second emits failure values.

See

partition for the pure Filter variant
partitionQueue for the lower-level queue result
filterMapEffect for effectful filtering that discards failed filter results

Signature

declare const partitionEffect: {
  <A, Pass, Fail, EX, RX>(filter: FilterEffect<NoInfer<A>, Pass, Fail, EX, RX>, options?: {
    readonly capacity?: number | "unbounded";
    readonly concurrency?: Concurrency;
  }): <E, R>(self: Stream<A, E, R>) => Effect<[passes: Stream<Pass, EX | E, never>, fails: Stream<Fail, EX | E, never>], never, Scope | RX | R>;
  <A, E, R, Pass, Fail, EX, RX>(self: Stream<A, E, R>, filter: FilterEffect<NoInfer<A>, Pass, Fail, EX, RX>, options?: {
    readonly capacity?: number | "unbounded";
    readonly concurrency?: Concurrency;
  }): Effect<[passes: Stream<Pass, E | EX, never>, fails: Stream<Fail, E | EX, never>], never, Scope | R | RX>;
}

partitionQueue

Since v4.0.0

Partitions a stream using a Filter and exposes passing and failing values as scoped queues.

Details

The queues are backed by a fiber in the current scope and should be consumed while that scope remains open. Each queue fails with the stream error or Cause.Done when the source ends.

Signature

declare const partitionQueue: {
  <A, Pass, Fail>(filter: Filter<NoInfer<A>, Pass, Fail>, options?: {
    readonly capacity?: number | "unbounded";
  }): <E, R>(self: Stream<A, E, R>) => Effect<[passes: Dequeue<Pass, Done<void> | E>, fails: Dequeue<Fail, Done<void> | E>], never, Scope | R>;
  <A, E, R, Pass, Fail>(self: Stream<A, E, R>, filter: Filter<NoInfer<A>, Pass, Fail>, options?: {
    readonly capacity?: number | "unbounded";
  }): Effect<[passes: Dequeue<Pass, Done<void> | E>, fails: Dequeue<Fail, Done<void> | E>], never, Scope | R>;
}

Example Partitioning a stream into queues

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

const program = Effect.gen(function*() {
  const [passes, fails] = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.partitionQueue((n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n))
  )

  const passValues = yield* Stream.fromQueue(passes).pipe(Stream.runCollect)
  const failValues = yield* Stream.fromQueue(fails).pipe(Stream.runCollect)

  passValues // => [ 2, 4 ]
  failValues // => [ 1, 3 ]
})

await Effect.runPromise(Effect.scoped(program))

take

Since v2.0.0

Takes the first n elements from this stream.

Details

Finite fractional values of n are rounded down. NaN and non-positive values return Stream.empty without evaluating the source stream.

Signature

declare const take: {
  (n: number): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, n: number): Stream<A, E, R>;
}

Example Taking values from the left

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.take(3),
    Stream.runCollect
  )
  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

takeRight

Since v2.0.0

Keeps the last n elements from this stream.

Details

Finite fractional values of n are rounded down. NaN and non-positive values return Stream.empty without evaluating the source stream.

Signature

declare const takeRight: {
  (n: number): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, n: number): Stream<A, E, R>;
}

Example Taking elements from the right

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.range(1, 6).pipe(
    Stream.takeRight(3),
    Stream.runCollect
  )
  values // => [ 4, 5, 6 ]
})

await Effect.runPromise(program)

takeUntil

Since v2.0.0

Takes elements until the predicate matches.

Details

When excludeLast is true, the matching element is dropped.

Signature

declare const takeUntil: {
  <A>(predicate: (a: NoInfer<A>, n: number) => boolean, options?: {
    readonly excludeLast?: boolean;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, predicate: (a: A, n: number) => boolean, options?: {
    readonly excludeLast?: boolean;
  }): Stream<A, E, R>;
}

Example Taking until a predicate matches

import { Effect, Stream } from "effect"

const stream = Stream.range(1, 5)

const program = Effect.gen(function*() {
  const inclusive = yield* stream.pipe(
    Stream.takeUntil((n) => n % 3 === 0),
    Stream.runCollect
  )
  inclusive // => [ 1, 2, 3 ]

  const exclusive = yield* stream.pipe(
    Stream.takeUntil((n) => n % 3 === 0, { excludeLast: true }),
    Stream.runCollect
  )
  exclusive // => [ 1, 2 ]
})
await Effect.runPromise(program)

takeUntilEffect

Since v2.0.0

Takes stream elements until an effectful predicate returns true.

When to use

Use when the stopping condition needs an Effect or service and predicate failure should fail the stream.

Signature

declare const takeUntilEffect: {
  <A, E2, R2>(predicate: (a: NoInfer<A>, n: number) => Effect<boolean, E2, R2>, options?: {
    readonly excludeLast?: boolean;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, predicate: (a: A, n: number) => Effect<boolean, E2, R2>, options?: {
    readonly excludeLast?: boolean;
  }): Stream<A, E | E2, R | R2>;
}

Example Taking until an effectful predicate matches

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.range(1, 5).pipe(
    Stream.takeUntilEffect((n) => Effect.succeed(n % 3 === 0)),
    Stream.runCollect
  )
  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

takeWhile

Since v2.0.0

Takes the longest initial prefix of elements that satisfy the predicate.

Signature

declare const takeWhile: {
  <A, B>(refinement: (a: NoInfer<A>, n: number) => a is B): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A>(predicate: (a: NoInfer<A>, n: number) => boolean): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, refinement: (a: NoInfer<A>, n: number) => a is B): Stream<B, E, R>;
  <A, E, R>(self: Stream<A, E, R>, predicate: (a: NoInfer<A>, n: number) => boolean): Stream<A, E, R>;
}

Example Taking while a predicate holds

import { Effect, Stream } from "effect"

const stream = Stream.range(1, 5).pipe(
  Stream.takeWhile((n) => n % 3 !== 0)
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [ 1, 2 ]
})

await Effect.runPromise(program)

takeWhileEffect

Since v4.0.0

Takes elements from the stream while the effectful predicate is true.

When to use

Use when the leading-prefix predicate needs an Effect or service and the stream should stop before the first false result.

Signature

declare const takeWhileEffect: {
  <A, E2, R2>(predicate: (a: NoInfer<A>, n: number) => Effect<boolean, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, predicate: (a: NoInfer<A>, n: number) => Effect<boolean, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Effectfully taking while a predicate holds

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.range(1, 5).pipe(
    Stream.takeWhileEffect((n) => Effect.succeed(n % 3 !== 0)),
    Stream.runCollect
  )
  result // => [ 1, 2 ]
})

await Effect.runPromise(program)

takeWhileFilter

Since v4.0.0

Takes the longest initial prefix accepted by a Filter and emits the filter's success values.

When to use

Use to keep the leading stream elements that a Filter accepts, emit the filter's success values, and stop at the first filter failure.

Details

The stream stops at the first Result.fail returned by the filter.

See

takeWhile for keeping original elements with a boolean predicate or refinement
filterMap for filtering across the whole stream instead of only the leading prefix
dropWhileFilter for dropping the accepted prefix and keeping the remaining original elements

Signature

declare const takeWhileFilter: {
  <A, B, X>(f: Filter<NoInfer<A>, B, X>): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, B, X>(self: Stream<A, E, R>, f: Filter<NoInfer<A>, B, X>): Stream<B, E, R>;
}

when

Since v2.0.0

Returns the specified stream if the given condition is satisfied, otherwise returns an empty stream.

Signature

declare const when: {
  <EX = never, RX = never>(test: Effect<boolean, EX, RX>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, EX | E, RX | R>;
  <A, E, R, EX = never, RX = never>(self: Stream<A, E, R>, test: Effect<boolean, EX, RX>): Stream<A, E | EX, R | RX>;
}

Example Conditionally keeping a stream

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(
    Stream.when(Stream.make(1, 2, 3), Effect.succeed(false))
  )
  result // => []
})

await Effect.runPromise(program)

Grouping

chunks

Since v2.0.0

Exposes the underlying chunks as a stream of non-empty arrays.

Signature

declare function chunks<A, E, R>(self: Stream<A, E, R>): Stream<readonly [A, A], E, R>

Example Exposing stream chunks

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const chunks = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.rechunk(2),
    Stream.chunks,
    Stream.runCollect
  )
  chunks // => [ [ 1, 2 ], [ 3, 4 ] ]
})

await Effect.runPromise(program)

groupAdjacentBy

Since v2.0.0

Groups consecutive elements that have equal keys into non-empty arrays.

When to use

Use when you already have a stream ordered by the grouping key and want to emit each consecutive run as a non-empty array while keeping later non-adjacent runs separate.

Details

The key is computed with f; adjacent elements whose keys are equal by Equal.equals are emitted as one [key, group]. Later non-adjacent runs with the same key are emitted separately.

See

groupByKey for grouping all elements with the same key across the stream
groupBy for custom grouped stream construction

Signature

declare const groupAdjacentBy: {
  <A, K>(f: (a: NoInfer<A>) => K): <E, R>(self: Stream<A, E, R>) => Stream<readonly [K, [A, ...Array<A>]], E, R>;
  <A, E, R, K>(self: Stream<A, E, R>, f: (a: NoInfer<A>) => K): Stream<readonly [K, [A, ...Array<A>]], E, R>;
}

groupBy

Since v2.0.0

Groups elements into keyed substreams using an effectful classifier.

Signature

declare const groupBy: {
  <A, K, V, E2, R2>(f: (a: NoInfer<A>) => Effect<readonly [K, V], E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly idleTimeToLive?: Duration.Input;
  }): <E, R>(self: Stream<A, E, R>) => Stream<readonly [K, Stream<V, never, never>], E2 | E, R2 | R>;
  <A, E, R, K, V, E2, R2>(self: Stream<A, E, R>, f: (a: NoInfer<A>) => Effect<readonly [K, V], E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly idleTimeToLive?: Duration.Input;
  }): Stream<readonly [K, Stream<V, never, never>], E | E2, R | R2>;
}

Example Grouping elements into keyed substreams using an effectful classifier

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const grouped = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.groupBy((n) =>
      Effect.succeed([n % 2 === 0 ? "even" : "odd", n] as const)
    ),
    Stream.mapEffect(
      Effect.fnUntraced(function*([key, stream]) {
        return [key, yield* Stream.runCollect(stream)] as const
      }),
      { concurrency: "unbounded" }
    ),
    Stream.runCollect
  )

  grouped // => [ [ 'odd', [ 1, 3, 5 ] ], [ 'even', [ 2, 4 ] ] ]
})

await Effect.runPromise(program)

groupByKey

Since v2.0.0

Groups elements by a key and emits a stream per key.

Signature

declare const groupByKey: {
  <A, K>(f: (a: NoInfer<A>) => K, options?: {
    readonly bufferSize?: number;
    readonly idleTimeToLive?: Duration.Input;
  }): <E, R>(self: Stream<A, E, R>) => Stream<readonly [K, Stream<A, never, never>], E, R>;
  <A, E, R, K>(self: Stream<A, E, R>, f: (a: NoInfer<A>) => K, options?: {
    readonly bufferSize?: number;
    readonly idleTimeToLive?: Duration.Input;
  }): Stream<readonly [K, Stream<A, never, never>], E, R>;
}

Example Grouping elements by key

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const grouped = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.groupByKey((n) => n % 2 === 0 ? "even" : "odd"),
    Stream.mapEffect(
      ([key, stream]) =>
        Stream.runCollect(stream).pipe(
          Effect.map((values) => [key, values] as const)
        ),
      { concurrency: "unbounded" }
    ),
    Stream.runCollect
  )
  grouped // => [ [ 'odd', [ 1, 3, 5 ] ], [ 'even', [ 2, 4 ] ] ]
})

await Effect.runPromise(program)

grouped

Since v2.0.0

Partitions the stream into non-empty arrays of the specified size.

Details

The final array may be smaller if there are not enough elements to fill it. Finite fractional sizes are rounded down. NaN and non-positive sizes are treated as 1.

Signature

declare const grouped: {
  (n: number): <A, E, R>(self: Stream<A, E, R>) => Stream<readonly [A, A], E, R>;
  <A, E, R>(self: Stream<A, E, R>, n: number): Stream<readonly [A, A], E, R>;
}

Example Grouping elements by size

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const grouped = yield* Stream.range(1, 8).pipe(
    Stream.grouped(3),
    Stream.runCollect
  )
  grouped // => [ [ 1, 2, 3 ], [ 4, 5, 6 ], [ 7, 8 ] ]
})

await Effect.runPromise(program)

groupedWithin

Since v2.0.0

Partitions the stream into arrays, emitting when the chunk size is reached or the duration passes.

Details

The duration starts when the first element of a group arrives, so an idle stream does not wake up. Finite fractional chunk sizes are rounded down. NaN and non-positive sizes are treated as 1.

Signature

declare const groupedWithin: {
  (chunkSize: number, duration: Input): <A, E, R>(self: Stream<A, E, R>) => Stream<Array<A>, E, R>;
  <A, E, R>(self: Stream<A, E, R>, chunkSize: number, duration: Input): Stream<Array<A>, E, R>;
}

Example Grouping elements by size or time

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.groupedWithin(2, "5 seconds"),
    Stream.runCollect
  )
  values // => [ [ 1, 2 ], [ 3 ] ]
})

await Effect.runPromise(program)

rechunk

Since v2.0.0

Groups the stream into arrays of the specified size, preserving element order.

Details

Finite fractional sizes are rounded down. NaN and non-positive sizes are treated as 1.

Signature

declare const rechunk: {
  (size: number): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, size: number): Stream<A, E, R>;
}

Example Rechunking stream elements

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.rechunk(2),
    Stream.chunks,
    Stream.runCollect
  )
  result // => [ [ 1, 2 ], [ 3, 4 ], [ 5 ] ]
})

await Effect.runPromise(program)

sliding

Since v2.0.0

Emits a sliding window of n elements.

Details

Finite fractional window sizes are rounded down. NaN and non-positive sizes are treated as 1.

Signature

declare const sliding: {
  (chunkSize: number): <A, E, R>(self: Stream<A, E, R>) => Stream<readonly [A, A], E, R>;
  <A, E, R>(self: Stream<A, E, R>, chunkSize: number): Stream<readonly [A, A], E, R>;
}

Example Emitting sliding windows

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

await Effect.runPromise(Effect.gen(function*() {
  const result = yield* pipe(
    Stream.make(1, 2, 3, 4, 5),
    Stream.sliding(2),
    Stream.runCollect
  )
  result // => [ [ 1, 2 ], [ 2, 3 ], [ 3, 4 ], [ 4, 5 ] ]
}))

slidingSize

Since v2.0.0

Emits sliding windows of chunkSize elements, advancing by stepSize.

Details

Finite fractional window and step sizes are rounded down. NaN and non-positive sizes are treated as 1.

Signature

declare const slidingSize: {
  (chunkSize: number, stepSize: number): <A, E, R>(self: Stream<A, E, R>) => Stream<readonly [A, A], E, R>;
  <A, E, R>(self: Stream<A, E, R>, chunkSize: number, stepSize: number): Stream<readonly [A, A], E, R>;
}

Example Emitting sliding windows with a step size

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const chunks = yield* Stream.make(1, 2, 3, 4, 5).pipe(
    Stream.slidingSize(3, 2),
    Stream.runCollect
  )
  chunks // => [ [ 1, 2, 3 ], [ 3, 4, 5 ] ]
})

await Effect.runPromise(program)

split

Since v2.0.0

Splits the stream into non-empty groups whenever the predicate matches.

Details

Matching elements act as delimiters and are not included in the output.

Signature

declare const split: {
  <A, B>(refinement: Refinement<NoInfer<A>, B>): <E, R>(self: Stream<A, E, R>) => Stream<readonly [Exclude<A, B>, Exclude<A, B>], E, R>;
  <A>(predicate: Predicate<NoInfer<A>>): <E, R>(self: Stream<A, E, R>) => Stream<readonly [A, A], E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, refinement: Refinement<A, B>): Stream<readonly [Exclude<A, B>, Exclude<A, B>], E, R>;
  <A, E, R>(self: Stream<A, E, R>, predicate: Predicate<A>): Stream<readonly [A, A], E, R>;
}

Example Splitting on matching values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.range(0, 9).pipe(
    Stream.split((n) => n % 4 === 0),
    Stream.runCollect
  )
  result // => [ [ 1, 2, 3 ], [ 5, 6, 7 ], [ 9 ] ]
})

await Effect.runPromise(program)

Guards

isStream

Since v4.0.0

Checks whether a value is a Stream.

Signature

declare function isStream(u: unknown): u is Stream<unknown, unknown, unknown>

Example Checking whether a value is a Stream

import { Stream } from "effect"

Stream.isStream(Stream.make(1, 2, 3)) // => true
Stream.isStream({ data: [1, 2, 3] }) // => false

Interruption

haltWhen

Since v2.0.0

Stops a stream after the current pull when an effect completes.

When to use

Use to stop before the next pull after an external signal completes.

Details

The effect is forked, its success value is discarded, and its failure fails the stream.

Gotchas

This does not interrupt or truncate an in-progress pull. A pull may emit multiple elements in a single chunk, in which case the entire chunk is emitted. Use interruptWhen when the stream should be interrupted immediately.

Signature

declare const haltWhen: {
  <X, E2, R2>(effect: Effect<X, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, effect: Effect<X, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Halting a stream after an effect completes

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

const program = Effect.gen(function*() {
  const halt = yield* Deferred.make<void>()
  const values = yield* Stream.fromArray([1, 2, 3]).pipe(
    Stream.tap((value) => value === 2 ? Deferred.succeed(halt, void 0) : Effect.void),
    Stream.haltWhen(Deferred.await(halt)),
    Stream.runCollect
  )
  values // => [ 1, 2 ]
})

await Effect.runPromise(program)

interruptWhen

Since v2.0.0

Interrupts the evaluation of this stream when the provided effect completes. The given effect will be forked as part of this stream, and its success will be discarded. This combinator will also interrupt any in-progress element being pulled from upstream.

Details

If the effect completes with a failure before the stream completes, the returned stream will emit that failure.

Signature

declare const interruptWhen: {
  <X, E2, R2>(effect: Effect<X, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, effect: Effect<X, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Interrupting when an effect completes

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

const program = Effect.gen(function*() {
  const interrupt = yield* Deferred.make<void>()
  const stream = Stream.make(1, 2, 3).pipe(
    Stream.tap((value) =>
      value === 2
        ? Deferred.succeed(interrupt, void 0)
        : Effect.void
    ),
    Stream.interruptWhen(Deferred.await(interrupt))
  )

  const result = yield* Stream.runCollect(stream)
  result // => [ 1 ]
})

await Effect.runPromise(program)

Mapping

as

Since v4.0.0

Replaces every element of the stream with the provided constant value.

See

map for deriving the replacement value from each element

Signature

declare const as: {
  <B>(value: B): <A, E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, value: B): Stream<B, E, R>;
}

Example Replacing stream elements

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.as("x"),
    Stream.runCollect
  )
  values // => [ 'x', 'x', 'x' ]
})

await Effect.runPromise(program)

map

Since v2.0.0

Transforms the elements of this stream using the supplied function.

Signature

declare const map: {
  <A, B>(f: (a: A, i: number) => B): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, f: (a: A, i: number) => B): Stream<B, E, R>;
}

Example Mapping stream values

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

const stream = Stream.fromArray([1, 2, 3]).pipe(Stream.map((n, i) => n + i))
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 3, 5]

mapAccum

Since v2.0.0

Maps elements statefully, emitting zero or more outputs per input.

Signature

declare const mapAccum: {
  <S, A, B>(initial: LazyArg<S>, f: (s: S, a: A) => readonly [S, readonly Array<B>], options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, S, B>(self: Stream<A, E, R>, initial: LazyArg<S>, f: (s: S, a: A) => readonly [S, readonly Array<B>], options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): Stream<B, E, R>;
}

Example Statefully mapping stream values

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

const program = Effect.gen(function*() {
  const totals = yield* Stream.make(0, 1, 2, 3, 4, 5, 6).pipe(
    Stream.mapAccum(() => 0, (total, n) => {
      const next = total + n
      return [next, [next]] as const
    }),
    Stream.runCollect
  )

  totals // => [0, 1, 3, 6, 10, 15, 21]
})

await Effect.runPromise(program)

mapAccumArray

Since v4.0.0

Maps over non-empty chunk arrays statefully, emitting zero or more values per chunk.

Details

The mapping function runs once per chunk and the state is threaded across chunks.

Signature

declare const mapAccumArray: {
  <S, A, B>(initial: LazyArg<S>, f: (s: S, a: readonly [A, A]) => readonly [S, readonly Array<B>], options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, S, B>(self: Stream<A, E, R>, initial: LazyArg<S>, f: (s: S, a: readonly [A, A]) => readonly [S, readonly Array<B>], options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): Stream<B, E, R>;
}

Example Statefully mapping stream chunks

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const output = yield* Stream.make(1, 2, 3, 4, 5, 6).pipe(
    Stream.rechunk(2),
    Stream.mapAccumArray(() => 0, (sum: number, chunk) => {
      const next = chunk.reduce((acc, n) => acc + n, sum)
      return [next, [next]]
    }),
    Stream.runCollect
  )
  output // => [ 3, 10, 21 ]
})

await Effect.runPromise(program)

mapAccumArrayEffect

Since v4.0.0

Maps each non-empty input chunk statefully and effectfully, emitting zero or more output values per chunk.

When to use

Use when stateful mapping should process each emitted non-empty chunk with an Effect instead of each element separately.

Details

The mapping effect receives the current state and chunk, then returns the next state plus the values to emit. The state is threaded across chunks.

Signature

declare const mapAccumArrayEffect: {
  <S, A, B, E2, R2>(initial: LazyArg<S>, f: (s: S, a: readonly [A, A]) => Effect<readonly [S, readonly Array<B>], E2, R2>, options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): <E, R>(self: Stream<A, E, R>) => Stream<B, E2 | E, R2 | R>;
  <A, E, R, S, B, E2, R2>(self: Stream<A, E, R>, initial: LazyArg<S>, f: (s: S, a: readonly [A, A]) => Effect<readonly [S, readonly Array<B>], E2, R2>, options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): Stream<B, E | E2, R | R2>;
}

Example Effectfully mapping stream chunks with state

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const totals = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.rechunk(2),
    Stream.mapAccumArrayEffect(() => 0, (total, chunk) =>
      Effect.gen(function*() {
        const next = chunk.reduce((sum, value) => sum + value, total)
        return [next, [next]] as const
      })
    ),
    Stream.runCollect
  )
  totals // => [ 3, 10 ]
})

await Effect.runPromise(program)

mapAccumEffect

Since v2.0.0

Maps each element statefully and effectfully, emitting zero or more output values per input.

When to use

Use when stateful element mapping needs Effects or can fail while emitting zero or more values per input element.

Details

The mapping effect receives the current state and element, then returns the next state plus the values to emit. The state is threaded through the stream.

Signature

declare const mapAccumEffect: {
  <S, A, B, E2, R2>(initial: LazyArg<S>, f: (s: S, a: A) => Effect<readonly [S, readonly Array<B>], E2, R2>, options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): <E, R>(self: Stream<A, E, R>) => Stream<B, E2 | E, R2 | R>;
  <A, E, R, S, B, E2, R2>(self: Stream<A, E, R>, initial: LazyArg<S>, f: (s: S, a: A) => Effect<readonly [S, readonly Array<B>], E2, R2>, options?: {
    readonly onHalt?: (state: S) => ReadonlyArray<B>;
  }): Stream<B, E | E2, R | R2>;
}

Example Effectfully mapping stream values with state

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 1, 1).pipe(
    Stream.mapAccumEffect(() => 0, (total, n) =>
      Effect.succeed([total + n, [total + n]])
    ),
    Stream.runCollect
  )

  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

mapArray

Since v4.0.0

Transforms each emitted chunk using the provided function, which receives the chunk and its index.

Signature

declare const mapArray: {
  <A, B>(f: (a: readonly [A, A], i: number) => readonly [B, B]): <E, R>(self: Stream<A, E, R>) => Stream<B, E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, f: (a: readonly [A, A], i: number) => readonly [B, B]): Stream<B, E, R>;
}

Example Mapping stream chunks

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

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.rechunk(2),
    Stream.mapArray((chunk, index) => Array.map(chunk, (n) => n + index)),
    Stream.runCollect
  )
  result // => [ 1, 2, 4, 5 ]
})

await Effect.runPromise(program)

mapArrayEffect

Since v4.0.0

Maps over non-empty array chunks emitted by the stream effectfully.

When to use

Use when transformation needs to see and replace each non-empty emitted chunk effectfully instead of mapping individual stream elements.

Signature

declare const mapArrayEffect: {
  <A, B, E2, R2>(f: (a: readonly [A, A], i: number) => Effect<readonly [B, B], E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<B, E2 | E, R2 | R>;
  <A, E, R, B, E2, R2>(self: Stream<A, E, R>, f: (a: readonly [A, A], i: number) => Effect<readonly [B, B], E2, R2>): Stream<B, E | E2, R | R2>;
}

Example Effectfully mapping stream chunks

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

const program = Effect.gen(function*() {
  const result = yield* Stream.fromArray([1, 2, 3, 4]).pipe(
    Stream.rechunk(2),
    Stream.mapArrayEffect((chunk, index) =>
      Effect.succeed(Array.map(chunk, (n) => n + index * 10))
    ),
    Stream.runCollect
  )
  result // => [ 1, 2, 13, 14 ]
})

await Effect.runPromise(program)

mapBoth

Since v2.0.0

Maps both the failure and success channels of a stream.

Signature

declare const mapBoth: {
  <A, A2, E, E2>(options: {
    readonly onElement: (a: A) => A2;
    readonly onError: (e: E) => E2;
  }): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>;
  <A, E, R, A2, E2>(self: Stream<A, E, R>, options: {
    readonly onElement: (a: A) => A2;
    readonly onError: (e: E) => E2;
  }): Stream<A2, E2, R>;
}

Example Mapping both the failure and success channels of a stream

import { Effect, Stream } from "effect"

const mapper = {
  onElement: (value: number) => value * 2,
  onError: (error: string) => `error: ${error}`
}

const program = Effect.gen(function*() {
  const success = yield* Stream.make(1, 2).pipe(
    Stream.mapBoth(mapper),
    Stream.runCollect
  )
  success // => [ 2, 4 ]

  const failure = yield* Stream.fail("boom").pipe(
    Stream.mapBoth(mapper),
    Stream.catch((error: string) => Stream.succeed(error)),
    Stream.runCollect
  )
  failure // => [ 'error: boom' ]
})

await Effect.runPromise(program)

mapEffect

Since v2.0.0

Maps over elements of the stream with the specified effectful function.

When to use

Use when each stream element transformation needs an Effect, service dependency, failure channel, or configured concurrency.

Signature

declare const mapEffect: {
  <A, A2, E2, R2>(f: (a: A, i: number) => Effect<A2, E2, R2>, options?: {
    readonly concurrency?: Concurrency;
    readonly unordered?: boolean;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (a: A, i: number) => Effect<A2, E2, R2>, options?: {
    readonly concurrency?: Concurrency;
    readonly unordered?: boolean;
  }): Stream<A2, E | E2, R | R2>;
}

Example Effectfully mapping stream values

import { Effect, Stream } from "effect"

const events: Array<string> = []
const stream = Stream.make(1, 2, 3)

const mappedStream = stream.pipe(
  Stream.mapEffect((n) =>
    Effect.sync(() => {
      events.push(`Processing: ${n}`)
      return n * 2
    })
  )
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(mappedStream)
  result // => [2, 4, 6]
})

await Effect.runPromise(program)
events // => ["Processing: 1", "Processing: 2", "Processing: 3"]

Merging

combine

Since v2.0.0

Combines elements from this stream and the specified stream by repeatedly applying a stateful function that can pull from either side.

Details

Where possible, prefer Stream.combineArray for a more efficient implementation.

Signature

declare const combine: {
  <A2, E2, R2, S, E, A, A3, E3, R3>(that: Stream<A2, E2, R2>, s: LazyArg<S>, f: (s: S, pullLeft: Pull<A, E, void>, pullRight: Pull<A2, E2, void>) => Effect<readonly [A3, S], E3, R3>): <R>(self: Stream<A, E, R>) => Stream<A3, E3, R2 | R3 | R>;
  <A, E, R, A2, E2, R2, S, A3, E3, R3>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>, s: LazyArg<S>, f: (s: S, pullLeft: Pull<A, E, void>, pullRight: Pull<A2, E2, void>) => Effect<readonly [A3, S], E3, R3>): Stream<A3, E3, R | R2 | R3>;
}

Example Combining streams with state

import { Effect, Stream } from "effect"

const stream = Stream.combine(
  Stream.make("A", "B", "C"),
  Stream.make(1, 2, 3),
  () => true,
  (takeLeft, pullLeft, pullRight) =>
    takeLeft
      ? Effect.map(pullLeft, (value) => [`L:${value}`, false] as const)
      : Effect.map(pullRight, (value) => [`R:${value}`, true] as const)
)

const program = Effect.gen(function*() {
  const output = yield* Stream.runCollect(stream)
  output // => [ 'L:A', 'R:1', 'L:B', 'R:2', 'L:C', 'R:3' ]
})

await Effect.runPromise(program)

combineArray

Since v4.0.0

Combines two streams chunk-by-chunk with a stateful pull function.

When to use

Use to coordinate pulling chunks from two streams when each emitted chunk depends on both sides and local state.

Details

The combining function receives the current state and pull functions for the left and right streams. It returns the next non-empty chunk together with the next state.

Signature

declare const combineArray: {
  <A2, E2, R2, S, E, A, A3, E3, R3>(that: Stream<A2, E2, R2>, s: LazyArg<S>, f: (s: S, pullLeft: Pull<readonly [A, A], E, void>, pullRight: Pull<readonly [A2, A2], E2, void>) => Effect<readonly [readonly [A3, A3], S], E3, R3>): <R>(self: Stream<A, E, R>) => Stream<A3, Exclude<E3, Done<any>>, R2 | R3 | R>;
  <R, A2, E2, R2, S, E, A, A3, E3, R3>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>, s: LazyArg<S>, f: (s: S, pullLeft: Pull<readonly [A, A], E, void>, pullRight: Pull<readonly [A2, A2], E2, void>) => Effect<readonly [readonly [A3, A3], S], E3, R3>): Stream<A3, Exclude<E3, Done<any>>, R | R2 | R3>;
}

Example Combining stream chunks with state

import { Effect, Stream } from "effect"

const stream = Stream.make(1, 2).pipe(
  Stream.combineArray(
    Stream.make(10, 20),
    () => true,
    (useLeft, pullLeft, pullRight) =>
      Effect.gen(function*() {
        const array = useLeft ? yield* pullLeft : yield* pullRight
        return [array, !useLeft] as const
      })
  )
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [ 1, 2, 10, 20 ]
})

await Effect.runPromise(program)

interleave

Since v2.0.0

Interleaves this stream with the specified stream by alternating pulls from each stream; when one ends, the remaining values from the other stream are emitted.

Signature

declare const interleave: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<A | A2, E | E2, R | R2>;
}

Example Interleaving streams

import { Effect, Stream } from "effect"

const stream = Stream.interleave(
  Stream.make(2, 3),
  Stream.make(5, 6, 7)
)

const program = Effect.gen(function*() {
  const collected = yield* Stream.runCollect(stream)
  collected // => [ 2, 5, 3, 6, 7 ]
})

await Effect.runPromise(program)

interleaveWith

Since v2.0.0

Interleaves two streams deterministically by following a boolean decider stream.

Details

The decider controls how many elements are pulled; if one side ends, pulls for that side are ignored.

Signature

declare const interleaveWith: {
  <A2, E2, R2, E3, R3>(that: Stream<A2, E2, R2>, decider: Stream<boolean, E3, R3>): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2 | E3 | E, R2 | R3 | R>;
  <A, E, R, A2, E2, R2, E3, R3>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>, decider: Stream<boolean, E3, R3>): Stream<A | A2, E | E2 | E3, R | R2 | R3>;
}

Example Interleaving two streams deterministically by following a boolean decider stream

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const left = Stream.make(1, 3, 5)
  const right = Stream.make(2, 4, 6)
  const decider = Stream.make(true, false, false, true, true)

  const values = yield* Stream.runCollect(
    Stream.interleaveWith(left, right, decider)
  )

  values // => [ 1, 2, 4, 3, 5 ]
})

await Effect.runPromise(program)

merge

Since v2.0.0

Merges two streams, emitting elements from both as they arrive.

Details

By default, the merged stream ends when both streams end. Use haltStrategy to change the termination behavior.

Signature

declare const merge: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>, options?: {
    readonly haltStrategy?: HaltStrategy;
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>, options?: {
    readonly haltStrategy?: HaltStrategy;
  }): Stream<A | A2, E | E2, R | R2>;
}

Example Merging stream values

import { Effect, Stream } from "effect"

const fast = Stream.make(1, 2, 3)
const slow = Stream.fromEffect(Effect.delay(Effect.succeed(4), "50 millis"))

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(Stream.merge(fast, slow))
  result // => [ 1, 2, 3, 4 ]
})

await Effect.runPromise(program)

mergeAll

Since v2.0.0

Merges a collection of streams, running up to the specified number concurrently.

When to use

Use to merge an iterable of already-created streams while bounding how many inner streams may run at the same time.

Details

The concurrency option is required and may be a number or "unbounded". bufferSize controls buffering between inner streams, and outputs are emitted as they arrive under concurrent merging.

See

merge for merging exactly two streams and choosing a halt strategy
flatten for flattening a stream that already emits streams

Signature

declare const mergeAll: {
  (options: {
    readonly bufferSize?: number;
    readonly concurrency: Concurrency;
  }): <A, E, R>(streams: Iterable<Stream<A, E, R>>) => Stream<A, E, R>;
  <A, E, R>(streams: Iterable<Stream<A, E, R>>, options: {
    readonly bufferSize?: number;
    readonly concurrency: Concurrency;
  }): Stream<A, E, R>;
}

Example Merging streams with bounded concurrency

import { Effect, Stream } from "effect"

const streams = [
  Stream.fromEffect(Effect.delay(Effect.succeed("A"), "20 millis")),
  Stream.fromEffect(Effect.delay(Effect.succeed("B"), "10 millis"))
]

const program = Effect.gen(function*() {
  const values = yield* Stream.mergeAll(streams, { concurrency: 2 }).pipe(
    Stream.runCollect
  )
  values // => [ 'B', 'A' ]
})

await Effect.runPromise(program)

mergeEffect

Since v4.0.0

Merges this stream with a background effect, keeping the stream's elements.

When to use

Use when an effect should run concurrently for the lifetime of a stream while only the stream's elements remain in the output.

Details

The effect runs concurrently, fails the stream if it fails, and is interrupted when the stream completes.

Signature

declare const mergeEffect: {
  <A2, E2, R2>(effect: Effect<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, effect: Effect<A2, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Merging with a background effect

import { Effect, Stream } from "effect"

const events: Array<string> = []
const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.mergeEffect(Effect.sync(() => events.push("side task"))),
    Stream.runCollect
  )

  values // => [1, 2, 3]
})

await Effect.runPromise(program)
events // => ["side task"]

mergeLeft

Since v2.0.0

Merges two streams while emitting only the values from the left stream.

When to use

Use when the right stream is needed for its effects or failures, but downstream consumers should only receive values from the left stream.

Details

The right stream still runs for its effects, and any failures from the right stream are propagated. The merged stream completes when the left stream completes, interrupting the right stream.

Signature

declare const mergeLeft: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<AL, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<AL, EL | ER, RL | RR>;
}

Example Merging streams while keeping left values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const left = Stream.make(1, 2)
  const right = Stream.make("a", "b")
  const values = yield* left.pipe(Stream.mergeLeft(right), Stream.runCollect)
  values // => [ 1, 2 ]
})

await Effect.runPromise(program)

mergeResult

Since v4.0.0

Merges this stream and the specified stream together, tagging values from the left stream as Result.succeed and values from the right stream as Result.fail.

When to use

Use when values from both streams should be emitted and downstream code needs left values wrapped as successful Result values and right values wrapped as failed Result values.

Signature

declare const mergeResult: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<Result<A, A2>, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<Result<A, A2>, E | E2, R | R2>;
}

Example Merging streams into results

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

const left = Stream.fromEffect(Effect.succeed("left"))
const right = Stream.fromEffect(Effect.delay(Effect.succeed("right"), "10 millis"))

const merged = left.pipe(
  Stream.mergeResult(right),
  Stream.map(
    Result.match({
      onFailure: (value) => `right:${value}`,
      onSuccess: (value) => `left:${value}`
    })
  )
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(merged)
  result // => [ 'left:left', 'right:right' ]
})

await Effect.runPromise(program)

mergeRight

Since v2.0.0

Merges this stream and the specified stream together, emitting only the values from the right stream while the left stream runs for its effects.

When to use

Use when the left stream is needed for its effects or failures, but downstream consumers should only receive values from the right stream.

Details

The merged stream ends when the right stream completes, interrupting the left stream. Failures from the left stream still fail the merged stream.

Signature

declare const mergeRight: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<AR, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<AR, EL | ER, RL | RR>;
}

Example Merging streams while keeping right values

import { Effect, Stream } from "effect"

const left = Stream.make("left-1", "left-2").pipe(
  Stream.tap(() => Effect.sync(() => undefined))
)
const right = Stream.make(1, 2)

const merged = Stream.mergeRight(left, right)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(merged)
  result // => [ 1, 2 ]
})

await Effect.runPromise(program)

Models

EventListener

interface

Since v3.4.0

Interface representing an event listener target.

Signature

interface EventListener<A = unknown> {
  addEventListener(event: string, f: (event: A) => void, options?: boolean | {
    readonly capture?: boolean;
    readonly once?: boolean;
    readonly passive?: boolean;
    readonly signal?: AbortSignal;
  }): void;
  removeEventListener(event: string, f: (event: A) => void, options?: boolean | {
    readonly capture?: boolean;
  }): void;
}

HaltStrategy

type

Since v4.0.0

Describes how merged streams decide when to halt.

Signature

type HaltStrategy = Channel.HaltStrategy

Stream

interface

Since v2.0.0

A Stream<A, E, R> describes a program that can emit many A values, fail with E, and require R.

Details

Streams are pull-based with backpressure and emit chunks to amortize effect evaluation. They support monadic composition and error handling similar to Effect, adapted for multiple values.

Signature

interface Stream<out A, out E = never, out R = never> extends Variance<A, E, R>, Pipeable {
  [ignoreSymbol]?: StreamUnifyIgnore;
  [typeSymbol]?: unknown;
  [unifySymbol]?: StreamUnify<Stream<A, E, R>>;
  readonly channel: Channel<readonly [A, A], E, void, unknown, unknown, unknown, R>;
}

Example Creating and consuming streams

import { Effect, Stream } from "effect"

const values = await Effect.runPromise(
  Stream.make(1, 2, 3).pipe(
    Stream.map((n) => n * 2),
    Stream.runCollect
  )
)
values // => [2, 4, 6]

StreamUnify

interface

Since v2.0.0

Type-level unification hook for Stream within the Effect type system.

Signature

interface StreamUnify<A extends {
  [typeSymbol]?: any;
}> extends EffectUnify<A> {
  Stream?: () => A[typeof typeSymbol] extends Stream<A0, E0, R0> | _ ? Stream<A0, E0, R0> : never;
}

StreamUnifyIgnore

interface

Since v2.0.0

Type-level marker that excludes Stream from unification.

Signature

interface StreamUnifyIgnore {
  Effect?: true;
}

Variance

interface

Since v2.0.0

Type-level variance marker for Stream.

Details

The emitted value A, error E, and service requirement R type parameters are covariant.

Signature

interface Variance<out A, out E, out R> {
  readonly "~effect/Stream": VarianceStruct<A, E, R>;
}

VarianceStruct

interface

Since v3.4.0

Structural encoding used by Variance to record each Stream type parameter's variance.

Details

_A, _E, and _R are covariant markers.

Signature

interface VarianceStruct<out A, out E, out R> {
  readonly _A: Covariant<A>;
  readonly _E: Covariant<E>;
  readonly _R: Covariant<R>;
}

Other

catch

Signature

declare const catch: {
  <E, A2, E2, R2>(f: (error: E) => Stream<A2, E2, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (error: E) => Stream<A2, E2, R2>): Stream<A | A2, E2, R | R2>;
}

let

Signature

declare const let: {
  <N extends string, A extends Record<string, any>, B>(name: N, f: (a: NoInfer<A>) => B): <E, R>(self: Stream<A, E, R>) => Stream<Simplify<Omit<A, N> & Record<N, B>>, E, R>;
  <A extends Record<string, any>, E, R, B, N extends string>(self: Stream<A, E, R>, name: N, f: (a: NoInfer<A>) => B): Stream<Simplify<Omit<A, N> & Record<N, B>>, E, R>;
}

Providing Services

provide

Since v4.0.0

Provides a layer or context to the stream, removing the corresponding service requirements. Use options.local to build the layer every time; by default, layers are shared between provide calls.

Signature

declare const provide: {
  <AL, EL = never, RL = never>(layer: Layer<AL, EL, RL> | Context<AL>, options?: {
    readonly local?: boolean;
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<A, EL | E, RL | Exclude<R, AL>>;
  <A, E, R, AL, EL = never, RL = never>(self: Stream<A, E, R>, layer: Layer<AL, EL, RL> | Context<AL>, options?: {
    readonly local?: boolean;
  }): Stream<A, E | EL, RL | Exclude<R, AL>>;
}

Example Providing stream requirements

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

class Env extends Context.Service<Env, { readonly name: string }>()("Env") {}

const layer = Layer.succeed(Env)({ name: "Ada" })

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const env = yield* Effect.service(Env)
    return `Hello, ${env.name}`
  })
)

const withEnv = stream.pipe(Stream.provide(layer))

await Effect.runPromise(Stream.runCollect(withEnv)) // => ["Hello, Ada"]

provideContext

Since v2.0.0

Provides multiple services to the stream using a context.

Signature

declare const provideContext: {
  <R2>(context: Context<R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, Exclude<R, R2>>;
  <A, E, R, R2>(self: Stream<A, E, R>, context: Context<R2>): Stream<A, E, Exclude<R, R2>>;
}

Example Providing multiple services to the stream using a context

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

class Config extends Context.Service<Config, { readonly prefix: string }>()("Config") {}
class Greeter extends Context.Service<Greeter, { greet: (name: string) => string }>()("Greeter") {}

const context = Context.make(Config, { prefix: "Hello" }).pipe(
  Context.add(Greeter, { greet: (name: string) => `${name}!` })
)

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const config = yield* Effect.service(Config)
    const greeter = yield* Effect.service(Greeter)
    return greeter.greet(config.prefix)
  })
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(Stream.provideContext(stream, context))
  result // => [ 'Hello!' ]
})

await Effect.runPromise(program)

provideService

Since v2.0.0

Provides the stream with a single required service, eliminating that requirement from its environment.

Signature

declare const provideService: {
  <I, S>(key: Key<I, S>, service: NoInfer<S>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, Exclude<R, I>>;
  <A, E, R, I, S>(self: Stream<A, E, R>, key: Key<I, S>, service: NoInfer<S>): Stream<A, E, Exclude<R, I>>;
}

Example Providing a stream service

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

class Greeter extends Context.Service<Greeter, {
  greet: (name: string) => string
}>()("Greeter") {}

const stream = Stream.fromEffect(
  Effect.service(Greeter).pipe(
    Effect.map((greeter) => greeter.greet("Ada"))
  )
)

const program = Effect.gen(function*() {
  const collected = yield* Stream.runCollect(
    stream.pipe(
      Stream.provideService(Greeter, {
        greet: (name) => `Hello, ${name}`
      })
    )
  )
  collected // => [ 'Hello, Ada' ]
})

await Effect.runPromise(program)

provideServiceEffect

Since v2.0.0

Provides a service to the stream using an effect, removing the requirement and adding the effect's error and environment.

Signature

declare const provideServiceEffect: {
  <I, S, ES, RS>(key: Key<I, S>, service: Effect<NoInfer<S>, ES, RS>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, ES | E, RS | Exclude<R, I>>;
  <A, E, R, I, S, ES, RS>(self: Stream<A, E, R>, key: Key<I, S>, service: Effect<NoInfer<S>, ES, RS>): Stream<A, E | ES, RS | Exclude<R, I>>;
}

Example Providing a stream service effectfully

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

class ApiConfig extends Context.Service<ApiConfig, { readonly baseUrl: string }>()("ApiConfig") {}

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const config = yield* Effect.service(ApiConfig)
    return config.baseUrl
  })
)

const events: Array<string> = []
const withConfig = stream.pipe(
  Stream.provideServiceEffect(
    ApiConfig,
    Effect.succeed({ baseUrl: "https://example.com" }).pipe(
      Effect.tap(() => Effect.sync(() => events.push("loading")))
    )
  )
)

await Effect.runPromise(Stream.runCollect(withConfig)) // => ["https://example.com"]
events // => ["loading"]

updateContext

Since v4.0.0

Transforms the stream's required services by mapping the current context to a new one.

Signature

declare const updateContext: {
  <R, R2>(f: (context: Context<R2>) => Context<R>): <A, E>(self: Stream<A, E, R>) => Stream<A, E, R2>;
  <A, E, R, R2>(self: Stream<A, E, R>, f: (context: Context<R2>) => Context<R>): Stream<A, E, R2>;
}

Example Updating the stream context

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

class Logger extends Context.Service<Logger, { prefix: string }>()("Logger") {}
class Config extends Context.Service<Config, { name: string }>()("Config") {}

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const logger = yield* Effect.service(Logger)
    const config = yield* Effect.service(Config)
    return `${logger.prefix}${config.name}`
  })
)

const updated = stream.pipe(
  Stream.updateContext((context: Context.Context<Logger>) =>
    Context.add(context, Config, { name: "World" })
  )
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(updated)
  values // => [ 'Hello World' ]
})

await Effect.runPromise(
  Effect.provideService(program, Logger, { prefix: "Hello " })
)

updateService

Since v2.0.0

Updates a single service in the stream environment by applying a function.

Signature

declare const updateService: {
  <I, S>(key: Key<I, S>, f: (service: NoInfer<S>) => S): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, I | R>;
  <A, E, R, I, S>(self: Stream<A, E, R>, key: Key<I, S>, f: (service: NoInfer<S>) => S): Stream<A, E, R | I>;
}

Example Updating a stream service

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

class Counter extends Context.Service<Counter, { count: number }>()("Counter") {}

const stream = Stream.fromEffect(Effect.service(Counter)).pipe(
  Stream.updateService(Counter, (counter) => ({ count: counter.count + 1 }))
)

const program = Effect.gen(function*() {
  const counters = yield* Stream.runCollect(stream)
  const message = `Updated count: ${counters[0].count}` // => "Updated count: 1"
})

await Effect.runPromise(Effect.provideService(program, Counter, { count: 0 }))

Racing

race

Since v3.7.0

Runs both streams concurrently until one stream emits its first value, then mirrors that winning stream and interrupts the other.

Details

A failure or completion from one side before the other side emits does not win the race unless both sides fail or complete before emitting. After a winner is chosen, that stream's later failures are propagated.

Signature

declare const race: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<AR | AL, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<AL | AR, EL | ER, RL | RR>;
}

Example Racing two streams

import { Effect, Stream } from "effect"

const stream = Stream.race(
  Stream.empty,
  Stream.make(0, 1, 2)
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [ 0, 1, 2 ]
})

await Effect.runPromise(program)

raceAll

Since v3.5.0

Runs all streams concurrently until one stream emits its first value, then mirrors that winning stream and interrupts the rest.

Details

Failures or completion from losing streams before a winner is chosen are ignored unless every stream fails or completes before emitting. After a winner is chosen, that stream's later failures are propagated.

Signature

declare function raceAll<S extends readonly Array<Stream<any, any, any>>>(...streams: S): Stream<Success<S[number]>, Error<S[number]>, Services<S[number]>>

Example Racing multiple streams

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

const program = Effect.gen(function*() {
  const result = yield* Stream.raceAll(
    Stream.empty,
    Stream.make(0, 1, 2)
  ).pipe(Stream.runCollect)
  result // => [ 0, 1, 2 ]
})

await Effect.runPromise(program)

Rate Limiting

debounce

Since v2.0.0

Drops earlier elements within the debounce window and emits only the latest element after the pause.

Signature

declare const debounce: {
  (duration: Input): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, duration: Input): Stream<A, E, R>;
}

Example Debouncing stream elements

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

const stream = Stream.make(1, 2, 3).pipe(Stream.debounce(Duration.zero))

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [ 3 ]
})
await Effect.runPromise(program)

schedule

Since v2.0.0

Schedules the stream's elements according to the provided schedule.

Signature

declare const schedule: {
  <X, E2, R2, A>(schedule: Schedule<X, NoInfer<A>, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, schedule: Schedule<X, NoInfer<A>, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Scheduling stream elements

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

const program = Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3).pipe(
    Stream.schedule(Schedule.recurs(3)),
    Stream.runCollect
  )

  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

throttle

Since v2.0.0

Rate-limits stream chunks with a synchronous cost function.

When to use

Use to throttle chunks when each chunk's cost can be computed synchronously.

Details

Uses a token bucket. The bucket can accumulate up to units + burst tokens, and each chunk consumes the cost returned by cost.

If using the "enforce" strategy, arrays that do not meet the bandwidth constraints are dropped. If using the "shape" strategy, arrays are delayed until they can be emitted without exceeding the bandwidth constraints.

Defaults to the "shape" strategy.

Signature

declare const throttle: {
  <A>(options: {
    readonly burst?: number;
    readonly cost: (arr: Arr.NonEmptyReadonlyArray<A>) => number;
    readonly duration: Duration.Input;
    readonly strategy?: "enforce" | "shape";
    readonly units: number;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A, E, R>;
  <A, E, R>(self: Stream<A, E, R>, options: {
    readonly burst?: number;
    readonly cost: (arr: Arr.NonEmptyReadonlyArray<A>) => number;
    readonly duration: Duration.Input;
    readonly strategy?: "enforce" | "shape";
    readonly units: number;
  }): Stream<A, E, R>;
}

Example Throttling stream chunks

import { Effect, Stream } from "effect"

const stream = Stream.range(0, 5).pipe(
  Stream.rechunk(1),
  Stream.throttle({
    cost: (arr) => arr.length,
    units: 1,
    duration: 0,
    strategy: "shape"
  })
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [ 0, 1, 2, 3, 4, 5 ]
})
await Effect.runPromise(program)

throttleEffect

Since v2.0.0

Rate-limits stream chunks with an effectful cost function.

When to use

Use to throttle chunks when computing each chunk's cost requires an effect.

Details

Uses a token bucket. The bucket can accumulate up to units + burst tokens, and each chunk consumes the cost returned by the effectful cost function.

Defaults to the "shape" strategy.

Signature

declare const throttleEffect: {
  <A, E2, R2>(options: {
    readonly burst?: number;
    readonly cost: (arr: Arr.NonEmptyReadonlyArray<A>) => Effect.Effect<number, E2, R2>;
    readonly duration: Duration.Input;
    readonly strategy?: "enforce" | "shape";
    readonly units: number;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, options: {
    readonly burst?: number;
    readonly cost: (arr: Arr.NonEmptyReadonlyArray<A>) => Effect.Effect<number, E2, R2>;
    readonly duration: Duration.Input;
    readonly strategy?: "enforce" | "shape";
    readonly units: number;
  }): Stream<A, E | E2, R | R2>;
}

Example Throttling stream chunks effectfully

import { Effect, Stream } from "effect"

const stream = Stream.range(0, 5).pipe(
  Stream.rechunk(1),
  Stream.throttleEffect({
    cost: (arr) => Effect.succeed(arr.length),
    units: 1,
    duration: 0,
    strategy: "shape"
  })
)

await Effect.runPromise(Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [ 0, 1, 2, 3, 4, 5 ]
}))

Repetition

forever

Since v2.0.0

Repeats this stream forever.

Signature

declare function forever<A, E, R>(self: Stream<A, E, R>): Stream<A, E, R>

Example Repeating a stream forever

import { Effect, Stream } from "effect"

const stream = Stream.make("A", "B").pipe(
  Stream.forever,
  Stream.take(5)
)

const program = Effect.gen(function*() {
  const output = yield* Stream.runCollect(stream)
  output // => [ 'A', 'B', 'A', 'B', 'A' ]
})

await Effect.runPromise(program)

repeat

Since v2.0.0

Repeats the entire stream according to the provided schedule.

Signature

declare const repeat: {
  <B, E2, R2>(schedule: Schedule<B, void, E2, R2> | ($: <SO, SE, SR>(_: Schedule<SO, void, SE, SR>) => Schedule<SO, void, SE, SR>) => Schedule<B, void, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, B, E2, R2>(self: Stream<A, E, R>, schedule: Schedule<B, void, E2, R2> | ($: <SO, SE, SR>(_: Schedule<SO, void, SE, SR>) => Schedule<SO, void, SE, SR>) => Schedule<B, void, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Repeating a stream on a schedule

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

const program = Effect.gen(function* () {
  const result = yield* Stream.make(1).pipe(
    Stream.repeat(Schedule.recurs(4)),
    Stream.runCollect
  )

  result // => [ 1, 1, 1, 1, 1 ]
})

await Effect.runPromise(program)

repeatElements

Since v2.0.0

Repeats each element of the stream according to the provided schedule, including the original emission.

Signature

declare const repeatElements: {
  <B, E2, R2>(schedule: Schedule<B, unknown, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, B, E2, R2>(self: Stream<A, E, R>, schedule: Schedule<B, unknown, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Repeating stream elements

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

const program = Effect.gen(function*() {
  const values = yield* Stream.make("A", "B", "C").pipe(
    Stream.repeatElements(Schedule.recurs(1)),
    Stream.runCollect
  )
  values // => [ 'A', 'A', 'B', 'B', 'C', 'C' ]
})

await Effect.runPromise(program)

Resource Management

ensuring

Since v2.0.0

Executes the provided finalizer after this stream's finalizers run.

Signature

declare const ensuring: {
  <R2>(finalizer: Effect<unknown, never, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, R2 | R>;
  <A, E, R, R2>(self: Stream<A, E, R>, finalizer: Effect<unknown, never, R2>): Stream<A, E, R | R2>;
}

Example Ensuring finalization

import { Effect, Stream } from "effect"

const events: Array<string> = []
const stream = Stream.fromArray([1, 2]).pipe(
  Stream.ensuring(Effect.sync(() => events.push("cleanup")))
)

const program = Effect.gen(function*() {
  const collected = yield* Stream.runCollect(stream)
  collected // => [1, 2]
})

await Effect.runPromise(program)
events // => ["cleanup"]

onError

Since v2.0.0

Runs the provided effect when the stream fails, passing the failure cause.

Gotchas

Note: Unlike Effect.onError there is no guarantee that the provided effect will not be interrupted.

Signature

declare const onError: {
  <E, X, R2>(cleanup: (cause: Cause<E>) => Effect<X, never, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A, E, R2 | R>;
  <A, E, R, X, R2>(self: Stream<A, E, R>, cleanup: (cause: Cause<E>) => Effect<X, never, R2>): Stream<A, E, R | R2>;
}

Example Running an effect on errors

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

const errors: Array<string> = []
const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2, 3).pipe(
    Stream.concat(Stream.fail("boom")),
    Stream.onError((cause) => Effect.sync(() => errors.push(String(Cause.squash(cause)))))
  )

  yield* Stream.runCollect(stream)
})

await Effect.runPromise(Effect.exit(program))
errors // => ["boom"]

onExit

Since v4.0.0

Runs the provided finalizer when the stream exits, passing the exit value.

Signature

declare const onExit: {
  <E, R2>(finalizer: (exit: Exit<unknown, E>) => Effect<unknown, never, R2>): <A, R>(self: Stream<A, E, R>) => Stream<A, E, R2 | R>;
  <A, E, R, R2>(self: Stream<A, E, R>, finalizer: (exit: Exit<unknown, E>) => Effect<unknown, never, R2>): Stream<A, E, R | R2>;
}

Example Running a finalizer on exit

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

const exits: Array<string> = []
const stream = Stream.make(1, 2, 3).pipe(
  Stream.onExit((exit) =>
    Exit.isSuccess(exit)
      ? Effect.sync(() => exits.push("success"))
      : Effect.sync(() => exits.push("failure"))
  )
)

await Effect.runPromise(Effect.gen(function*() {
  yield* Stream.runCollect(stream)
}))
exits // => ["success"]

scoped

Since v2.0.0

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

Signature

declare function scoped<A, E, R>(self: Stream<A, E, R>): Stream<A, E, Exclude<R, Scope>>

Example Scoping a stream

import { Effect, Stream } from "effect"

const events: Array<string> = []
const stream = Stream.scoped(
  Stream.fromEffect(
    Effect.acquireRelease(
      Effect.sync(() => {
        events.push("acquire")
        return "resource"
      }),
      () => Effect.sync(() => events.push("release"))
    )
  )
)

await Effect.runPromise(Stream.runCollect(stream)) // => ["resource"]
events // => ["acquire", "release"]

Sequencing

concat

Since v2.0.0

Concatenates two streams, emitting all elements from the first stream followed by all elements from the second stream.

Signature

declare const concat: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<A | A2, E | E2, R | R2>;
}

Example Concatenating streams

import { Effect, Stream } from "effect"

const stream = Stream.concat(Stream.make(1, 2, 3), Stream.make(4, 5, 6))

await Effect.runPromise(Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [ 1, 2, 3, 4, 5, 6 ]
}))

drain

Since v2.0.0

Converts this stream to one that runs its effects but emits no elements.

Signature

declare function drain<A, E, R>(self: Stream<A, E, R>): Stream<never, E, R>

Example Draining stream values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.range(1, 6).pipe(Stream.drain, Stream.runCollect)
  result // => []
})

await Effect.runPromise(program)

drainFork

Since v2.0.0

Runs the provided stream in the background while this stream runs, interrupting it when this stream completes and failing if the background stream fails or defects.

Signature

declare const drainFork: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Draining a stream in the background

import { Effect, Stream } from "effect"

const events: Array<string> = []
const foreground = Stream.make(1, 2)
const background = Stream.fromEffect(Effect.sync(() => events.push("background task")))

const program = Effect.gen(function*() {
  const values = yield* foreground.pipe(
    Stream.drainFork(background),
    Stream.runCollect
  )
  values // => [1, 2]
})

await Effect.runPromise(program)
events // => ["background task"]

flatMap

Since v2.0.0

Maps each element to a stream and flattens the resulting streams.

Details

With the default sequential concurrency, inner streams are concatenated in input order. When concurrency is greater than 1 or "unbounded", multiple inner streams may run at the same time and their outputs are merged as they arrive.

Signature

declare const flatMap: {
  <A, A2, E2, R2>(f: (a: A) => Stream<A2, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (a: A) => Stream<A2, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
  }): Stream<A2, E | E2, R | R2>;
}

Example Flat mapping stream values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.flatMap((n) => Stream.make(n, n * 2)),
    Stream.runCollect
  )
  values // => [ 1, 2, 2, 4, 3, 6 ]
})

await Effect.runPromise(program)

flatten

Since v2.0.0

Flattens a stream of streams into a single stream.

Details

With the default sequential concurrency, inner streams are concatenated in strict order. When concurrency is greater than 1 or "unbounded", multiple inner streams may run at the same time and their outputs are merged as they arrive.

Signature

declare const flatten: <Arg extends Stream<Stream<any, any, any>, any, any> | {
  readonly bufferSize?: number;
  readonly concurrency?: Concurrency;
} | undefined = {
  readonly bufferSize?: number;
  readonly concurrency?: Concurrency;
}>(selfOrOptions?: Arg, options?: {
  readonly bufferSize?: number;
  readonly concurrency?: Concurrency;
}) => [Arg] extends [Stream<Stream<infer _A, infer _E, infer _R>, infer _E2, infer _R2>] ? Stream<_A, _E | _E2, _R | _R2> : <A, E, R, E2, R2>(self: Stream<Stream<A, E, R>, E2, R2>) => Stream<A, E | E2, R | R2>

Example Flattening nested streams

import { Effect, Stream } from "effect"

const streamOfStreams = Stream.make(
  Stream.make(1, 2),
  Stream.make(3, 4),
  Stream.make(5, 6)
)

const program = Effect.gen(function*() {
  const values = yield* Stream.runCollect(Stream.flatten(streamOfStreams))
  values // => [ 1, 2, 3, 4, 5, 6 ]
})

await Effect.runPromise(program)

flattenArray

Since v4.0.0

Flattens a stream of non-empty arrays into a stream of elements.

Signature

declare function flattenArray<A, E, R>(self: Stream<readonly [A, A], E, R>): Stream<A, E, R>

Example Flattening a stream of non-empty arrays into a stream of elements

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

const stream = Stream.make(Array.make(1, 2), Array.make(3))

const program = Effect.gen(function* () {
  const result = yield* Stream.runCollect(Stream.flattenArray(stream))
  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

flattenEffect

Since v2.0.0

Flattens a stream of Effect values into a stream of their results.

When to use

Use when stream elements already are effects and their successes should become stream elements while their failures enter the stream error channel.

Signature

declare const flattenEffect: <Arg extends Stream<Effect.Effect<any, any, any>, any, any> | {
  readonly concurrency?: Concurrency;
  readonly unordered?: boolean;
} | undefined = {
  readonly concurrency?: Concurrency;
  readonly unordered?: boolean;
}>(selfOrOptions?: Arg, options?: {
  readonly concurrency?: Concurrency;
  readonly unordered?: boolean;
}) => [Arg] extends [Stream<Effect.Effect<infer _A, infer _EX, infer _RX>, infer _E, infer _R>] ? Stream<_A, _EX | _E, _RX | _R> : <A, EX, RX, E, R>(self: Stream<Effect.Effect<A, EX, RX>, E, R>) => Stream<A, EX | E, RX | R>

Example Flattening a stream of Effect values into a stream of their results

import { Effect, Stream } from "effect"

const stream = Stream.make(Effect.succeed(1), Effect.succeed(2), Effect.succeed(3))

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream.pipe(Stream.flattenEffect()))
  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

flattenIterable

Since v4.0.0

Flattens the iterables emitted by this stream into the stream's structure.

Signature

declare function flattenIterable<A, E, R>(self: Stream<Iterable<A, any, any>, E, R>): Stream<A, E, R>

Example Flattening iterable values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream = Stream.make([1, 2], [3, 4]).pipe(Stream.flattenIterable)
  const values = yield* Stream.runCollect(stream)
  values // => [ 1, 2, 3, 4 ]
})

await Effect.runPromise(program)

flattenTake

Since v2.0.0

Unwraps Take values, emitting elements from non-empty arrays and ending or failing when the Exit signals completion.

Signature

declare function flattenTake<A, E, E2, R>(self: Stream<Take<A, E, void>, E2, R>): Stream<A, E | E2, R>

Example Flattening Take values

import { Array, Effect, Exit, Stream } from "effect"

const program = Effect.gen(function*() {
  const takes = Stream.make(
    Array.make(1, 2),
    Array.make(3),
    Exit.succeed<void>(undefined)
  )

  const values = yield* Stream.flattenTake(takes).pipe(Stream.runCollect)
  values // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

intersperse

Since v2.0.0

Inserts the provided element between emitted elements.

Signature

declare const intersperse: {
  <A2>(element: A2): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A, E, R>;
  <A, E, R, A2>(self: Stream<A, E, R>, element: A2): Stream<A | A2, E, R>;
}

Example Interspersing stream elements

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

const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2, 3, 4).pipe(Stream.intersperse(0))
  const result = yield* Stream.runCollect(stream)
  result // => [1, 0, 2, 0, 3, 0, 4]
})

await Effect.runPromise(program)

intersperseAffixes

Since v2.0.0

Adds a start value, middle value, and end value around stream elements.

Details

The start and end values are always emitted, even when the stream is empty.

Signature

declare const intersperseAffixes: {
  <A2, A3, A4>(options: {
    readonly end: A4;
    readonly middle: A3;
    readonly start: A2;
  }): <A, E, R>(self: Stream<A, E, R>) => Stream<A2 | A3 | A4 | A, E, R>;
  <A, E, R, A2, A3, A4>(self: Stream<A, E, R>, options: {
    readonly end: A4;
    readonly middle: A3;
    readonly start: A2;
  }): Stream<A | A2 | A3 | A4, E, R>;
}

Example Interspersing stream affixes

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

const stream = Stream.make("a", "b", "c").pipe(
  Stream.intersperseAffixes({ start: "[", middle: ",", end: "]" })
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => ["[", "a", ",", "b", ",", "c", "]"]
})

await Effect.runPromise(program)

onEnd

Since v3.6.0

Runs the provided effect when the stream ends successfully.

Signature

declare const onEnd: {
  <X, EX, RX>(onEnd: Effect<X, EX, RX>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, EX | E, RX | R>;
  <A, E, R, X, EX, RX>(self: Stream<A, E, R>, onEnd: Effect<X, EX, RX>): Stream<A, E | EX, R | RX>;
}

Example Running an effect on end

import { Effect, Stream } from "effect"

const events: Array<string> = []
const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.onEnd(Effect.sync(() => events.push("ended"))),
    Stream.runCollect
  )
  values // => [1, 2, 3]
})

await Effect.runPromise(program)
events // => ["ended"]

onFirst

Since v4.0.0

Runs the provided effect with the first element emitted by the stream.

Signature

declare const onFirst: {
  <A, X, EX, RX>(onFirst: (element: NoInfer<A>) => Effect<X, EX, RX>): <E, R>(self: Stream<A, E, R>) => Stream<A, EX | E, RX | R>;
  <A, E, R, X, EX, RX>(self: Stream<A, E, R>, onFirst: (element: NoInfer<A>) => Effect<X, EX, RX>): Stream<A, E | EX, R | RX>;
}

Example Running an effect on the first value

import { Effect, Stream } from "effect"

const first: Array<number> = []
await Effect.runPromise(Effect.gen(function* () {
  yield* Stream.fromArray([1, 2, 3]).pipe(
    Stream.onFirst((value) => Effect.sync(() => first.push(value))),
    Stream.runDrain
  )
}))
first // => [1]

onStart

Since v3.6.0

Runs the provided effect before this stream starts.

Signature

declare const onStart: {
  <X, EX, RX>(onStart: Effect<X, EX, RX>): <A, E, R>(self: Stream<A, E, R>) => Stream<A, EX | E, RX | R>;
  <A, E, R, X, EX, RX>(self: Stream<A, E, R>, onStart: Effect<X, EX, RX>): Stream<A, E | EX, R | RX>;
}

Example Running an effect on start

import { Effect, Stream } from "effect"

const events: Array<string> = []
const program = Effect.gen(function*() {
  const stream = Stream.fromArray([1, 2, 3]).pipe(
    Stream.onStart(Effect.sync(() => events.push("started")))
  )

  const values = yield* Stream.runCollect(stream)
  values // => [1, 2, 3]
})

await Effect.runPromise(program)
events // => ["started"]

pipeThrough

Since v2.0.0

Pipes the stream through Sink.toChannel, emitting only the sink leftovers.

Details

If the sink completes mid-chunk, the remaining elements become the output stream.

Signature

declare const pipeThrough: {
  <A2, A, L, E2, R2>(sink: Sink<A2, A, L, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<L, E2 | E, R2 | R>;
  <A, E, R, A2, L, E2, R2>(self: Stream<A, E, R>, sink: Sink<A2, A, L, E2, R2>): Stream<L, E | E2, R | R2>;
}

Example Piping through a sink

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

const program = Effect.gen(function*() {
  const leftovers = yield* Stream.make(1, 2, 3, 4).pipe(
    Stream.pipeThrough(Sink.take(2)),
    Stream.runCollect
  )

  leftovers // => [ 3, 4 ]
})

await Effect.runPromise(program)

pipeThroughChannel

Since v2.0.0

Pipes this stream through a channel that consumes and emits chunked elements.

Details

The channel receives NonEmptyReadonlyArray chunks and can transform both the output elements and error type.

Signature

declare const pipeThroughChannel: {
  <R2, E, E2, A, A2>(channel: Channel<readonly [A2, A2], E2, unknown, readonly [A, A], E, unknown, R2>): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R2 | R>;
  <R, R2, E, E2, A, A2>(self: Stream<A, E, R>, channel: Channel<readonly [A2, A2], E2, unknown, readonly [A, A], E, unknown, R2>): Stream<A2, E2, R | R2>;
}

Example Piping through a channel

import { Array, Channel, Effect, Stream } from "effect"

type NumberChunk = readonly [number, ...Array<number>]

const doubleChunks = Channel.identity<NumberChunk, never, unknown>().pipe(
  Channel.map((chunk) => Array.map(chunk, (n) => n * 2))
)

const program = Effect.gen(function*() {
  const result = yield* Stream.fromArray([1, 2, 3]).pipe(
    Stream.rechunk(2),
    Stream.pipeThroughChannel(doubleChunks),
    Stream.runCollect
  )
  result // => [ 2, 4, 6 ]
})

await Effect.runPromise(program)

pipeThroughChannelOrFail

Since v2.0.0

Pipes values through the provided channel while preserving this stream's failures alongside any channel failures.

Details

Upstream failures are not passed to the channel, so the resulting stream can fail with either the original stream error or the channel error.

Signature

declare const pipeThroughChannelOrFail: {
  <R2, E, E2, A, A2>(channel: Channel<readonly [A2, A2], E2, unknown, readonly [A, A], E, unknown, R2>): <R>(self: Stream<A, E, R>) => Stream<A2, E | E2, R2 | R>;
  <R, R2, E, E2, A, A2>(self: Stream<A, E, R>, channel: Channel<readonly [A2, A2], E2, unknown, readonly [A, A], E, unknown, R2>): Stream<A2, E | E2, R | R2>;
}

Example Piping through a channel with failures

import { Array, Channel, Effect, Stream } from "effect"

type NumberChunk = readonly [number, ...Array<number>]

const stringifyChunks = Channel.identity<NumberChunk, "StreamError", unknown>().pipe(
  Channel.map((chunk) => Array.map(chunk, String))
)

await Effect.runPromise(Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3).pipe(
    Stream.rechunk(2),
    Stream.pipeThroughChannelOrFail(stringifyChunks),
    Stream.runCollect
  )

  result // => ["1", "2", "3"]
}))

prepend

Since v2.0.0

Prepends the values from the provided iterable before the stream's elements.

Signature

declare const prepend: {
  <B>(values: Iterable<B>): <A, E, R>(self: Stream<A, E, R>) => Stream<B | A, E, R>;
  <A, E, R, B>(self: Stream<A, E, R>, values: Iterable<B>): Stream<A | B, E, R>;
}

Example Prepending values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const values = yield* Stream.make(3, 4).pipe(
    Stream.prepend([1, 2]),
    Stream.runCollect
  )

  values // => [ 1, 2, 3, 4 ]
})

await Effect.runPromise(program)

switchMap

Since v4.0.0

Switches to the latest stream produced by the mapping function, interrupting the previous stream when a new element arrives.

Signature

declare const switchMap: {
  <A, A2, E2, R2>(f: (a: A) => Stream<A2, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A2, E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, f: (a: A) => Stream<A2, E2, R2>, options?: {
    readonly bufferSize?: number;
    readonly concurrency?: Concurrency;
  }): Stream<A2, E | E2, R | R2>;
}

Example Switching to the latest stream

import { Effect, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.switchMap((n) => (n === 3 ? Stream.make(n) : Stream.never)),
  Stream.runCollect
)

await Effect.runPromise(Effect.gen(function*() {
  const result = yield* program
  result // => [ 3 ]
}))

tap

Since v2.0.0

Runs the provided effect for each element while preserving the elements.

Signature

declare const tap: {
  <A, X, E2, R2>(f: (a: NoInfer<A>) => Effect<X, E2, R2>, options?: {
    readonly concurrency?: Concurrency;
  }): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, X, E2, R2>(self: Stream<A, E, R>, f: (a: NoInfer<A>) => Effect<X, E2, R2>, options?: {
    readonly concurrency?: Concurrency;
  }): Stream<A, E | E2, R | R2>;
}

Example Tapping stream values

import { Effect, Stream } from "effect"

const events: Array<string> = []
const program = Effect.gen(function*() {
  const result = yield* Stream.fromArray([1, 2, 3]).pipe(
    Stream.tap((n) => Effect.sync(() => events.push(`before mapping: ${n}`))),
    Stream.map((n) => n * 2),
    Stream.tap((n) => Effect.sync(() => events.push(`after mapping: ${n}`))),
    Stream.runCollect
  )

  result // => [2, 4, 6]
})

await Effect.runPromise(program)
events // => ["before mapping: 1", "after mapping: 2", "before mapping: 2", "after mapping: 4", "before mapping: 3", "after mapping: 6"]

tapBoth

Since v2.0.0

Returns a stream that effectfully "peeks" at elements and failures.

Signature

declare const tapBoth: {
  <A, E, X, E2, R2, Y, E3, R3>(options: {
    readonly concurrency?: Concurrency;
    readonly onElement: (a: NoInfer<A>) => Effect.Effect<X, E2, R2>;
    readonly onError: (a: NoInfer<E>) => Effect.Effect<Y, E3, R3>;
  }): <R>(self: Stream<A, E, R>) => Stream<A, E | E2 | E3, R2 | R3 | R>;
  <A, E, R, X, E2, R2, Y, E3, R3>(self: Stream<A, E, R>, options: {
    readonly concurrency?: Concurrency;
    readonly onElement: (a: NoInfer<A>) => Effect.Effect<X, E2, R2>;
    readonly onError: (a: NoInfer<E>) => Effect.Effect<Y, E3, R3>;
  }): Stream<A, E | E2 | E3, R | R2 | R3>;
}

Example Tapping values and errors

import { Effect, Stream } from "effect"

const events: Array<string> = []
const program = Effect.gen(function*() {
  const stream = Stream.make(1, 2).pipe(
    Stream.concat(Stream.fail("boom")),
    Stream.tapBoth({
      onElement: (value) => Effect.sync(() => events.push(`seen: ${value}`)),
      onError: (error) => Effect.sync(() => events.push(`error: ${error}`))
    }),
    Stream.catch(() => Stream.make(3))
  )
  const result = yield* Stream.runCollect(stream)
  result // => [1, 2, 3]
})

await Effect.runPromise(program)
events // => ["seen: 1", "seen: 2", "error: boom"]

tapSink

Since v2.0.0

Runs a sink for all stream elements while still emitting them downstream.

Signature

declare const tapSink: {
  <A, E2, R2>(sink: Sink<unknown, A, unknown, E2, R2>): <E, R>(self: Stream<A, E, R>) => Stream<A, E2 | E, R2 | R>;
  <A, E, R, E2, R2>(self: Stream<A, E, R>, sink: Sink<unknown, A, unknown, E2, R2>): Stream<A, E | E2, R | R2>;
}

Example Tapping values with a sink

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

const program = Effect.gen(function*() {
  const seen = yield* Ref.make<Array<number>>([])
  const sink = Sink.forEach((value: number) =>
    Ref.update(seen, (items) => [...items, value])
  )
  const result = yield* Stream.make(1, 2, 3).pipe(
    Stream.tapSink(sink),
    Stream.runCollect
  )
  const tapped = yield* Ref.get(seen)
  tapped // => [ 1, 2, 3 ]
  result // => [ 1, 2, 3 ]
})

await Effect.runPromise(program)

Text

decodeText

Since v2.0.0

Decodes Uint8Array chunks into strings using TextDecoder with an optional encoding.

Signature

declare const decodeText: <Arg extends Stream<Uint8Array, any, any> | {
  readonly encoding?: string;
} | undefined = {
  readonly encoding?: string;
}>(streamOrOptions?: Arg, options?: {
  readonly encoding?: string;
}) => [Arg] extends [Stream<Uint8Array, infer _E, infer _R>] ? Stream<string, _E, _R> : <E, R>(self: Stream<Uint8Array, E, R>) => Stream<string, E, R>

Example Decoding Uint8Array chunks into strings using TextDecoder with an optional encoding

import { Effect, Stream } from "effect"

const encoder = new TextEncoder()
const stream = Stream.make(
  encoder.encode("Hello"),
  encoder.encode(" World")
)

const program = Effect.gen(function*() {
  const decoded = yield* stream.pipe(
    Stream.decodeText,
    Stream.runCollect
  )
  decoded // => [ 'Hello', ' World' ]
})

await Effect.runPromise(program)

encodeText

Since v2.0.0

Encodes a stream of strings into UTF-8 Uint8Array chunks.

Signature

declare function encodeText<E, R>(self: Stream<string, E, R>): Stream<Uint8Array<ArrayBufferLike>, E, R>

Example Encoding a stream of strings into UTF-8 Uint8Array chunks

import { Effect, Stream } from "effect"

const stream = Stream.make("Hello", " ", "World")
const program = Effect.gen(function*() {
  const encoded = Stream.encodeText(stream)
  const chunks = yield* Stream.runCollect(encoded)
  const bytes = chunks.map((chunk) => [...chunk])
  bytes // => [ [ 72, 101, 108, 108, 111 ], [ 32 ], [ 87, 111, 114, 108, 100 ] ]
})

await Effect.runPromise(program)

splitLines

Since v2.0.0

Splits a stream of strings into lines, handling \n, \r, and \r\n delimiters across chunks.

Signature

declare function splitLines<E, R>(self: Stream<string, E, R>): Stream<string, E, R>

Example Splitting streamed text into lines

import { Effect, Stream } from "effect"

await Effect.runPromise(Effect.gen(function* () {
  const lines = yield* Stream.runCollect(
    Stream.make("a\nb\r\n", "c\n").pipe(Stream.splitLines)
  )
  lines // => [ 'a', 'b', 'c' ]
}))

Tracing

withSpan

Since v2.0.0

Wraps the stream with a new span for tracing.

Signature

declare const withSpan: {
  (name: string, options?: SpanOptions): <A, E, R>(self: Stream<A, E, R>) => Stream<A, E, Exclude<R, ParentSpan>>;
  <A, E, R>(self: Stream<A, E, R>, name: string, options?: SpanOptions): Stream<A, E, Exclude<R, ParentSpan>>;
}

Example Wrapping a stream in a span

import { Effect, Stream } from "effect"

const stream = Stream.fromArray([1, 2, 3]).pipe(Stream.withSpan("numbers"))

await Effect.runPromise(
  Effect.gen(function*() {
    const values = yield* Stream.runCollect(stream)
    values // => [ 1, 2, 3 ]
  })
)

Type IDs

TypeId

Since v4.0.0

Runtime identifier stored on Stream values and used by isStream to recognize them.

Details

This marker is part of the runtime representation of Stream values. Prefer isStream when narrowing unknown values.

See

isStream for the public guard that checks this identifier

Signature

declare const TypeId: "~effect/Stream"

TypeId

type

Since v4.0.0

String literal type used as the unique brand for Stream values.

Signature

type TypeId = "~effect/Stream"

Utility Types

Error

type

Since v3.4.0

Extract the error type from a Stream type.

Signature

type Error<T> = T extends Stream<infer _A, infer _E, infer _R> ? _E : never

Example Extracting the error type from a Stream type

import { Stream } from "effect"

type NumberStream = Stream.Stream<number, string, never>
type ErrorType = Stream.Error<NumberStream>
const error: ErrorType = "boom"

Services

type

Since v4.0.0

Extract the services type from a Stream type.

Signature

type Services<T> = T extends Stream<infer _A, infer _E, infer _R> ? _R : never

Example Extracting the services type from a Stream type

import { Stream } from "effect"

interface Database {
  query: (sql: string) => unknown
}
type NumberStream = Stream.Stream<number, string, { db: Database }>
type RequiredServices = Stream.Services<NumberStream>
const services: RequiredServices = { db: { query: (sql) => sql } }
services.db.query("SELECT 1") // => "SELECT 1"

StreamTypeLambda

interface

Since v2.0.0

Type lambda for Stream used in higher-kinded type operations.

Signature

interface StreamTypeLambda extends TypeLambda {
  readonly type: Stream<unknown, unknown, unknown>;
}

Example Using the stream type lambda

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

// Create a Stream type using the type lambda
type NumberStream = HKT.Kind<Stream.StreamTypeLambda, never, never, string, number>
// Equivalent to: Stream<number, string, never>
const stream: NumberStream = Stream.make(1, 2, 3)
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]

Success

type

Since v3.4.0

Extract the success type from a Stream type.

Signature

type Success<T> = T extends Stream<infer _A, infer _E, infer _R> ? _A : never

Example Extracting the success type from a Stream type

import { Stream } from "effect"

type NumberStream = Stream.Stream<number, string, never>
type SuccessType = Stream.Success<NumberStream>
const value: SuccessType = 42

Zipping

cross

Since v2.0.0

Creates the cartesian product of two streams, running the right stream for each element in the left stream.

Details

See also Stream.zip for the more common point-wise variant.

Signature

declare const cross: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<[AL, AR], ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<[AL, AR], EL | ER, RL | RR>;
}

Example Computing cartesian products

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const left = Stream.make(1, 2)
  const right = Stream.make("a", "b")
  const values = yield* Stream.runCollect(Stream.cross(left, right))
  values // => [ [ 1, 'a' ], [ 1, 'b' ], [ 2, 'a' ], [ 2, 'b' ] ]
})

await Effect.runPromise(program)

crossWith

Since v2.0.0

Creates a cartesian product of elements from two streams using a function.

Details

The right stream is rerun for every element in the left stream.

See also Stream.zipWith for the more common point-wise variant.

Signature

declare const crossWith: {
  <AR, ER, RR, AL, A>(right: Stream<AR, ER, RR>, f: (left: AL, right: AR) => A): <EL, RL>(left: Stream<AL, EL, RL>) => Stream<A, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR, A>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>, f: (left: AL, right: AR) => A): Stream<A, EL | ER, RL | RR>;
}

Example Combining cartesian products

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const left = Stream.make(1, 2)
  const right = Stream.make("a", "b")
  const combined = Stream.crossWith(left, right, (n, s) => `${n}-${s}`)
  const result = yield* Stream.runCollect(combined)
  result // => [ '1-a', '1-b', '2-a', '2-b' ]
})

await Effect.runPromise(program)

zip

Since v2.0.0

Zips this stream with another point-wise and emits tuples of elements from both streams. The new stream ends when either stream ends.

Signature

declare const zip: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>): <A, E, R>(self: Stream<A, E, R>) => Stream<[A, A2], E2 | E, R2 | R>;
  <A, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<[A, A2], E | E2, R | R2>;
}

Example Zipping streams

import { Effect, Stream } from "effect"

const stream1 = Stream.make(1, 2, 3)
const stream2 = Stream.make("a", "b", "c")

const zipped = Stream.zip(stream1, stream2)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(zipped)
  result // => [ [ 1, 'a' ], [ 2, 'b' ], [ 3, 'c' ] ]
})

await Effect.runPromise(program)

zipFlatten

Since v2.0.0

Zips this stream with another point-wise and emits tuples of elements from both streams, flattening the left tuple.

Details

The new stream will end when one of the sides ends.

Signature

declare const zipFlatten: {
  <A2, E2, R2>(that: Stream<A2, E2, R2>): <A extends readonly Array<any>, E, R>(self: Stream<A, E, R>) => Stream<[...Array<A>, A2], E2 | E, R2 | R>;
  <A extends readonly Array<any>, E, R, A2, E2, R2>(self: Stream<A, E, R>, that: Stream<A2, E2, R2>): Stream<[...Array<A>, A2], E | E2, R | R2>;
}

Example Zipping and flattening tuples

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const stream1 = Stream.make(
    [1, "a"] as const,
    [2, "b"] as const,
    [3, "c"] as const
  )
  const stream2 = Stream.make("x", "y", "z")
  const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

  result // => [ [ 1, 'a', 'x' ], [ 2, 'b', 'y' ], [ 3, 'c', 'z' ] ]
})

await Effect.runPromise(program)

zipLatest

Since v2.0.0

Combines two streams by emitting each new element with the latest value from the other stream.

When to use

Use when two streams should start emitting combined pairs after both have produced at least one value.

Gotchas

Note: tracking the latest value is done on a per-array basis. That means that emitted elements that are not the last value in arrays will never be used for zipping.

Signature

declare const zipLatest: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<[AL, AR], ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<[AL, AR], EL | ER, RL | RR>;
}

Example Zipping latest values

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const result = yield* Stream.zipLatest(
    Stream.make(1),
    Stream.make("a")
  ).pipe(Stream.runCollect)

  result // => [ [ 1, 'a' ] ]
})
await Effect.runPromise(program)

zipLatestAll

Since v3.3.0

Zips multiple streams so that when a value is emitted by any stream, it is combined with the latest values from the other streams to produce a result.

When to use

Use when each stream should contribute its latest value after all streams have emitted at least once.

Gotchas

Note: tracking the latest value is done on a per-array basis. That means that emitted elements that are not the last value in arrays will never be used for zipping.

Signature

declare function zipLatestAll<T extends readonly Array<Stream<any, any, any>>>(...streams: T): Stream<[T[number]] extends [never] ? never : { [K in string | number | symbol]: T[K] extends Stream<A, _E, _R> ? A : never }, [T[number]] extends [never] ? never : T[number] extends Stream<_A, _E, _R> ? _E : never, [T[number]] extends [never] ? never : T[number] extends Stream<_A, _E, _R> ? _R : never>

Example Zipping latest values from many streams

import { Effect, Stream } from "effect"

const stream = Stream.zipLatestAll(
  Stream.make(1, 2, 3).pipe(Stream.rechunk(1)),
  Stream.make("a", "b", "c").pipe(Stream.rechunk(1)),
  Stream.make(true, false, true).pipe(Stream.rechunk(1))
)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [[1, "a", true], [2, "a", true], [2, "b", true], [2, "b", false], [3, "b", false], [3, "c", false], [3, "c", true]]
})

await Effect.runPromise(program)

zipLatestWith

Since v2.0.0

Combines the latest values from both streams whenever either emits, using the provided function.

When to use

Use when two streams should start emitting custom combined values after both have produced at least one value.

Gotchas

Note: tracking the latest value is done on a per-array basis. That means that emitted elements that are not the last value in arrays will never be used for zipping.

Signature

declare const zipLatestWith: {
  <AR, ER, RR, AL, A>(right: Stream<AR, ER, RR>, f: (left: AL, right: AR) => A): <EL, RL>(left: Stream<AL, EL, RL>) => Stream<A, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR, A>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>, f: (left: AL, right: AR) => A): Stream<A, EL | ER, RL | RR>;
}

Example Zipping latest values with a function

import { Effect, Stream } from "effect"

await Effect.runPromise(Effect.gen(function*() {
  const result = yield* Stream.make(1, 2, 3).pipe(
    Stream.rechunk(1),
    Stream.zipLatestWith(
      Stream.make(10, 20).pipe(Stream.rechunk(1)),
      (n, m) => n + m
    ),
    Stream.runCollect
  )

  result // => [ 11, 12, 22, 23 ]
}))

zipLeft

Since v2.0.0

Zips this stream with another point-wise and keeps only the values from the left stream.

Details

The resulting stream ends when either side ends.

Signature

declare const zipLeft: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<AL, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<AL, EL | ER, RL | RR>;
}

Example Zipping streams while keeping left values

import { Effect, Stream } from "effect"

const stream1 = Stream.make(1, 2, 3, 4)
const stream2 = Stream.make("a", "b")

const program = Effect.gen(function*() {
  const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)
  result // => [ 1, 2 ]
})

await Effect.runPromise(program)

zipRight

Since v2.0.0

Zips this stream with another point-wise, keeping only right values and ending when either stream ends.

Signature

declare const zipRight: {
  <AR, ER, RR>(right: Stream<AR, ER, RR>): <AL, EL, RL>(left: Stream<AL, EL, RL>) => Stream<AR, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>): Stream<AR, EL | ER, RL | RR>;
}

Example Zipping streams while keeping right values

import { Effect, Stream } from "effect"

const stream1 = Stream.make(1, 2)
const stream2 = Stream.make("a", "b", "c", "d")

const program = Effect.gen(function*() {
  const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)
  result // => [ 'a', 'b' ]
})

await Effect.runPromise(program)

zipWith

Since v2.0.0

Zips two streams point-wise with a combining function, ending when either stream ends.

Signature

declare const zipWith: {
  <AR, ER, RR, AL, A>(right: Stream<AR, ER, RR>, f: (left: AL, right: AR) => A): <EL, RL>(left: Stream<AL, EL, RL>) => Stream<A, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR, A>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>, f: (left: AL, right: AR) => A): Stream<A, EL | ER, RL | RR>;
}

Example Zipping streams with a function

import { Effect, Stream } from "effect"

const stream1 = Stream.make(1, 2, 3, 4, 5, 6)
const stream2 = Stream.make("a", "b", "c")

const zipped = Stream.zipWith(stream1, stream2, (n, s) => `${n}-${s}`)

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(zipped)
  result // => [ '1-a', '2-b', '3-c' ]
})

await Effect.runPromise(program)

zipWithArray

Since v4.0.0

Zips two streams by applying a function to non-empty arrays of elements.

Details

The function returns output plus leftover arrays that carry into the next pull.

Signature

declare const zipWithArray: {
  <AR, ER, RR, AL, A>(right: Stream<AR, ER, RR>, f: (left: readonly [AL, AL], right: readonly [AR, AR]) => readonly [readonly [A, A], readonly Array<AL>, readonly Array<AR>]): <EL, RL>(left: Stream<AL, EL, RL>) => Stream<A, ER | EL, RR | RL>;
  <AL, EL, RL, AR, ER, RR, A>(left: Stream<AL, EL, RL>, right: Stream<AR, ER, RR>, f: (left: readonly [AL, AL], right: readonly [AR, AR]) => readonly [readonly [A, A], readonly Array<AL>, readonly Array<AR>]): Stream<A, EL | ER, RL | RR>;
}

Example Zipping stream chunks

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

const left = Stream.fromArrays([1, 2, 3], [4, 5])
const right = Stream.fromArrays(["a", "b"], ["c", "d", "e"])

const zipped = Stream.zipWithArray(left, right, (leftChunk, rightChunk) => {
  const minLength = Math.min(leftChunk.length, rightChunk.length)
  const output = Array.makeBy(minLength, (i) => [leftChunk[i], rightChunk[i]] as const)

  return [output, leftChunk.slice(minLength), rightChunk.slice(minLength)]
})

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(zipped)
  result // => [ [ 1, 'a' ], [ 2, 'b' ], [ 3, 'c' ], [ 4, 'd' ], [ 5, 'e' ] ]
})

await Effect.runPromise(program)

zipWithIndex

Since v2.0.0

Zips this stream together with the index of elements.

Signature

declare function zipWithIndex<A, E, R>(self: Stream<A, E, R>): Stream<[A, number], E, R>

Example Zipping elements with indices

import { Effect, Stream } from "effect"

const program = Effect.gen(function*() {
  const indexed = yield* Stream.make("a", "b", "c", "d").pipe(
    Stream.zipWithIndex,
    Stream.runCollect
  )
  indexed // => [ [ 'a', 0 ], [ 'b', 1 ], [ 'c', 2 ], [ 'd', 3 ] ]
})

await Effect.runPromise(program)

zipWithNext

Since v2.0.0

Zips each element with the next element, pairing the final element with Option.none().

Signature

declare function zipWithNext<A, E, R>(self: Stream<A, E, R>): Stream<[A, Option<A>], E, R>

Example Zipping elements with next values

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

const stream = Stream.zipWithNext(Stream.make(1, 2, 3, 4))

await Effect.runPromise(Effect.gen(function*() {
  const values = yield* Stream.runCollect(stream)
  values // => [[1, Option.some(2)], [2, Option.some(3)], [3, Option.some(4)], [4, Option.none()]]
}))

zipWithPrevious

Since v2.0.0

Zips each element with its previous element, starting with None.

Signature

declare function zipWithPrevious<A, E, R>(self: Stream<A, E, R>): Stream<[Option<A>, A], E, R>

Example Zipping elements with previous values

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

const stream = Stream.zipWithPrevious(Stream.make(1, 2, 3, 4))

const program = Effect.gen(function*() {
  const result = yield* Stream.runCollect(stream)
  result // => [[Option.none(), 1], [Option.some(1), 2], [Option.some(2), 3], [Option.some(3), 4]]
})

await Effect.runPromise(program)

zipWithPreviousAndNext

Since v2.0.0

Zips each element with its previous and next values.

Signature

declare function zipWithPreviousAndNext<A, E, R>(self: Stream<A, E, R>): Stream<[Option<A>, A, Option<A>], E, R>

Example Zipping elements with neighbors

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

const program = Effect.gen(function*() {
  const values = yield* Stream.make(1, 2, 3).pipe(
    Stream.zipWithPreviousAndNext,
    Stream.runCollect
  )
  values // => [[Option.none(), 1, Option.some(2)], [Option.some(1), 2, Option.some(3)], [Option.some(2), 3, Option.none()]]
})

await Effect.runPromise(program)