<E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
<A, E, R, E2, A2>(
self: Stream<A, E, R>,
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
): Stream<A2, E2, R>Maps both the failure and success channels of a stream.
Example (Mapping both the failure and success channels of a stream)
import { Console, Effect, Stream } from "effect"
const mapper = {
onFailure: (error: string) => `error: ${error}`,
onSuccess: (value: number) => value * 2
}
const program = Effect.gen(function*() {
const success = yield* Stream.make(1, 2).pipe(
Stream.mapBoth(mapper),
Stream.runCollect
)
yield* Console.log(success)
const failure = yield* Stream.fail("boom").pipe(
Stream.mapBoth(mapper),
Stream.catch((error: string) => Stream.succeed(error)),
Stream.runCollect
)
yield* Console.log(failure)
})
Effect.runPromise(program)
// Output: [ 2, 4 ]
// Output: [ "error: boom" ]mapping
Source effect/Stream.ts:192616 lines
export const const mapBoth: {
<E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}): <R>(
self: Stream<A, E, R>
) => Stream<A2, E2, R>
<A, E, R, E2, A2>(
self: Stream<A, E, R>,
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
): Stream<A2, E2, R>
}
Maps both the failure and success channels of a stream.
Example (Mapping both the failure and success channels of a stream)
import { Console, Effect, Stream } from "effect"
const mapper = {
onFailure: (error: string) => `error: ${error}`,
onSuccess: (value: number) => value * 2
}
const program = Effect.gen(function*() {
const success = yield* Stream.make(1, 2).pipe(
Stream.mapBoth(mapper),
Stream.runCollect
)
yield* Console.log(success)
const failure = yield* Stream.fail("boom").pipe(
Stream.mapBoth(mapper),
Stream.catch((error: string) => Stream.succeed(error)),
Stream.runCollect
)
yield* Console.log(failure)
})
Effect.runPromise(program)
// Output: [ 2, 4 ]
// Output: [ "error: boom" ]
mapBoth: {
<function (type parameter) E in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) E2 in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) A in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) A2 in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
onFailure: (e: E) => E2onFailure: (e: Ee: function (type parameter) E in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) E2 in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
onSuccess: (a: A) => A2onSuccess: (a: Aa: function (type parameter) A in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) A2 in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) R in <R>(self: Stream<A, E, R>): Stream<A2, E2, 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 <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) E in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) R in <R>(self: Stream<A, E, R>): Stream<A2, E2, R>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) A2 in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) E2 in <E, E2, A, A2>(options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): <R>(self: Stream<A, E, R>) => Stream<A2, E2, R>
function (type parameter) R in <R>(self: Stream<A, E, R>): Stream<A2, E2, R>R>
<function (type parameter) A in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) R in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) A2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
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, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) R in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
onFailure: (e: E) => E2onFailure: (e: Ee: function (type parameter) E in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
onSuccess: (a: A) => A2onSuccess: (a: Aa: function (type parameter) A in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) A2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
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) A2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) R in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
import dualdual(2, <function (type parameter) A in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) R in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) A2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
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, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) R in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
onFailure: (e: E) => E2onFailure: (e: Ee: function (type parameter) E in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
onSuccess: (a: A) => A2onSuccess: (a: Aa: function (type parameter) A in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) A2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
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) A2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) E2 in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
function (type parameter) R in <A, E, R, E2, A2>(self: Stream<A, E, R>, options: {
readonly onFailure: (e: E) => E2;
readonly onSuccess: (a: A) => A2;
}): Stream<A2, E2, R>
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; <…;
}
const map: {
<A, B>(f: (a: A, i: number) => B): <E, R>(
self: Stream<A, E, R>
) => Stream<B, E, R>
<A, E, R, B>(
self: Stream<A, E, R>,
f: (a: A, i: number) => B
): Stream<B, E, R>
}
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
onSuccess: (a: A) => A2onSuccess),
const mapError: {
<E, E2>(f: (error: E) => E2): <A, R>(
self: Stream<A, E, R>
) => Stream<A, E2, R>
<A, E, R, E2>(
self: Stream<A, E, R>,
f: (error: E) => E2
): Stream<A, E2, R>
}
options: {
readonly onFailure: (e: E) => E2
readonly onSuccess: (a: A) => A2
}
onFailure: (e: E) => E2onFailure)
))