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.
Accessors
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
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
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)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)Accumulates state across the stream, emitting the initial state and each updated state.
Signature
declare const scan: { <S, A>(initial: 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: 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
Accumulates state effectfully and emits the initial state plus each accumulated state.
Signature
declare const scanEffect: { <S, A, E2, R2>(initial: 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: 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
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
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.
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 ]}))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
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
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.
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)Buffering
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
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
The default chunk size used by Stream constructors and combinators.
Signature
declare const DefaultChunkSize: numberExample
(Reading the default chunk size)
import { Stream } from "effect"
Stream.DefaultChunkSize // => 4096Constructors
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]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))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)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)) // => []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!")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
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")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!")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
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
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
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
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
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
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 = falseawait Effect.runPromise( Stream.fromEffectDrain(Effect.sync(() => { drained = true })).pipe(Stream.runDrain))drained // => truefromEffectRepeat
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 = 0const stream = Stream.fromEffectRepeat(Effect.sync(() => ++n)).pipe(Stream.take(5))await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, 4, 5]fromEffectSchedule
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
Creates a stream from an event listener.
Signature
declare function fromEventListener<A = unknown>(target: EventListener<A>, 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
Creates a new Stream from an iterable collection of values.
Details
chunkSize: Maximum number of values emitted per chunk.
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
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
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
Creates a stream that consumes values from an iterator.
Details
The maxChunkSize parameter controls how many values are pulled per chunk.
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
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
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)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]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
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
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
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)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)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]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) // => []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]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.
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)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"]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]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]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]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]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]transformPull
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
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]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)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)Decoding
decodeText
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)Deduplication
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
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
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
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
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
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]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
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)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)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) // => 6runCollect
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)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)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]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
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
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
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
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]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
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, never, 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
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)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>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
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 Array.fromAsync(Stream.toAsyncIterable(stream)) // => [1, 2, 3]toAsyncIterableEffect
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(() => Array.fromAsync(iterable))})
await Effect.runPromise(program) // => [1, 2, 3]toAsyncIterableWith
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 Array.fromAsync(iterable) // => [1, 2, 3]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
Converts a stream to a PubSub of Take values for concurrent consumption.
Details
Take values include the stream's end and failure signals.
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))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)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
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 Array.fromAsync(readableStream)values // => [ 1, 2, 3 ]toReadableStreamEffect
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
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.
Signature
declare const toReadableStreamWith: <A, XR>(context: Context<XR>, options?: { 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 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 Array.fromAsync(readableStream)values // => [ 1, 2, 3, 4, 5 ]Encoding
encodeText
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)Error Handling
catchCause
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
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
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)catchFilter
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
_tagbased recovery from one tagged error - catchTags for
_tagbased 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>;}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
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
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)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)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> }, 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> }, 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)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
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]}))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)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"]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
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
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)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)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)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]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"]withExecutionPlan
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
Drops the first n elements from this stream.
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)Drops the last specified number of elements from this stream.
Details
Keeps the last n elements in memory to drop them on completion.
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)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
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)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
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
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>;}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
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)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
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
Filtervariant - 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
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: SizeInput, 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: SizeInput, onLimitReached: LazyArg<Stream<Uint8Array<ArrayBufferLike>, E, R>>): Stream<Uint8Array<ArrayBufferLike>, E, R>;}Example
(Truncating at a byte limit)
import { Effect, Stream } from "effect"
const program = Stream.make( new Uint8Array([1, 2]), new Uint8Array([3, 4, 5])).pipe( Stream.limitBytes(4, () => Stream.empty), Stream.runCollect, Effect.map((chunks) => chunks.map((chunk) => [...chunk])))
await Effect.runPromise(program) // => [[1, 2]]Splits a stream into scoped excluded and satisfying 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
bufferSize elements ahead of the slower one.
Signature
declare const partition: { <A, Pass, Fail>(filter: Filter<NoInfer<A>, Pass, Fail>, options?: { readonly bufferSize?: number; }): <E, R>(self: Stream<A, E, R>) => Effect<[excluded: Stream<Fail, E, never>, satisfying: Stream<Pass, E, never>], never, Scope | R>; <A, E, R, Pass, Fail>(self: Stream<A, E, R>, filter: Filter<NoInfer<A>, Pass, Fail>, options?: { readonly bufferSize?: number; }): Effect<[excluded: Stream<Fail, E, never>, satisfying: Stream<Pass, E, never>], never, Scope | R>;}Example
(Partitioning a stream)
import { Effect, Result, Stream } from "effect"
const program = Effect.gen(function*() { const [excluded, satisfying] = yield* Stream.partition( Stream.make(1, 2, 3, 4), (n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n) ) const left = yield* Stream.runCollect(excluded) const right = yield* Stream.runCollect(satisfying) left // => [ 1, 3 ] right // => [ 2, 4 ]})await Effect.runPromise(Effect.scoped(program))partitionEffect
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
Filtervariant, which returns the failing stream before the passing stream - 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?: number | "unbounded"; }): <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?: number | "unbounded"; }): Effect<[passes: Stream<Pass, E | EX, never>, fails: Stream<Fail, E | EX, never>], never, Scope | R | RX>;}partitionQueue
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))Takes the first n elements from this stream, returning Stream.empty when n < 1.
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)Keeps the last n elements from this 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)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
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)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
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
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>;}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
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
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>;}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
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)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.
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
Partitions the stream into arrays, emitting when the chunk size is reached or the duration passes.
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)Groups the stream into arrays of the specified size, preserving element order.
Details
The size is clamped to at least 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)Emits a sliding window of n elements.
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
Emits sliding windows of chunkSize elements, advancing by stepSize.
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)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
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)) // => trueStream.isStream({ data: [1, 2, 3] }) // => falseInterruption
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
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
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 }]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?: number | "unbounded"; }): <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?: number | "unbounded"; }): 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)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?: number | "unbounded";} | undefined = { readonly bufferSize?: number; readonly concurrency?: number | "unbounded";}>(selfOrOptions?: Arg, options?: { readonly bufferSize?: number; readonly concurrency?: number | "unbounded";}) => [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)flattenEffect
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?: number | "unbounded"; readonly unordered?: boolean;} | undefined = { readonly concurrency?: number | "unbounded"; readonly unordered?: boolean;}>(selfOrOptions?: Arg, options?: { readonly concurrency?: number | "unbounded"; 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
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)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]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
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) => Array<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
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
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)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
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)Maps both the failure and success channels of a stream.
Signature
declare const mapBoth: { <E, E2, A, A2>(options: { readonly onFailure: (e: E) => E2; readonly onSuccess: (a: A) => A2; }): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>; <A, E, R, E2, A2>(self: Stream<A, E, R>, options: { readonly onFailure: (e: E) => E2; readonly onSuccess: (a: A) => A2; }): Stream<A2, E2, R>;}Example
(Mapping both the failure and success channels of a stream)
import { Effect, Stream } from "effect"
const mapper = { onFailure: (error: string) => `error: ${error}`, onSuccess: (value: number) => value * 2}
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)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?: number | "unbounded"; 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?: number | "unbounded"; 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
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)interleave
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
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)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)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
Signature
declare const mergeAll: { (options: { readonly bufferSize?: number; readonly concurrency: number | "unbounded"; }): <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: number | "unbounded"; }): 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
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"]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
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
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
Interface representing an event listener target.
Signature
interface EventListener<A> { 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
Describes how merged streams decide when to halt.
Signature
type HaltStrategy = Channel.HaltStrategyA 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
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
Type-level marker that excludes Stream from unification.
Signature
interface StreamUnifyIgnore { Effect?: true;}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
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
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>;}Signature
declare const let: { <N extends string, A extends object, B>(name: Exclude<N, keyof A>, f: (a: NoInfer<A>) => B): <E, R>(self: Stream<A, E, R>) => Stream<{ [K in string | number | symbol]: K extends keyof A ? A[K] : B }, E, R>; <A extends object, E, R, N extends string, B>(self: Stream<A, E, R>, name: Exclude<N, keyof A>, f: (a: NoInfer<A>) => B): Stream<{ [K in string | number | symbol]: K extends keyof A ? A[K] : B }, E, R>;}Providing Services
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
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
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
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
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
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
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)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
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)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)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
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.
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 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 ]}))Resource Management
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"]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"]Sequencing
Binds the result of a stream to a field in the do-notation record.
Signature
declare const bind: { <N extends string, A, B, E2, R2>(tag: Exclude<N, keyof A>, f: (_: NoInfer<A>) => Stream<B, E2, R2>, options?: { readonly bufferSize?: number; readonly concurrency?: number | "unbounded"; }): <E, R>(self: Stream<A, E, R>) => Stream<{ [K in string | number | symbol]: K extends keyof A ? A[K] : B }, E2 | E, R2 | R>; <A, E, R, N extends string, B, E2, R2>(self: Stream<A, E, R>, tag: Exclude<N, keyof A>, f: (_: NoInfer<A>) => Stream<B, E2, R2>, options?: { readonly bufferSize?: number; readonly concurrency?: number | "unbounded"; }): Stream<{ [K in string | number | symbol]: K extends keyof A ? A[K] : 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
Binds an Effect-produced value into the do-notation record for each stream element.
Signature
declare const bindEffect: { <N extends string, A, B, E2, R2>(tag: Exclude<N, keyof A>, f: (_: NoInfer<A>) => Effect<B, E2, R2>, options?: { readonly bufferSize?: number; readonly concurrency?: number | "unbounded"; readonly unordered?: boolean; }): <E, R>(self: Stream<A, E, R>) => Stream<{ [K in string | number | symbol]: K extends keyof A ? A[K] : B }, E2 | E, R2 | R>; <A, E, R, N extends string, B, E2, R2>(self: Stream<A, E, R>, tag: Exclude<N, keyof A>, f: (_: NoInfer<A>) => Effect<B, E2, R2>, options?: { readonly bufferSize?: number; readonly concurrency?: number | "unbounded"; readonly unordered?: boolean; }): Stream<{ [K in string | number | symbol]: K extends keyof A ? A[K] : 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)combineArray
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)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 ]}))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)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"]flattenArray
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)flattenTake
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)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)intersperse
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
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)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"]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]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
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
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
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"]}))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)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
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)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?: number | "unbounded"; }): <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?: number | "unbounded"; }): 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 ]}))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?: number | "unbounded"; }): <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?: number | "unbounded"; }): 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"]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?: number | "unbounded"; 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?: number | "unbounded"; 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"]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)Splitting
splitLines
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
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
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"String literal type used as the unique brand for Stream values.
Signature
type TypeId = "~effect/Stream"Utility Types
Extract the error type from a Stream type.
Signature
type Error<T extends Stream<any, any, any>> = [T] extends [Stream<infer _A, infer _E, infer _R>] ? _E : neverExample
(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"Extract the services type from a Stream type.
Signature
type Services<T extends Stream<any, any, any>> = [T] extends [Stream<infer _A, infer _E, infer _R>] ? _R : neverExample
(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
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 lambdatype 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]Extract the success type from a Stream type.
Signature
type Success<T extends Stream<any, any, any>> = [T] extends [Stream<infer _A, infer _E, infer _R>] ? _A : neverExample
(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 = 42Zipping
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, ER, RR, AR, EL, RL>(left: Stream<AL, ER, RR>, right: Stream<AR, EL, RL>): Stream<[AL, AR], ER | EL, RR | RL>;}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)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)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
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)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
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
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 ]}))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)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)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
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
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
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
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
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)