/**
* Describes effectful sources that emit values over time.
*
* A `Stream` 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.
*
* @stability stable
* @since 2.0.0
*/
// @effect-diagnostics returnEffectInGen:off
import * as Arr from "./Array.ts"
import * as ByteSize from "./ByteSize.ts"
import * as Cause from "./Cause.ts"
import * as Channel from "./Channel.ts"
import { Clock } from "./Clock.ts"
import * as Context from "./Context.ts"
import * as Duration from "./Duration.ts"
import * as Effect from "./Effect.ts"
import * as Equal from "./Equal.ts"
import * as ExecutionPlan from "./ExecutionPlan.ts"
import * as Exit from "./Exit.ts"
import * as Fiber from "./Fiber.ts"
import type * as Filter from "./Filter.ts"
import type { LazyArg } from "./Function.ts"
import { constant, constTrue, constVoid, dual, identity } from "./Function.ts"
import type { TypeLambda } from "./HKT.ts"
import * as Count from "./internal/count.ts"
import * as internalExecutionPlan from "./internal/executionPlan.ts"
import * as internal from "./internal/stream.ts"
import { addSpanStackTrace } from "./internal/tracer.ts"
import * as Iterable from "./Iterable.ts"
import * as Latch from "./Latch.ts"
import type * as Layer from "./Layer.ts"
import type { Severity } from "./LogLevel.ts"
import * as MutableHashMap from "./MutableHashMap.ts"
import * as MutableList from "./MutableList.ts"
import * as Option from "./Option.ts"
import type { Pipeable } from "./Pipeable.ts"
import type { Predicate, Refinement } from "./Predicate.ts"
import { hasProperty, isNotUndefined, isTagged } from "./Predicate.ts"
import * as PubSub from "./PubSub.ts"
import * as Pull from "./Pull.ts"
import * as Queue from "./Queue.ts"
import * as RcMap from "./RcMap.ts"
import * as RcRef from "./RcRef.ts"
import * as Result from "./Result.ts"
import * as Schedule from "./Schedule.ts"
import * as Scope from "./Scope.ts"
import * as Sink from "./Sink.ts"
import { isString } from "./String.ts"
import type * as Take from "./Take.ts"
import type { ParentSpan, SpanOptions } from "./Tracer.ts"
import type {
Concurrency,
Covariant,
ExcludeReason,
ExcludeTag,
ExtractReason,
ExtractTag,
NarrowReason,
NoInfer,
OmitReason,
ReasonOf,
ReasonTags,
Simplify,
Tags,
TupleOf,
unassigned
} from "./Types.ts"
import type * as Unify from "./Unify.ts"
/**
* String literal type used as the unique brand for `Stream` values.
*
* @stability stable
* @category type IDs
* @since 4.0.0
*/
export type TypeId = "~effect/Stream"
/**
* 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 {@link isStream} for the public guard that checks this identifier
*
* @stability stable
* @category type IDs
* @since 4.0.0
*/
export const TypeId: TypeId = "~effect/Stream"
/**
* A `Stream` 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.
*
* **Example** (Creating and consuming streams)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface Stream extends Variance, Pipeable {
readonly channel: Channel.Channel, E, void, unknown, unknown, unknown, R>
[Unify.typeSymbol]?: unknown
[Unify.unifySymbol]?: StreamUnify
[Unify.ignoreSymbol]?: StreamUnifyIgnore
}
/**
* Type-level unification hook for Stream within the Effect type system.
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface StreamUnify extends Effect.EffectUnify {
Stream?: () => A[Unify.typeSymbol] extends Stream | infer _ ? Stream : never
}
/**
* Type-level marker that excludes Stream from unification.
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface StreamUnifyIgnore {
Effect?: true
}
/**
* Type lambda for Stream used in higher-kinded type operations.
*
* **Example** (Using the stream type lambda)
*
* ```ts import.meta.vitest
* import { Effect, HKT, Stream } from "effect"
*
* // Create a Stream type using the type lambda
* type NumberStream = HKT.Kind
* // Equivalent to: Stream
* const stream: NumberStream = Stream.make(1, 2, 3)
* await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]
* ```
*
* @stability stable
* @category utility types
* @since 2.0.0
*/
export interface StreamTypeLambda extends TypeLambda {
readonly type: Stream
}
/**
* Type-level variance marker for `Stream`.
*
* **Details**
*
* The emitted value `A`, error `E`, and service requirement `R` type
* parameters are covariant.
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface Variance {
readonly [TypeId]: VarianceStruct
}
/**
* Structural encoding used by `Variance` to record each `Stream` type
* parameter's variance.
*
* **Details**
*
* `_A`, `_E`, and `_R` are covariant markers.
*
* @stability stable
* @category models
* @since 3.4.0
*/
export interface VarianceStruct {
readonly _A: Covariant
readonly _E: Covariant
readonly _R: Covariant
}
/**
* Extract the success type from a Stream type.
*
* **Example** (Extracting the success type from a Stream type)
*
* ```ts import.meta.vitest
* import { Stream } from "effect"
*
* type NumberStream = Stream.Stream
* type SuccessType = Stream.Success
* const value: SuccessType = 42
* ```
*
* @stability stable
* @category utility types
* @since 3.4.0
*/
export type Success = T extends Stream ? _A : never
/**
* Extract the error type from a Stream type.
*
* **Example** (Extracting the error type from a Stream type)
*
* ```ts import.meta.vitest
* import { Stream } from "effect"
*
* type NumberStream = Stream.Stream
* type ErrorType = Stream.Error
* const error: ErrorType = "boom"
* ```
*
* @stability stable
* @category utility types
* @since 3.4.0
*/
export type Error = T extends Stream ? _E : never
/**
* Extract the services type from a Stream type.
*
* **Example** (Extracting the services type from a Stream type)
*
* ```ts import.meta.vitest
* import { Stream } from "effect"
*
* interface Database {
* query: (sql: string) => unknown
* }
* type NumberStream = Stream.Stream
* type RequiredServices = Stream.Services
* const services: RequiredServices = { db: { query: (sql) => sql } }
* services.db.query("SELECT 1") // => "SELECT 1"
* ```
*
* @stability stable
* @category utility types
* @since 4.0.0
*/
export type Services = T extends Stream ? _R : never
/**
* Checks whether a value is a Stream.
*
* **Example** (Checking whether a value is a Stream)
*
* ```ts import.meta.vitest
* import { Stream } from "effect"
*
* Stream.isStream(Stream.make(1, 2, 3)) // => true
* Stream.isStream({ data: [1, 2, 3] }) // => false
* ```
*
* @stability stable
* @category guards
* @since 4.0.0
*/
export const isStream = (u: unknown): u is Stream => hasProperty(u, TypeId)
/**
* The default chunk size used by Stream constructors and combinators.
*
* **Example** (Reading the default chunk size)
*
* ```ts import.meta.vitest
* import { Stream } from "effect"
*
* Stream.DefaultChunkSize // => 4096
* ```
*
* @stability stable
* @category constants
* @since 2.0.0
*/
export const DefaultChunkSize: number = Channel.DefaultChunkSize
/**
* Describes how merged streams decide when to halt.
*
* @stability stable
* @category models
* @since 4.0.0
*/
export type HaltStrategy = Channel.HaltStrategy
/**
* Creates a stream from a array-emitting `Channel`.
*
* **Example** (Creating a stream from an array-emitting channel)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromChannel: , E, R>(
channel: Channel.Channel
) => Stream ? A : never, E, R> = internal.fromChannel
/**
* Creates a stream from an effect.
*
* **Example** (Creating a stream from an effect)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const stream = Stream.fromEffect(Effect.succeed(42))
* await Effect.runPromise(Stream.runCollect(stream)) // => [42]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromEffect = (effect: Effect.Effect): Stream =>
fromChannel(Channel.fromEffect(Effect.map(effect, Arr.of)))
/**
* Accesses a service from the context and emits it as a single element.
*
* **Example** (Accessing a service as a stream)
*
* ```ts import.meta.vitest
* import { Context, Effect, Stream } from "effect"
*
* class Greeter extends Context.Service 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!"]
* ```
*
* @stability stable
* @category accessors
* @since 4.0.0
*/
export const service = (service: Context.Key): Stream => fromEffect(Effect.service(service))
/**
* 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.
*
* **Example** (Accessing an optional service as a stream)
*
* ```ts import.meta.vitest
* import { Context, Effect, Option, Stream } from "effect"
*
* class Greeter extends Context.Service 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!"]
* ```
*
* @stability stable
* @category accessors
* @since 4.0.0
*/
export const serviceOption = (service: Context.Key): Stream> =>
fromEffect(Effect.serviceOption(service))
/**
* Creates a stream that runs the effect and emits no elements.
*
* **Example** (Draining an effect into a stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* let drained = false
* await Effect.runPromise(
* Stream.fromEffectDrain(Effect.sync(() => {
* drained = true
* })).pipe(Stream.runDrain)
* )
* drained // => true
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromEffectDrain = (effect: Effect.Effect): Stream =>
fromPull(Effect.succeed(Effect.flatMap(effect, () => Cause.done())))
/**
* Creates a stream from an effect producing a value of type `A` which repeats forever.
*
* **Example** (Repeating an effect forever)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* let n = 0
* const stream = Stream.fromEffectRepeat(Effect.sync(() => ++n)).pipe(Stream.take(5))
* await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, 4, 5]
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromEffectRepeat = (effect: Effect.Effect): Stream, R> =>
fromPull(Effect.succeed(Effect.map(effect, Arr.of)))
/**
* Creates a stream from an effect producing a value of type `A`, which is
* repeated using the specified schedule.
*
* **Example** (Repeating an effect with a schedule)
*
* ```ts import.meta.vitest
* 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"]
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromEffectSchedule = (
effect: Effect.Effect,
schedule: Schedule.Schedule
): Stream =>
fromPull(Effect.gen(function*() {
const step = yield* Schedule.toStepWithMetadata(schedule)
let s = yield* Effect.provideService(effect, Schedule.CurrentMetadata, Schedule.CurrentMetadata.defaultValue())
let initial = true
const pull = Effect.suspend(() => step(s as AS)).pipe(
Effect.flatMap((meta) => Effect.provideService(effect, Schedule.CurrentMetadata, meta)),
Effect.map((next) => {
s = next
return Arr.of(next)
})
) as Pull.Pull, E | ES, void, R | RS>
return Effect.suspend(() => {
if (initial) {
initial = false
return Effect.succeed(Arr.of(s))
}
return pull
})
}))
/**
* Creates a stream that emits `void` immediately once, then emits another
* `void` after each specified interval.
*
* **Example** (Emitting ticks on an interval)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* await Effect.runPromise(Stream.tick(0).pipe(Stream.take(3), Stream.runCollect)) // => [undefined, undefined, undefined]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const tick = (interval: Duration.Input): Stream =>
fromPull(Effect.sync(() => {
let first = true
const effect = Effect.succeed(Arr.of(undefined))
const delayed = Effect.delay(effect, interval)
return Effect.suspend(() => {
if (first) {
first = false
return effect
}
return delayed
})
}))
/**
* 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.
*
* **Example** (Creating a stream from a pull effect)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromPull = (
pull: Effect.Effect, E, void, R>, EX, RX>
): Stream | EX, R | RX> => fromChannel(Channel.fromPull(pull))
/**
* Derives a stream by transforming its pull effect.
*
* **Example** (Transforming a pull effect)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const transformPull = (
self: Stream,
f: (pull: Pull.Pull, E, void>, scope: Scope.Scope) => Effect.Effect<
Pull.Pull, E2, void, R2>,
EX,
RX
>
): Stream, R | R2 | RX> =>
fromChannel(
Channel.fromTransform((_, scope) =>
Effect.flatMap(Channel.toPullScoped(self.channel, scope), (pull) => f(pull as any, scope))
)
)
/**
* 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.
*
* **Example** (Transforming a stream by effectfully transforming its pull effect)
*
* ```ts import.meta.vitest
* import { Effect, Scope, Stream } from "effect"
*
* const finalized: Array = []
* 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]
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const transformPullBracket = (
self: Stream,
f: (
pull: Pull.Pull, E, void, R>,
scope: Scope.Scope,
forkedScope: Scope.Scope
) => Effect.Effect<
Pull.Pull, E2, void, R2>,
EX,
RX
>
): Stream, R | R2 | RX> =>
fromChannel(
Channel.fromTransformBracket((_, scope, forkedScope) =>
Effect.flatMap(Channel.toPullScoped(self.channel, scope), (pull) => f(pull, scope, forkedScope))
)
)
/**
* Creates a channel from a stream.
*
* **Example** (Converting a stream to a channel)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category destructors
* @since 2.0.0
*/
export const toChannel = (
stream: Stream
): Channel.Channel, E, void, unknown, unknown, unknown, R> => stream.channel
/**
* 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.
*
* **Example** (Creating a stream from a callback that can emit values into a queue)
*
* ```ts import.meta.vitest
* import { Effect, Queue, Stream } from "effect"
*
* const stream = Stream.callback((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]
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const callback = (
f: (queue: Queue.Queue) => Effect.Effect,
options?: {
readonly bufferSize?: number | undefined
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined
}
): Stream> => fromChannel(Channel.callbackArray(f, options))
/**
* Creates an empty stream.
*
* **Example** (Creating an empty stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* await Effect.runPromise(Stream.runCollect(Stream.empty)) // => []
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const empty: Stream = fromChannel(Channel.empty)
/**
* Creates a single-valued pure stream.
*
* **Example** (Creating a single-valued pure stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* await Effect.runPromise(Stream.runCollect(Stream.succeed(3))) // => [3]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const succeed = (value: A): Stream => fromChannel(Channel.succeed(Arr.of(value)))
/**
* Creates a stream from a sequence of values.
*
* **Example** (Creating a stream from a sequence of values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const stream = Stream.make(1, 2, 3)
*
* await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const make = >(...values: As): Stream => fromArray(values)
/**
* 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.
*
* **Example** (Evaluating a value synchronously)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* await Effect.runPromise(Stream.sync(() => 2 + 1).pipe(Stream.runCollect)) // => [3]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const sync = (evaluate: LazyArg): Stream => fromChannel(Channel.sync(() => Arr.of(evaluate())))
/**
* Creates a lazily constructed stream.
*
* **Details**
*
* The stream factory is evaluated each time the stream is run.
*
* **Example** (Creating a lazily constructed stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* await Effect.runPromise(Stream.suspend(() => Stream.make(1, 2, 3)).pipe(Stream.runCollect)) // => [1, 2, 3]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const suspend = (stream: LazyArg>): Stream => {
const suspended = fromChannel(Channel.suspend(() => stream().channel))
suspendedStreams.set(suspended, stream)
return suspended
}
const suspendedStreams = new WeakMap, LazyArg>>()
/**
* Terminates with the specified error.
*
* **Example** (Failing a stream)
*
* ```ts import.meta.vitest
* import { Effect, Exit, Stream } from "effect"
*
* await Effect.runPromise(Effect.exit(Stream.runCollect(Stream.fail("Uh oh!")))) // => Exit.fail("Uh oh!")
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fail = (error: E): Stream => fromChannel(Channel.fail(error))
/**
* Terminates with the specified lazily evaluated error.
*
* **Example** (Failing a stream lazily)
*
* ```ts import.meta.vitest
* 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!")
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const failSync = (evaluate: LazyArg): Stream => fromChannel(Channel.failSync(evaluate))
/**
* Creates a stream that fails with the specified `Cause`.
*
* **Example** (Failing with a cause)
*
* ```ts import.meta.vitest
* 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"]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const failCause = (cause: Cause.Cause): Stream => fromChannel(Channel.failCause(cause))
/**
* The stream that dies with the specified defect.
*
* **Example** (Dying with a defect)
*
* ```ts import.meta.vitest
* 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))
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const die = (defect: unknown): Stream => fromChannel(Channel.die(defect))
/**
* The stream that always fails with the specified lazily evaluated `Cause`.
*
* **Example** (Failing with a lazy cause)
*
* ```ts import.meta.vitest
* 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")
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const failCauseSync = (evaluate: LazyArg>): Stream =>
fromChannel(Channel.failCauseSync(evaluate))
/**
* Creates a stream that consumes values from an iterator.
*
* **Details**
*
* The `maxChunkSize` parameter controls how many values are pulled per chunk.
* Finite fractional values are rounded down, while `NaN` and non-positive
* values are treated as `1`.
*
* **Example** (Consuming values from an iterator)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromIteratorSucceed = (iterator: IterableIterator, maxChunkSize?: number): Stream =>
fromChannel(Channel.fromIteratorArray(() => iterator, maxChunkSize))
/**
* Creates a new `Stream` from an iterable collection of values.
*
* **Details**
*
* - `chunkSize`: Maximum number of values emitted per chunk. Finite fractional
* values are rounded down, while `NaN` and non-positive values are treated as
* `1`.
*
* **Example** (Creating a stream from an iterable)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromIterable = (
iterable: Iterable,
options?: {
readonly chunkSize?: number | undefined
}
): Stream =>
Array.isArray(iterable) && options?.chunkSize === undefined
? fromArray(iterable)
: fromChannel(Channel.fromIterableArray(iterable, options?.chunkSize))
/**
* 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.
*
* **Example** (Creating a stream from an iterable effect)
*
* ```ts import.meta.vitest
* import { Context, Effect, Stream } from "effect"
*
* class UserRepo extends Context.Service>
* }>()("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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromIterableEffect = (iterable: Effect.Effect, E, R>): Stream =>
unwrap(Effect.map(iterable, fromIterable))
/**
* Creates a stream by repeatedly running an effect that yields an iterable of values.
*
* **Example** (Repeating an iterable effect)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromIterableEffectRepeat = (
iterable: Effect.Effect, E, R>
): Stream, R> => flatMap(fromEffectRepeat(iterable), fromIterable)
/**
* Creates a stream from an array of values.
*
* **Example** (Creating a stream from an array of values)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromArray = (array: ReadonlyArray): Stream =>
Arr.isReadonlyArrayNonEmpty(array) ? fromChannel(Channel.succeed(array)) : empty
/**
* 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.
*
* **Example** (Creating a stream from an effect that produces an array of values)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromArrayEffect = (
effect: Effect.Effect, E, R>
): Stream, R> => unwrap(Effect.map(effect, fromArray)) as any
/**
* Creates a stream from an arbitrary number of arrays.
*
* **Example** (Creating a stream from an arbitrary number of arrays)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromArrays = >>(
...arrays: Arr
): Stream => fromChannel(Channel.fromArray(Arr.filter(arrays, Arr.isReadonlyArrayNonEmpty)))
/**
* 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.
*
* **Example** (Creating a stream from a queue of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.unbounded()
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromQueue = (queue: Queue.Dequeue): Stream> =>
fromChannel(Channel.fromQueueArray(queue))
/**
* Creates a stream from a subscription to a `PubSub`.
*
* **Example** (Creating a stream from a subscription to a PubSub)
*
* ```ts import.meta.vitest
* import { Effect, Fiber, PubSub, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const pubsub = yield* PubSub.unbounded({ 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromPubSub = (pubsub: PubSub.PubSub): Stream => fromChannel(Channel.fromPubSubArray(pubsub))
/**
* Creates a stream from a PubSub of `Take` values.
*
* **Details**
*
* `Take` values include end and failure signals.
*
* **Example** (Creating a stream from PubSub takes)
*
* ```ts import.meta.vitest
* import { Effect, Exit, PubSub, Stream, Take } from "effect"
*
* const program = Effect.gen(function*() {
* const pubsub = yield* PubSub.unbounded>({
* replay: 3
* })
*
* yield* PubSub.publish(pubsub, [1])
* yield* PubSub.publish(pubsub, [2])
* yield* PubSub.publish(pubsub, Exit.succeed(undefined))
*
* const values = yield* Stream.fromPubSubTake(pubsub).pipe(Stream.runCollect)
* values // => [1, 2]
* })
*
* await Effect.runPromise(program)
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromPubSubTake = (pubsub: PubSub.PubSub>): Stream =>
fromChannel(Channel.fromPubSubTake(pubsub))
/**
* Creates a stream from a lazily supplied Web `ReadableStream`.
*
* **Details**
*
* The stream reads from a `ReadableStreamDefaultReader` and maps read failures
* with `onError`. Finalization cancels the reader and releases its lock, or
* only releases the lock when `releaseLockOnEnd` is set. A rejected cancel
* hook becomes a defect.
*
* **Example** (Creating a stream from a ReadableStream)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromReadableStream = (
options: {
readonly evaluate: LazyArg>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}
): Stream => fromChannel(Channel.fromReadableStream(options))
/**
* Creates a stream from an AsyncIterable.
*
* **Example** (Creating a stream from an AsyncIterable)
*
* ```ts import.meta.vitest
* 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]
* }))
*
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromAsyncIterable = (
iterable: AsyncIterable,
onError: (error: unknown) => E
): Stream => fromChannel(Channel.fromAsyncIterableArray(iterable, onError))
/**
* Creates a stream that emits each output of a schedule that does not require input,
* for as long as the schedule continues.
*
* **Example** (Creating a stream from a schedule)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const fromSchedule = (schedule: Schedule.Schedule): Stream =>
fromPull(
Effect.map(
Schedule.toStepWithSleep(schedule),
(step) => Pull.catchDone(Effect.map(step(void 0), Arr.of), () => Cause.done())
)
)
/**
* 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.
*
* **Example** (Creating a stream from a PubSub subscription)
*
* ```ts import.meta.vitest
* import { Effect, PubSub, Stream } from "effect"
*
* const program = Effect.scoped(Effect.gen(function*() {
* const pubsub = yield* PubSub.unbounded()
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export const fromSubscription = (pubsub: PubSub.Subscription): Stream =>
fromChannel(Channel.fromSubscriptionArray(pubsub))
/**
* Interface representing an event listener target.
*
* @stability stable
* @category models
* @since 3.4.0
*/
export interface EventListener {
addEventListener(
event: string,
f: (event: A) => void,
options?: {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly signal?: AbortSignal
} | boolean
): void
removeEventListener(
event: string,
f: (event: A) => void,
options?: {
readonly capture?: boolean
} | boolean
): void
}
/**
* Creates a stream from an event listener.
*
* **Gotchas**
*
* Event listeners cannot backpressure the source. When a bounded `bufferSize`
* is full, additional events are dropped rather than suspending the listener.
*
* **Example** (Creating a stream from an event listener)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* class NumberTarget implements Stream.EventListener {
* 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]
* }))
*
* ```
*
* @stability stable
* @category constructors
* @since 3.1.0
*/
export const fromEventListener = (
target: EventListener,
type: string,
options?: boolean | {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
} | undefined
): Stream =>
callback((queue) => {
const once = typeof options === "object" && options.once
function emit(event: any) {
Queue.offerUnsafe(queue, event)
if (once) Queue.endUnsafe(queue)
}
return Effect.acquireRelease(
Effect.sync(() => target.addEventListener(type, emit, options)),
() => Effect.sync(() => target.removeEventListener(type, emit, options))
)
}, { bufferSize: typeof options === "object" ? options.bufferSize : undefined })
/**
* 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.
*
* **Example** (Unfolding stream state)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const unfold = (
s: S,
f: (s: S) => Effect.Effect
): Stream =>
fromPull(Effect.sync(() => {
let state = s
return Effect.flatMap(Effect.suspend(() => f(state)), (next) => {
if (next === undefined) return Cause.done()
state = next[1]
return Effect.succeed(Arr.of(next[0]))
})
}))
/**
* Creates a stream by repeatedly evaluating an effectful page function.
*
* **When to use**
*
* Use to consume paginated APIs where each step returns a batch of values
* together with an optional next state.
*
* **Details**
*
* This is similar to {@link unfold}, but each step can emit zero or more values
* and independently decide whether another state should be requested.
*
* **Example** (Paginating stream state)
*
* ```ts import.meta.vitest
* import { Effect, Option, Stream } from "effect"
*
* const stream = Stream.paginate(0, (n: number) =>
* Effect.succeed(
* [
* [n],
* n < 3 ? Option.some(n + 1) : Option.none()
* ] as const
* ))
*
* await Effect.runPromise(Stream.runCollect(stream)) // => [0, 1, 2, 3]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const paginate = (
s: S,
f: (
s: S
) => Effect.Effect, Option.Option], E, R>
): Stream =>
fromPull(Effect.sync(() => {
let state = s
let done = false
return Effect.suspend(function loop(): Pull.Pull, E, void, R> {
if (done) return Cause.done()
return Effect.flatMap(f(state), ([a, s]) => {
if (Option.isNone(s)) {
done = true
} else {
state = s.value
}
if (!Arr.isReadonlyArrayNonEmpty(a)) return loop()
return Effect.succeed(a)
})
})
}))
/**
* Creates an infinite stream by repeatedly applying a function to a seed value.
*
* **Example** (Iterating from a seed value)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const iterate = (value: A, next: (value: A) => A): Stream =>
unfold(value, (a) => Effect.succeed([a, next(a)]))
/**
* Constructs a stream from a range of integers, including both endpoints.
*
* **Details**
*
* If the provided `min` is greater than `max`, the stream will not emit any
* values. Finite fractional `chunkSize` values are rounded down, while `NaN`
* and non-positive values are treated as `1`.
*
* **Example** (Creating a numeric range)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const range = (
min: number,
max: number,
chunkSize = Channel.DefaultChunkSize
): Stream =>
min > max ? empty : fromPull(Effect.sync(() => {
const size = Count.normalizeNonEmpty(chunkSize)
let start = min
let done = false
return Effect.suspend(() => {
if (done) return Cause.done()
const remaining = max - start + 1
if (remaining > size) {
const chunk = Arr.range(start, start + size - 1)
start += size
return Effect.succeed(chunk)
}
const chunk = Arr.range(start, start + remaining - 1)
done = true
return Effect.succeed(chunk)
})
}))
/**
* The stream that never produces any value or fails with any error.
*
* **Example** (Creating a never-ending stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const program = Stream.never.pipe(
* Stream.take(0),
* Stream.runCollect
* )
*
* await Effect.runPromise(program) // => []
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const never: Stream = fromChannel(Channel.never)
/**
* Creates a stream produced from an `Effect`.
*
* **Example** (Unwrapping a stream effect)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export const unwrap = (
effect: Effect.Effect, E, R>
): Stream> => fromChannel(Channel.unwrap(Effect.map(effect, toChannel)))
/**
* Runs a stream that requires `Scope` in a managed scope, ensuring its
* finalizers are run when the stream completes.
*
* **Example** (Scoping a stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category resource management
* @since 2.0.0
*/
export const scoped = (
self: Stream
): Stream> => fromChannel(Channel.scoped(self.channel))
/**
* Transforms the elements of this stream using the supplied function.
*
* **Example** (Mapping stream values)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
export const map: {
/**
* Transforms the elements of this stream using the supplied function.
*
* **Example** (Mapping stream values)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
(f: (a: A, i: number) => B): (self: Stream) => Stream
/**
* Transforms the elements of this stream using the supplied function.
*
* **Example** (Mapping stream values)
*
* ```ts import.meta.vitest
* 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]
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
(self: Stream, f: (a: A, i: number) => B): Stream
} = dual(2, (self: Stream, f: (a: A, i: number) => B): Stream =>
suspend(() => {
let i = 0
return fromChannel(Channel.map(
self.channel,
Arr.map((o) => f(o, i++))
))
}))
/**
* Replaces every element of the stream with the provided constant value.
*
* **Example** (Replacing stream elements)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const values = yield* Stream.make(1, 2, 3).pipe(
* Stream.as("x"),
* Stream.runCollect
* )
* values // => [ 'x', 'x', 'x' ]
* })
*
* await Effect.runPromise(program)
* ```
*
* @see {@link map} for deriving the replacement value from each element
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
export const as: {
/**
* Replaces every element of the stream with the provided constant value.
*
* **Example** (Replacing stream elements)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const values = yield* Stream.make(1, 2, 3).pipe(
* Stream.as("x"),
* Stream.runCollect
* )
* values // => [ 'x', 'x', 'x' ]
* })
*
* await Effect.runPromise(program)
* ```
*
* @see {@link map} for deriving the replacement value from each element
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
(value: B): (self: Stream) => Stream
/**
* Replaces every element of the stream with the provided constant value.
*
* **Example** (Replacing stream elements)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const values = yield* Stream.make(1, 2, 3).pipe(
* Stream.as("x"),
* Stream.runCollect
* )
* values // => [ 'x', 'x', 'x' ]
* })
*
* await Effect.runPromise(program)
* ```
*
* @see {@link map} for deriving the replacement value from each element
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
(self: Stream, value: B): Stream
} = dual(2, (self: Stream, value: B): Stream => map(self, () => value))
/**
* Maps both the failure and success channels of a stream.
*
* **Details**
*
* Maps every typed error in the cause, preserving defects, interrupts and
* reason order. Unlike typed recovery, mapping does not convert errors to defects.
*
* **Example** (Mapping both the failure and success channels of a stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const mapper = {
* onElement: (value: number) => value * 2,
* onError: (error: string) => `error: ${error}`
* }
*
* const program = Effect.gen(function*() {
* const success = yield* Stream.make(1, 2).pipe(
* Stream.mapBoth(mapper),
* Stream.runCollect
* )
* success // => [ 2, 4 ]
*
* const failure = yield* Stream.fail("boom").pipe(
* Stream.mapBoth(mapper),
* Stream.catch((error: string) => Stream.succeed(error)),
* Stream.runCollect
* )
* failure // => [ 'error: boom' ]
* })
*
* await Effect.runPromise(program)
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
export const mapBoth: {
/**
* Maps both the failure and success channels of a stream.
*
* **Details**
*
* Maps every typed error in the cause, preserving defects, interrupts and
* reason order. Unlike typed recovery, mapping does not convert errors to defects.
*
* **Example** (Mapping both the failure and success channels of a stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const mapper = {
* onElement: (value: number) => value * 2,
* onError: (error: string) => `error: ${error}`
* }
*
* const program = Effect.gen(function*() {
* const success = yield* Stream.make(1, 2).pipe(
* Stream.mapBoth(mapper),
* Stream.runCollect
* )
* success // => [ 2, 4 ]
*
* const failure = yield* Stream.fail("boom").pipe(
* Stream.mapBoth(mapper),
* Stream.catch((error: string) => Stream.succeed(error)),
* Stream.runCollect
* )
* failure // => [ 'error: boom' ]
* })
*
* await Effect.runPromise(program)
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
(
options: { readonly onElement: (a: A) => A2; readonly onError: (e: E) => E2 }
): (self: Stream) => Stream
/**
* Maps both the failure and success channels of a stream.
*
* **Details**
*
* Maps every typed error in the cause, preserving defects, interrupts and
* reason order. Unlike typed recovery, mapping does not convert errors to defects.
*
* **Example** (Mapping both the failure and success channels of a stream)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const mapper = {
* onElement: (value: number) => value * 2,
* onError: (error: string) => `error: ${error}`
* }
*
* const program = Effect.gen(function*() {
* const success = yield* Stream.make(1, 2).pipe(
* Stream.mapBoth(mapper),
* Stream.runCollect
* )
* success // => [ 2, 4 ]
*
* const failure = yield* Stream.fail("boom").pipe(
* Stream.mapBoth(mapper),
* Stream.catch((error: string) => Stream.succeed(error)),
* Stream.runCollect
* )
* failure // => [ 'error: boom' ]
* })
*
* await Effect.runPromise(program)
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
(
self: Stream,
options: { readonly onElement: (a: A) => A2; readonly onError: (e: E) => E2 }
): Stream
} = dual(2, (
self: Stream,
options: { readonly onElement: (a: A) => A2; readonly onError: (e: E) => E2 }
): Stream =>
self.pipe(
map(options.onElement),
mapError(options.onError)
))
/**
* Transforms each emitted chunk using the provided function, which receives the chunk and its index.
*
* **Example** (Mapping stream chunks)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
export const mapArray: {
/**
* Transforms each emitted chunk using the provided function, which receives the chunk and its index.
*
* **Example** (Mapping stream chunks)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
(
f: (a: Arr.NonEmptyReadonlyArray, i: number) => Arr.NonEmptyReadonlyArray
): (self: Stream) => Stream
/**
* Transforms each emitted chunk using the provided function, which receives the chunk and its index.
*
* **Example** (Mapping stream chunks)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
(
self: Stream,
f: (a: Arr.NonEmptyReadonlyArray, i: number) => Arr.NonEmptyReadonlyArray
): Stream
} = dual(2, (
self: Stream,
f: (a: Arr.NonEmptyReadonlyArray, i: number) => Arr.NonEmptyReadonlyArray
): Stream => fromChannel(Channel.map(self.channel, f)))
/**
* 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.
*
* **Example** (Effectfully mapping stream values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
export const mapEffect: {
/**
* 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.
*
* **Example** (Effectfully mapping stream values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
(
f: (a: A, i: number) => Effect.Effect,
options?: {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
} | undefined
): (self: Stream) => Stream
/**
* 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.
*
* **Example** (Effectfully mapping stream values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category mapping
* @since 2.0.0
*/
(
self: Stream,
f: (a: A, i: number) => Effect.Effect,
options?: {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
} | undefined
): Stream
} = dual((args) => isStream(args[0]), (
self: Stream,
f: (a: A, i: number) => Effect.Effect,
options?: {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
} | undefined
): Stream =>
self.channel.pipe(
Channel.flattenArray,
Channel.mapEffect(f, options),
Channel.map(Arr.of),
fromChannel
))
/**
* 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.
*
* **Example** (Flattening a stream of Effect values into a stream of their results)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export const flattenEffect: <
Arg extends Stream, any, any> | {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
} | undefined = {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
}
>(
selfOrOptions?: Arg,
options?: {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
} | undefined
) => [Arg] extends [Stream, infer _E, infer _R>] ?
Stream<_A, _EX | _E, _RX | _R>
: (self: Stream, E, R>) => Stream = dual(
(args) => isStream(args[0]),
(
self: Stream, E, R>,
options?: {
readonly concurrency?: Concurrency | undefined
readonly unordered?: boolean | undefined
} | undefined
): Stream => mapEffect(self, identity, options)
)
/**
* 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.
*
* **Example** (Effectfully mapping stream chunks)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
export const 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.
*
* **Example** (Effectfully mapping stream chunks)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
(
f: (a: Arr.NonEmptyReadonlyArray, i: number) => Effect.Effect, E2, R2>
): (self: Stream) => Stream
/**
* 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.
*
* **Example** (Effectfully mapping stream chunks)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category mapping
* @since 4.0.0
*/
(
self: Stream,
f: (a: Arr.NonEmptyReadonlyArray, i: number) => Effect.Effect, E2, R2>
): Stream
} = dual(2, (
self: Stream,
f: (a: Arr.NonEmptyReadonlyArray, i: number) => Effect.Effect, E2, R2>
): Stream => fromChannel(Channel.mapEffect(self.channel, f)))
/**
* 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.
*
* **Example** (Converting failures to results)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category error handling
* @since 4.0.0
*/
export const result = (self: Stream): Stream, never, R> =>
self.pipe(
map(Result.succeed),
catch_((e) => succeed(Result.fail(e)))
)
/**
* Runs the provided effect for each element while preserving the elements.
*
* **Example** (Tapping stream values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export const tap: {
/**
* Runs the provided effect for each element while preserving the elements.
*
* **Example** (Tapping stream values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(
f: (a: NoInfer) => Effect.Effect,
options?: {
readonly concurrency?: Concurrency | undefined
} | undefined
): (self: Stream) => Stream
/**
* Runs the provided effect for each element while preserving the elements.
*
* **Example** (Tapping stream values)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(
self: Stream,
f: (a: NoInfer) => Effect.Effect,
options?: {
readonly concurrency?: Concurrency | undefined
} | undefined
): Stream
} = dual((args) => isStream(args[0]), (
self: Stream,
f: (a: NoInfer) => Effect.Effect,
options?: {
readonly concurrency?: Concurrency | undefined
} | undefined
): Stream =>
mapEffect(
self,
(a) => Effect.as(f(a), a),
options
))
/**
* Returns a stream that effectfully "peeks" at elements and failures.
*
* **Example** (Tapping values and errors)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export const tapBoth: {
/**
* Returns a stream that effectfully "peeks" at elements and failures.
*
* **Example** (Tapping values and errors)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(
options: {
readonly onElement: (a: NoInfer) => Effect.Effect
readonly onError: (a: NoInfer) => Effect.Effect
readonly concurrency?: Concurrency | undefined
}
): (self: Stream) => Stream
/**
* Returns a stream that effectfully "peeks" at elements and failures.
*
* **Example** (Tapping values and errors)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(
self: Stream,
options: {
readonly onElement: (a: NoInfer) => Effect.Effect
readonly onError: (a: NoInfer) => Effect.Effect
readonly concurrency?: Concurrency | undefined
}
): Stream
} = dual(2, (
self: Stream,
options: {
readonly onElement: (a: NoInfer) => Effect.Effect
readonly onError: (a: NoInfer) => Effect.Effect
readonly concurrency?: Concurrency | undefined
}
): Stream =>
self.pipe(
tapError(options.onError),
tap(options.onElement, { concurrency: options.concurrency })
))
/**
* Runs a sink for all stream elements while still emitting them downstream.
*
* **Example** (Tapping values with a sink)
*
* ```ts import.meta.vitest
* import { Effect, Ref, Sink, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const seen = yield* Ref.make>([])
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export const tapSink: {
/**
* Runs a sink for all stream elements while still emitting them downstream.
*
* **Example** (Tapping values with a sink)
*
* ```ts import.meta.vitest
* import { Effect, Ref, Sink, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const seen = yield* Ref.make>([])
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(sink: Sink.Sink): (self: Stream) => Stream
/**
* Runs a sink for all stream elements while still emitting them downstream.
*
* **Example** (Tapping values with a sink)
*
* ```ts import.meta.vitest
* import { Effect, Ref, Sink, Stream } from "effect"
*
* const program = Effect.gen(function*() {
* const seen = yield* Ref.make>([])
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(self: Stream, sink: Sink.Sink): Stream
} = dual(
2,
(
self: Stream,
sink: Sink.Sink
): Stream =>
transformPullBracket(
self,
Effect.fnUntraced(function*(pull, _, scope) {
const upstreamLatch = Latch.makeUnsafe()
const sinkLatch = Latch.makeUnsafe()
let chunk: Arr.NonEmptyReadonlyArray | undefined = undefined
let causeSink: Cause.Cause | undefined = undefined
let sinkDone = false
let streamDone = false
const sinkUpstream = upstreamLatch.whenOpen(Effect.suspend(() => {
if (chunk) {
const arr = chunk!
chunk = undefined
if (!streamDone) upstreamLatch.closeUnsafe()
return Effect.as(sinkLatch.open, arr)
}
return Cause.done()
}))
yield* Effect.suspend(() => sink.transform(sinkUpstream, scope)).pipe(
(eff) =>
Effect.onExitPrimitive(eff, (exit) => {
sinkDone = true
if (Exit.isFailure(exit)) {
causeSink = exit.cause
}
return sinkLatch.open
}, true),
Effect.forkIn(scope)
)
const pullAndOffer = pull.pipe(
Effect.flatMap((chunk_) => {
// The sink may have exited while the upstream pull was suspended.
if (causeSink) return Effect.failCause(causeSink)
if (sinkDone) return Effect.succeed(chunk_)
chunk = chunk_
sinkLatch.closeUnsafe()
upstreamLatch.openUnsafe()
return Effect.as(sinkLatch.await, chunk_)
}),
Pull.catchDone((): Effect.Effect => {
if (causeSink) return Effect.failCause(causeSink)
if (sinkDone) return Cause.done()
streamDone = true
sinkLatch.closeUnsafe()
upstreamLatch.openUnsafe()
return Effect.flatMap(sinkLatch.await, (): Pull.Pull =>
causeSink ? Effect.failCause(causeSink) : Cause.done())
})
)
return Effect.suspend((): Pull.Pull, E | E2, void, R> => {
if (causeSink) {
return Effect.failCause(causeSink)
} else if (sinkDone) {
return pull
}
return pullAndOffer
})
})
)
)
/**
* 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.
*
* **Example** (Flat mapping stream values)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export const flatMap: {
/**
* 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.
*
* **Example** (Flat mapping stream values)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(
f: (a: A) => Stream,
options?: {
readonly concurrency?: Concurrency | undefined
readonly bufferSize?: number | undefined
} | undefined
): (self: Stream) => Stream
/**
* 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.
*
* **Example** (Flat mapping stream values)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(
self: Stream,
f: (a: A) => Stream,
options?: {
readonly concurrency?: Concurrency | undefined
readonly bufferSize?: number | undefined
} | undefined
): Stream
} = dual((args) => isStream(args[0]),