/**
* 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
*/
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 * as Context from "./Context.ts";
import * as Duration from "./Duration.ts";
import * as Effect from "./Effect.ts";
import * as ExecutionPlan from "./ExecutionPlan.ts";
import * as Exit from "./Exit.ts";
import type * as Filter from "./Filter.ts";
import type { LazyArg } from "./Function.ts";
import type { TypeLambda } from "./HKT.ts";
import type * as Layer from "./Layer.ts";
import type { Severity } from "./LogLevel.ts";
import * as Option from "./Option.ts";
import type { Pipeable } from "./Pipeable.ts";
import type { Predicate, Refinement } from "./Predicate.ts";
import * as PubSub from "./PubSub.ts";
import * as Pull from "./Pull.ts";
import * as Queue from "./Queue.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 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 declare const TypeId: TypeId;
/**
* 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 declare const isStream: (u: unknown) => u is Stream;
/**
* 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 declare const DefaultChunkSize: number;
/**
* 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 declare const fromChannel: , E, R>(channel: Channel.Channel) => Stream ? A : never, E, R>;
/**
* 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 declare const fromEffect: (effect: Effect.Effect) => Stream;
/**
* 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 declare const service: (service: Context.Key) => Stream;
/**
* 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 declare const serviceOption: (service: Context.Key) => Stream>;
/**
* 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 declare const fromEffectDrain: (effect: Effect.Effect) => Stream;
/**
* 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 declare const fromEffectRepeat: (effect: Effect.Effect) => Stream, R>;
/**
* 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 declare const fromEffectSchedule: (effect: Effect.Effect, schedule: Schedule.Schedule) => Stream;
/**
* 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 declare const tick: (interval: Duration.Input) => Stream;
/**
* 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 declare const fromPull: (pull: Effect.Effect, E, void, R>, EX, RX>) => Stream | EX, R | RX>;
/**
* 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 declare const transformPull: (self: Stream, f: (pull: Pull.Pull, E, void>, scope: Scope.Scope) => Effect.Effect, E2, void, R2>, EX, RX>) => Stream, R | R2 | RX>;
/**
* 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 declare const transformPullBracket: (self: Stream, f: (pull: Pull.Pull, E, void, R>, scope: Scope.Scope, forkedScope: Scope.Scope) => Effect.Effect, E2, void, R2>, EX, RX>) => Stream, R | R2 | RX>;
/**
* 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 declare const toChannel: (stream: Stream) => Channel.Channel, E, void, unknown, unknown, unknown, R>;
/**
* 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 declare const callback: (f: (queue: Queue.Queue) => Effect.Effect, options?: {
readonly bufferSize?: number | undefined;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
}) => Stream>;
/**
* 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 declare const empty: Stream;
/**
* 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 declare const succeed: (value: A) => Stream;
/**
* 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 declare const make: >(...values: As) => Stream;
/**
* 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 declare const sync: (evaluate: LazyArg) => Stream;
/**
* 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 declare const suspend: (stream: LazyArg>) => Stream;
/**
* 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 declare const fail: (error: E) => Stream;
/**
* 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 declare const failSync: (evaluate: LazyArg) => Stream;
/**
* 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 declare const failCause: (cause: Cause.Cause) => Stream;
/**
* 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 declare const die: (defect: unknown) => Stream;
/**
* 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 declare const failCauseSync: (evaluate: LazyArg>) => Stream;
/**
* 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 declare const fromIteratorSucceed: (iterator: IterableIterator, maxChunkSize?: number) => Stream;
/**
* 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 declare const fromIterable: (iterable: Iterable, options?: {
readonly chunkSize?: number | undefined;
}) => Stream;
/**
* 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 declare const fromIterableEffect: (iterable: Effect.Effect, E, R>) => Stream;
/**
* 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 declare const fromIterableEffectRepeat: (iterable: Effect.Effect, E, R>) => Stream, R>;
/**
* 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 declare const fromArray: (array: ReadonlyArray) => Stream;
/**
* 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 declare const fromArrayEffect: (effect: Effect.Effect, E, R>) => Stream, R>;
/**
* 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 declare const fromArrays: >>(...arrays: Arr) => Stream;
/**
* 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 declare const fromQueue: (queue: Queue.Dequeue) => Stream>;
/**
* 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 declare const fromPubSub: (pubsub: PubSub.PubSub) => Stream;
/**
* 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 declare const fromPubSubTake: (pubsub: PubSub.PubSub>) => Stream;
/**
* 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 declare const fromReadableStream: (options: {
readonly evaluate: LazyArg>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}) => Stream;
/**
* 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 declare const fromAsyncIterable: (iterable: AsyncIterable, onError: (error: unknown) => E) => Stream;
/**
* 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 declare const fromSchedule: (schedule: Schedule.Schedule) => Stream;
/**
* 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 declare const fromSubscription: (pubsub: PubSub.Subscription) => Stream;
/**
* 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 declare const fromEventListener: (target: EventListener, type: string, options?: boolean | {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly bufferSize?: number | undefined;
} | undefined) => Stream;
/**
* 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 declare const unfold: (s: S, f: (s: S) => Effect.Effect) => Stream;
/**
* 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 declare const paginate: (s: S, f: (s: S) => Effect.Effect, Option.Option], E, R>) => Stream;
/**
* 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 declare const iterate: (value: A, next: (value: A) => A) => Stream;
/**
* 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 declare const range: (min: number, max: number, chunkSize?: number) => Stream;
/**
* 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 declare const never: Stream;
/**
* 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 declare const unwrap: (effect: Effect.Effect, E, R>) => Stream>;
/**
* 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 declare const scoped: (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
*/
export declare 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;
};
/**
* 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 declare 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;
};
/**
* 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 declare 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;
};
/**
* 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 declare 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;
};
/**
* 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 declare 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;
};
/**
* 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 declare const flattenEffect: , 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;
/**
* 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 declare 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;
};
/**
* 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 declare const result: (self: Stream) => Stream, never, R>;
/**
* 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 declare 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;
};
/**
* 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 declare 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;
};
/**
* 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 declare 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;
};
/**
* 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 declare 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;
};
/**
* Switches to the latest stream produced by the mapping function, interrupting
* the previous stream when a new element arrives.
*
* **Example** (Switching to the latest stream)
*
* ```ts import.meta.vitest
* 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 ]
* }))
* ```
*
* @stability stable
* @category sequencing
* @since 4.0.0
*/
export declare const switchMap: {
/**
* Switches to the latest stream produced by the mapping function, interrupting
* the previous stream when a new element arrives.
*
* **Example** (Switching to the latest stream)
*
* ```ts import.meta.vitest
* 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 ]
* }))
* ```
*
* @stability stable
* @category sequencing
* @since 4.0.0
*/
(f: (a: A) => Stream, options?: {
readonly concurrency?: Concurrency | undefined;
readonly bufferSize?: number | undefined;
} | undefined): (self: Stream) => Stream;
/**
* Switches to the latest stream produced by the mapping function, interrupting
* the previous stream when a new element arrives.
*
* **Example** (Switching to the latest stream)
*
* ```ts import.meta.vitest
* 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 ]
* }))
* ```
*
* @stability stable
* @category sequencing
* @since 4.0.0
*/
(self: Stream, f: (a: A) => Stream, options?: {
readonly concurrency?: Concurrency | undefined;
readonly bufferSize?: number | undefined;
} | undefined): Stream;
};
/**
* 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.
*
* **Example** (Flattening nested streams)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export declare const flatten: , any, any> | {
readonly concurrency?: Concurrency | undefined;
readonly bufferSize?: number | undefined;
} | undefined = {
readonly concurrency?: Concurrency | undefined;
readonly bufferSize?: number | undefined;
}>(selfOrOptions?: Arg, options?: {
readonly concurrency?: Concurrency | undefined;
readonly bufferSize?: number | undefined;
} | undefined) => [Arg] extends [Stream, infer _E2, infer _R2>] ? Stream<_A, _E | _E2, _R | _R2> : (self: Stream, E2, R2>) => Stream;
/**
* Flattens a stream of non-empty arrays into a stream of elements.
*
* **Example** (Flattening a stream of non-empty arrays into a stream of elements)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 4.0.0
*/
export declare const flattenArray: (self: Stream, E, R>) => Stream;
/**
* Converts this stream to one that runs its effects but emits no elements.
*
* **Example** (Draining stream values)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export declare const drain: (self: Stream) => Stream;
/**
* 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.
*
* **Example** (Draining a stream in the background)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
export declare const drainFork: {
/**
* 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.
*
* **Example** (Draining a stream in the background)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(that: Stream): (self: Stream) => Stream;
/**
* 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.
*
* **Example** (Draining a stream in the background)
*
* ```ts import.meta.vitest
* import { Effect, Stream } from "effect"
*
* const events: Array = []
* 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"]
* ```
*
* @stability stable
* @category sequencing
* @since 2.0.0
*/
(self: Stream, that: Stream): Stream;
};
/**
* Repeats the entire stream according to the provided schedule.
*
* **Example** (Repeating a stream on a schedule)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category repetition
* @since 2.0.0
*/
export declare const repeat: {
/**
* Repeats the entire stream according to the provided schedule.
*
* **Example** (Repeating a stream on a schedule)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category repetition
* @since 2.0.0
*/
(schedule: Schedule.Schedule | (($: (_: Schedule.Schedule) => Schedule.Schedule) => Schedule.Schedule)): (self: Stream) => Stream;
/**
* Repeats the entire stream according to the provided schedule.
*
* **Example** (Repeating a stream on a schedule)
*
* ```ts import.meta.vitest
* 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)
* ```
*
* @stability stable
* @category repetition
* @since 2.0.0
*/
(self: Stream, schedule: Schedule.Schedule | (($: (_: Schedule.Schedule) => Schedule.Schedule) => Schedule.Schedule