(
options:
| { readonly capacity: "unbounded" }
| {
readonly capacity: number
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
}
): <A, E, R>(
self: Stream<A, E, R>
) => Effect.Effect<
Queue.Dequeue<A, E | Cause.Done>,
never,
R | Scope.Scope
>
<A, E, R>(
self: Stream<A, E, R>,
options:
| { readonly capacity: "unbounded" }
| {
readonly capacity: number
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined
}
): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>Creates a scoped dequeue that is fed by the stream for concurrent consumption.
Details
Elements are offered to the queue as the stream runs. Stream completion is
signaled with Cause.Done, stream failures fail the queue, and the queue is
shut down when the surrounding scope closes.
Example (Converting a stream to a Queue for concurrent consumption)
import { Effect, Queue, Stream } from "effect"
const program = Effect.gen(function* () {
const queue = yield* Stream.toQueue(Stream.fromIterable([1, 2, 3]), { capacity: 8 })
const chunk = yield* Queue.takeBetween(queue, 1, 3)
return chunk
})destructors
Source effect/Stream.ts:1165431 lines
export const const toQueue: {
(
options:
| { readonly capacity: "unbounded" }
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
}
): <A, E, R>(
self: Stream<A, E, R>
) => Effect.Effect<
Queue.Dequeue<A, E | Cause.Done>,
never,
R | Scope.Scope
>
<A, E, R>(
self: Stream<A, E, R>,
options:
| { readonly capacity: "unbounded" }
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
}
): Effect.Effect<
Queue.Dequeue<A, E | Cause.Done>,
never,
R | Scope.Scope
>
}
Creates a scoped dequeue that is fed by the stream for concurrent
consumption.
Details
Elements are offered to the queue as the stream runs. Stream completion is
signaled with Cause.Done, stream failures fail the queue, and the queue is
shut down when the surrounding scope closes.
Example (Converting a stream to a Queue for concurrent consumption)
import { Effect, Queue, Stream } from "effect"
const program = Effect.gen(function* () {
const queue = yield* Stream.toQueue(Stream.fromIterable([1, 2, 3]), { capacity: 8 })
const chunk = yield* Queue.takeBetween(queue, 1, 3)
return chunk
})
toQueue: {
(
options: | {
readonly capacity: "unbounded"
}
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
}
capacity: "unbounded"capacity: "unbounded"
} | {
readonly capacity: numbercapacity: number
readonly strategy?: | "suspend"
| "sliding"
| "dropping"
| undefined
function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>R>(self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>R>) => import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<import QueueQueue.type Queue.Dequeue = /*unresolved*/ anyDequeue<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>E | import CauseCause.type Cause.Done = /*unresolved*/ anyDone>, never, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>R | import ScopeScope.type Scope.Scope = /*unresolved*/ anyScope>
<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
options: | {
readonly capacity: "unbounded"
}
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
}
capacity: "unbounded"capacity: "unbounded"
} | {
readonly capacity: numbercapacity: number
readonly strategy?: | "suspend"
| "sliding"
| "dropping"
| undefined
import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<import QueueQueue.type Queue.Dequeue = /*unresolved*/ anyDequeue<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
import CauseCause.type Cause.Done = /*unresolved*/ anyDone>, never, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
import ScopeScope.type Scope.Scope = /*unresolved*/ anyScope>
} = import dualdual(
2,
<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
options: | {
readonly capacity: "unbounded"
}
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
}
capacity: "unbounded"capacity: "unbounded"
} | {
readonly capacity: numbercapacity: number
readonly strategy?: | "suspend"
| "sliding"
| "dropping"
| undefined
import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<import QueueQueue.type Queue.Dequeue = /*unresolved*/ anyDequeue<function (type parameter) A in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
function (type parameter) E in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
import CauseCause.type Cause.Done = /*unresolved*/ anyDone>, never, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>, options: {
readonly capacity: "unbounded";
} | {
readonly capacity: number;
readonly strategy?: "dropping" | "sliding" | "suspend" | undefined;
}): Effect.Effect<Queue.Dequeue<A, E | Cause.Done>, never, R | Scope.Scope>
import ScopeScope.type Scope.Scope = /*unresolved*/ anyScope> =>
import ChannelChannel.toQueueArray(self: Stream<A, E, R>(parameter) self: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Stream<out A, out E = never, out R = never>.channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>(property) Stream<out A, out E = never, out R = never>.channel: {
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
options: | {
readonly capacity: "unbounded"
}
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
}