<const N extends number>(
options:
| {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined
readonly replay?: number | undefined
}
): <A, E, R>(
self: Stream<A, E, R>
) => Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
<A, E, R, const N extends number>(
self: Stream<A, E, R>,
options:
| {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly capacity: number
readonly n: N
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined
readonly replay?: number | undefined
}
): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>Creates a fixed-size tuple of streams that each emit the same elements as the source stream.
Details
The source stream starts after all downstream streams have been subscribed.
With the default suspend strategy, the source can only advance capacity
chunks ahead of the slowest downstream stream. If a downstream stream is
interrupted, it unsubscribes from the broadcast so it no longer contributes
backpressure.
Example (Broadcasting to two consumers)
import { Console, Effect, Stream } from "effect"
const program = Effect.scoped(
Effect.gen(function*() {
const [left, right] = yield* Stream.make(1, 2, 3).pipe(
Stream.broadcastN({ n: 2, capacity: 8 })
)
const values = yield* Effect.all([
Stream.runCollect(left),
Stream.runCollect(right)
], { concurrency: "unbounded" })
yield* Console.log(values)
})
)
Effect.runPromise(program)
// Output: [[1, 2, 3], [1, 2, 3]]Broadcast
Source effect/Stream.ts:878161 lines
export const const broadcastN: {
<N extends number>(
options:
| {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
): <A, E, R>(
self: Stream<A, E, R>
) => Effect.Effect<
TupleOf<N, Stream<A, E>>,
never,
Scope.Scope | R
>
<A, E, R, N extends number>(
self: Stream<A, E, R>,
options:
| {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly capacity: number
readonly n: N
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
): Effect.Effect<
TupleOf<N, Stream<A, E>>,
never,
Scope.Scope | R
>
}
Creates a fixed-size tuple of streams that each emit
the same elements as the source stream.
Details
The source stream starts after all downstream streams have been subscribed.
With the default suspend strategy, the source can only advance capacity
chunks ahead of the slowest downstream stream. If a downstream stream is
interrupted, it unsubscribes from the broadcast so it no longer contributes
backpressure.
Example (Broadcasting to two consumers)
import { Console, Effect, Stream } from "effect"
const program = Effect.scoped(
Effect.gen(function*() {
const [left, right] = yield* Stream.make(1, 2, 3).pipe(
Stream.broadcastN({ n: 2, capacity: 8 })
)
const values = yield* Effect.all([
Stream.runCollect(left),
Stream.runCollect(right)
], { concurrency: "unbounded" })
yield* Console.log(values)
})
)
Effect.runPromise(program)
// Output: [[1, 2, 3], [1, 2, 3]]
broadcastN: {
<const function (type parameter) N in <const N extends number>(options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): <A, E, R>(self: Stream<A, E, R>) => Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
options: | {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
n: const N extends numbern: function (type parameter) N in <const N extends number>(options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): <A, E, R>(self: Stream<A, E, R>) => Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
capacity: "unbounded"capacity: "unbounded"
readonly replay?: number | undefinedreplay?: number | undefined
} | {
readonly n: const N extends numbern: function (type parameter) N in <const N extends number>(options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): <A, E, R>(self: Stream<A, E, R>) => Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
capacity: numbercapacity: number
readonly strategy?: | "suspend"
| "sliding"
| "dropping"
| undefined
replay?: number | undefinedreplay?: number | undefined
}
): <function (type parameter) A in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>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<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>E, function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>R>) => import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<type TupleOf<N extends number, T> = N extends N ? number extends N ? T[] : TupleOf_<T, N, []> : neverConstructs a tuple type with exactly N elements of type T.
When to use
Use when you need a fixed-length array type, especially instead of manually
writing [T, T, T, ...] for longer tuples.
Details
- If
N is a literal number, produces a tuple of that exact length.
- If
N is the general number type (non-literal), degrades to Array<T>.
- Negative numbers produce
never.
Example (Checking fixed-length tuples)
import type { Types } from "effect"
// Exactly 3 numbers
const triple: Types.TupleOf<3, number> = [1, 2, 3]
//
TupleOf<function (type parameter) N in <const N extends number>(options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): <A, E, R>(self: Stream<A, E, R>) => Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
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<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>A, function (type parameter) E in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>E>>, never, import ScopeScope.type Scope.Scope = /*unresolved*/ anyScope | function (type parameter) R in <A, E, R>(self: Stream<A, E, R>): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>R>
<function (type parameter) A in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
function (type parameter) R in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, 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, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
function (type parameter) R in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
options: | {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly capacity: number
readonly n: N
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
n: const N extends numbern: function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
capacity: "unbounded"capacity: "unbounded"
readonly replay?: number | undefinedreplay?: number | undefined
} | {
readonly capacity: numbercapacity: number
readonly n: const N extends numbern: function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
strategy?: | "suspend"
| "sliding"
| "dropping"
| undefined
replay?: number | undefinedreplay?: number | undefined
}
): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<type TupleOf<N extends number, T> = N extends N ? number extends N ? T[] : TupleOf_<T, N, []> : neverConstructs a tuple type with exactly N elements of type T.
When to use
Use when you need a fixed-length array type, especially instead of manually
writing [T, T, T, ...] for longer tuples.
Details
- If
N is a literal number, produces a tuple of that exact length.
- If
N is the general number type (non-literal), degrades to Array<T>.
- Negative numbers produce
never.
Example (Checking fixed-length tuples)
import type { Types } from "effect"
// Exactly 3 numbers
const triple: Types.TupleOf<3, number> = [1, 2, 3]
//
TupleOf<function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
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, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
import ScopeScope.type Scope.Scope = /*unresolved*/ anyScope | function (type parameter) R in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly capacity: number;
readonly n: N;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Effect.Effect<TupleOf<N, Stream<A, E>>, never, Scope.Scope | R>
import dualdual(
2,
import EffectEffect.fnUntraced(function*<function (type parameter) A in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) R in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
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, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) R in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
options: | {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
n: const N extends numbern: function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
capacity: "unbounded"capacity: "unbounded"
readonly replay?: number | undefinedreplay?: number | undefined
} | {
readonly n: const N extends numbern: function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
capacity: numbercapacity: number
readonly strategy?: | "suspend"
| "sliding"
| "dropping"
| undefined
replay?: number | undefinedreplay?: number | undefined
}
) {
const const pubsub: PubSub.PubSub<
Take.Take<A, E, void>
>
const pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
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; <…;
}
const makePubSub: <any>(
options:
| {
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly capacity: number
readonly strategy?:
| "dropping"
| "sliding"
| "suspend"
| undefined
readonly replay?: number | undefined
}
) => any
import TakeTake.type Take<A, E = never, Done = void> = anyRepresents one pull result: either a non-empty batch of values, a failure
Exit, or a successful Exit that signals completion with a Done value.
When to use
Use to store, transfer, or interpret pull results later while preserving
emitted values, failures, and normal completion.
Take<function (type parameter) A in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
options: | {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
const streams: any[]streams = new var Array: ArrayConstructor
new (arrayLength?: number) => any[] (+2 overloads)
options: | {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
n: const N extends numbern)
const const parentScope: Scope.Scopeconst parentScope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
import ScopeScope.Scope
for (let let i: numberi = 0; let i: numberi < options: | {
readonly n: N
readonly capacity: "unbounded"
readonly replay?: number | undefined
}
| {
readonly n: N
readonly capacity: number
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
readonly replay?: number | undefined
}
n: const N extends numbern; let i: numberi++) {
const const scope: Scope.Closeableconst scope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
import ScopeScope.forkUnsafe(const parentScope: Scope.Scopeconst parentScope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
const subscription: PubSub.Subscription<
Take.Take<A, E, void>
>
const subscription: {
pubsub: PubSub.Atomic<any>;
subscribers: PubSub.Subscribers<any>;
subscription: PubSub.BackingSubscription<A>;
pollers: MutableList.MutableList<Deferred.Deferred<any>>;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<any>;
replayWindow: PubSub.ReplayWindow<A>;
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; <…;
}
import PubSubPubSub.subscribe(const pubsub: PubSub.PubSub<
Take.Take<A, E, void>
>
const pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
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; <…;
}
import EffectEffect.provideService(import ScopeScope.Scope, const scope: Scope.Closeableconst scope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
const streams: any[]streams[let i: numberi] = import ChannelChannel.fromEffectTake(import PubSubPubSub.take(const subscription: PubSub.Subscription<
Take.Take<A, E, void>
>
const subscription: {
pubsub: PubSub.Atomic<any>;
subscribers: PubSub.Subscribers<any>;
subscription: PubSub.BackingSubscription<A>;
pollers: MutableList.MutableList<Deferred.Deferred<any>>;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<any>;
replayWindow: PubSub.ReplayWindow<A>;
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; <…;
}
import ChannelChannel.onExit((exit: Exit.Exit<void, E>exit) => import ScopeScope.close(const scope: Scope.Closeableconst scope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
exit: Exit.Exit<void, E>exit)),
const fromChannel: <
Arr extends Arr.NonEmptyReadonlyArray<any>,
E,
R
>(
channel: Channel.Channel<
Arr,
E,
void,
unknown,
unknown,
unknown,
R
>
) => Stream<
Arr extends Arr.NonEmptyReadonlyArray<infer A>
? A
: never,
E,
R
>
Creates a stream from a array-emitting Channel.
Example (Creating a stream from an array-emitting channel)
import { Channel, Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const channel = Channel.succeed([1, 2, 3] as const)
const stream = Stream.fromChannel(channel)
const result = yield* Stream.runCollect(stream)
yield* Console.log(result)
})
// Output: [ 1, 2, 3 ]
fromChannel
)
}
yield* import ChannelChannel.runForEach(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; <…;
}
value: any(parameter) value: {
0: A;
length: number;
toString: () => string;
toLocaleString: { (): string; (locales: string | string[], options?: Intl.NumberFormatOptions & Intl.DateTimeFormatOptions): string };
concat: { (...items: Array<ConcatArray<A>>): Array<A>; (...items: Array<A | ConcatArray<A>>): Array<A> };
join: (separator?: string) => string;
slice: (start?: number, end?: number) => Array<A>;
indexOf: (searchElement: A, fromIndex?: number) => number;
lastIndexOf: (searchElement: A, fromIndex?: number) => number;
every: { (predicate: (value: A, index: number, array: ReadonlyArray<A>) => value is S, thisArg?: any): this is readonly S[]; (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any): boolean };
some: (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any) => boolean;
forEach: (callbackfn: (value: A, index: number, array: ReadonlyArray<A>) => void, thisArg?: any) => void;
map: (callbackfn: (value: A, index: number, array: ReadonlyArray<A>) => U, thisArg?: any) => Array<U>;
filter: { (predicate: (value: A, index: number, array: ReadonlyArray<A>) => value is S, thisArg?: any): Array<S>; (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any): Array<A> };
reduce: { (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A): A; (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A, initialValue: A): A; (callbac…;
reduceRight: { (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A): A; (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A, initialValue: A): A; (callbac…;
find: { (predicate: (value: A, index: number, obj: ReadonlyArray<A>) => value is S, thisArg?: any): S | undefined; (predicate: (value: A, index: number, obj: ReadonlyArray<A>) => unknown, thisArg?: any): A | undefined };
findIndex: (predicate: (value: A, index: number, obj: ReadonlyArray<A>) => unknown, thisArg?: any) => number;
entries: () => ArrayIterator<[number, A]>;
keys: () => ArrayIterator<number>;
values: () => ArrayIterator<A>;
includes: (searchElement: A, fromIndex?: number) => boolean;
flatMap: (callback: (this: This, value: A, index: number, array: Array<A>) => U | ReadonlyArray<U>, thisArg?: This | undefined) => Array<U>;
flat: (this: A, depth?: D | undefined) => Array<FlatArray<A, D>>;
at: (index: number) => A | undefined;
findLast: { (predicate: (value: A, index: number, array: ReadonlyArray<A>) => value is S, thisArg?: any): S | undefined; (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any): A | undefined };
findLastIndex: (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any) => number;
toReversed: () => Array<A>;
toSorted: (compareFn?: ((a: A, b: A) => number) | undefined) => Array<A>;
toSpliced: { (start: number, deleteCount: number, ...items: Array<A>): Array<A>; (start: number, deleteCount?: number): Array<A> };
with: (index: number, value: A) => Array<A>;
}
import PubSubPubSub.publish(const pubsub: PubSub.PubSub<
Take.Take<A, E, void>
>
const pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
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; <…;
}
value: any(parameter) value: {
0: A;
length: number;
toString: () => string;
toLocaleString: { (): string; (locales: string | string[], options?: Intl.NumberFormatOptions & Intl.DateTimeFormatOptions): string };
concat: { (...items: Array<ConcatArray<A>>): Array<A>; (...items: Array<A | ConcatArray<A>>): Array<A> };
join: (separator?: string) => string;
slice: (start?: number, end?: number) => Array<A>;
indexOf: (searchElement: A, fromIndex?: number) => number;
lastIndexOf: (searchElement: A, fromIndex?: number) => number;
every: { (predicate: (value: A, index: number, array: ReadonlyArray<A>) => value is S, thisArg?: any): this is readonly S[]; (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any): boolean };
some: (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any) => boolean;
forEach: (callbackfn: (value: A, index: number, array: ReadonlyArray<A>) => void, thisArg?: any) => void;
map: (callbackfn: (value: A, index: number, array: ReadonlyArray<A>) => U, thisArg?: any) => Array<U>;
filter: { (predicate: (value: A, index: number, array: ReadonlyArray<A>) => value is S, thisArg?: any): Array<S>; (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any): Array<A> };
reduce: { (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A): A; (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A, initialValue: A): A; (callbac…;
reduceRight: { (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A): A; (callbackfn: (previousValue: A, currentValue: A, currentIndex: number, array: ReadonlyArray<A>) => A, initialValue: A): A; (callbac…;
find: { (predicate: (value: A, index: number, obj: ReadonlyArray<A>) => value is S, thisArg?: any): S | undefined; (predicate: (value: A, index: number, obj: ReadonlyArray<A>) => unknown, thisArg?: any): A | undefined };
findIndex: (predicate: (value: A, index: number, obj: ReadonlyArray<A>) => unknown, thisArg?: any) => number;
entries: () => ArrayIterator<[number, A]>;
keys: () => ArrayIterator<number>;
values: () => ArrayIterator<A>;
includes: (searchElement: A, fromIndex?: number) => boolean;
flatMap: (callback: (this: This, value: A, index: number, array: Array<A>) => U | ReadonlyArray<U>, thisArg?: This | undefined) => Array<U>;
flat: (this: A, depth?: D | undefined) => Array<FlatArray<A, D>>;
at: (index: number) => A | undefined;
findLast: { (predicate: (value: A, index: number, array: ReadonlyArray<A>) => value is S, thisArg?: any): S | undefined; (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any): A | undefined };
findLastIndex: (predicate: (value: A, index: number, array: ReadonlyArray<A>) => unknown, thisArg?: any) => number;
toReversed: () => Array<A>;
toSorted: (compareFn?: ((a: A, b: A) => number) | undefined) => Array<A>;
toSpliced: { (start: number, deleteCount: number, ...items: Array<A>): Array<A>; (start: number, deleteCount?: number): Array<A> };
with: (index: number, value: A) => Array<A>;
}
import EffectEffect.onExit((exit: Exit.Exit<void, E>exit) => import PubSubPubSub.publish(const pubsub: PubSub.PubSub<
Take.Take<A, E, void>
>
const pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
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; <…;
}
exit: Exit.Exit<void, E>exit)),
import EffectEffect.forkScoped
)
return const streams: any[]streams as type TupleOf<N extends number, T> = N extends N ? number extends N ? T[] : TupleOf_<T, N, []> : neverConstructs a tuple type with exactly N elements of type T.
When to use
Use when you need a fixed-length array type, especially instead of manually
writing [T, T, T, ...] for longer tuples.
Details
- If
N is a literal number, produces a tuple of that exact length.
- If
N is the general number type (non-literal), degrades to Array<T>.
- Negative numbers produce
never.
Example (Checking fixed-length tuples)
import type { Types } from "effect"
// Exactly 3 numbers
const triple: Types.TupleOf<3, number> = [1, 2, 3]
//
TupleOf<function (type parameter) N in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
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, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>
function (type parameter) E in <A, E, R, const N extends number>(self: Stream<A, E, R>, options: {
readonly n: N;
readonly capacity: "unbounded";
readonly replay?: number | undefined;
} | {
readonly n: N;
readonly capacity: number;
readonly strategy?: "sliding" | "dropping" | "suspend" | undefined;
readonly replay?: number | undefined;
}): Generator<...>