<A, E>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}): Stream<A, E>Creates a stream from a lazily supplied Web ReadableStream.
Details
The stream reads from a ReadableStreamDefaultReader, maps read failures
with onError, and closes the reader when the stream finalizes. By default
the reader is canceled; set releaseLockOnEnd to release the lock instead.
Example (Creating a stream from a ReadableStream)
import { Console, 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)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3 ]export const const fromReadableStream: <
A,
E
>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}) => Stream<A, E>
Creates a stream from a lazily supplied Web ReadableStream.
Details
The stream reads from a ReadableStreamDefaultReader, maps read failures
with onError, and closes the reader when the stream finalizes. By default
the reader is canceled; set releaseLockOnEnd to release the lock instead.
Example (Creating a stream from a ReadableStream)
import { Console, 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)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3 ]
fromReadableStream = <function (type parameter) A in <A, E>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}): Stream<A, E>
function (type parameter) E in <A, E>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}): Stream<A, E>
options: {
readonly evaluate: LazyArg<ReadableStream<A>>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}
evaluate: LazyArg<ReadableStream<A>>evaluate: type LazyArg<A> = () => AA zero-argument function that produces a value when invoked.
When to use
Use to type a lazy value provider that should not run until called.
Example (Creating a lazy argument)
import { Function } from "effect"
const constNull: Function.LazyArg<null> = Function.constant(null)
LazyArg<interface ReadableStream<R = any>The ReadableStream interface of the Streams API represents a readable stream of byte data.
ReadableStream<function (type parameter) A in <A, E>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}): Stream<A, E>
onError: (error: unknown) => EonError: (error: unknownerror: unknown) => function (type parameter) E in <A, E>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}): Stream<A, E>
releaseLockOnEnd?: boolean | undefinedreleaseLockOnEnd?: boolean | undefined
}
): 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>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}): Stream<A, E>
function (type parameter) E in <A, E>(options: {
readonly evaluate: LazyArg<ReadableStream<A>>;
readonly onError: (error: unknown) => E;
readonly releaseLockOnEnd?: boolean | undefined;
}): Stream<A, E>
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(import ChannelChannel.const fromTransform: <
OutElem,
OutErr,
OutDone,
InElem,
InErr,
InDone,
EX,
EnvX,
Env
>(
transform: (
upstream: Pull.Pull<InElem, InErr, InDone>,
scope: Scope.Scope
) => Effect.Effect<
Pull.Pull<OutElem, OutErr, OutDone, EnvX>,
EX,
Env
>
) => Channel<
OutElem,
Pull.ExcludeDone<OutErr> | EX,
OutDone,
InElem,
InErr,
InDone,
Env | EnvX
>
Creates a Channel from a transformation function that operates on upstream pulls.
Example (Creating channels from transforms)
import { Channel, Effect } from "effect"
const channel = Channel.fromTransform((upstream, scope) =>
Effect.succeed(upstream)
)
fromTransform(import EffectEffect.const fnUntraced: <Effect.Effect<void, never, never>, Effect.Effect<[A, ...A[]], Cause.Done<void> | E, never>, [_: Pull.Pull<unknown, unknown, unknown, never>, scope: Scope.Scope]>(body: (this: unassigned, _: Pull.Pull<unknown, unknown, unknown, never>, scope: Scope.Scope) => Generator<Effect.Effect<void, never, never>, Effect.Effect<[A, ...A[]], Cause.Done<void> | E, never>, never>) => (_: Pull.Pull<unknown, unknown, unknown, never>, scope: Scope.Scope) => Effect.Effect<Effect.Effect<[A, ...A[]], Cause.Done<void> | E, never>, never, never> (+41 overloads)fnUntraced(function*(_: Pull.Pull<
unknown,
unknown,
unknown,
never
>
(parameter) _: {
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; <…;
toString: () => string;
toJSON: () => unknown;
}
scope: Scope.Scope(parameter) scope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
const reader: ReadableStreamDefaultReader<A>reader = options: {
readonly evaluate: LazyArg<ReadableStream<A>>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}
evaluate: LazyArg<ReadableStream<A>>evaluate().ReadableStream<A>.getReader(): ReadableStreamDefaultReader<A> (+2 overloads)The getReader() method of the ReadableStream interface creates a reader and locks the stream to it.
getReader()
yield* import ScopeScope.const addFinalizer: (
scope: Scope,
finalizer: Effect<unknown>
) => Effect<void>
Registers a finalizer effect on a scope.
Details
If the scope is open, the finalizer runs when the scope closes, regardless of
whether the scope closes successfully or with an error. If the scope is
already closed, the finalizer runs immediately.
Example (Adding finalizers)
import { Console, Effect, Exit, Scope } from "effect"
const program = Effect.gen(function*() {
const scope = yield* Scope.make()
// Add simple finalizers
yield* Scope.addFinalizer(scope, Console.log("Cleanup task 1"))
yield* Scope.addFinalizer(scope, Console.log("Cleanup task 2"))
yield* Scope.addFinalizer(scope, Effect.log("Cleanup task 3"))
// Do some work
yield* Console.log("Doing work...")
// Close the scope
yield* Scope.close(scope, Exit.void)
})
addFinalizer(
scope: Scope.Scope(parameter) scope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
options: {
readonly evaluate: LazyArg<ReadableStream<A>>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}
releaseLockOnEnd?: boolean | undefinedreleaseLockOnEnd
? import EffectEffect.const sync: <A>(
thunk: LazyArg<A>
) => Effect<A>
Creates an Effect that represents a synchronous side-effectful computation.
When to use
Use when you need to wrap a synchronous side-effectful operation that is not
expected to throw.
Details
The provided function is evaluated lazily when the effect runs.
Gotchas
The function must not throw. If it throws, the thrown value is treated as a
defect, not as a typed failure. Use try when throwing is expected.
Example (Capturing synchronous logging in an Effect)
import { Effect } from "effect"
const log = (message: string) =>
Effect.sync(() => {
console.log(message) // side effect
})
// ┌─── Effect<void, never, never>
// ▼
const program = log("Hello, World!")
sync(() => const reader: ReadableStreamDefaultReader<A>reader.ReadableStreamDefaultReader<A>.releaseLock(): voidThe releaseLock() method of the ReadableStreamDefaultReader interface releases the reader's lock on the stream.
releaseLock())
: import EffectEffect.const promise: <A>(
evaluate: (
signal: AbortSignal
) => PromiseLike<A>
) => Effect<A>
Creates an Effect that represents an asynchronous computation guaranteed to
succeed.
When to use
Use to convert a Promise into an Effect when the async operation is
guaranteed to succeed and will not reject.
Details
An optional AbortSignal can be provided to allow for interruption of the
wrapped Promise API.
Gotchas
The Promise must not reject. If it rejects, the rejection is treated as a
defect, not as a typed failure. Use tryPromise when rejection is expected.
Interruption aborts the provided AbortSignal, but the underlying
asynchronous operation only stops if it observes that signal.
Example (Wrapping a non-rejecting Promise)
import { Effect } from "effect"
const delay = (message: string) =>
Effect.promise<string>(
() =>
new Promise((resolve) => {
setTimeout(() => {
resolve(message)
}, 2000)
})
)
// ┌─── Effect<string, never, never>
// ▼
const program = delay("Async operation completed successfully!")
promise(() => const reader: ReadableStreamDefaultReader<A>reader.ReadableStreamGenericReader.cancel(reason?: any): Promise<void>cancel().Promise<void>.catch<void>(onrejected?: ((reason: any) => void | PromiseLike<void>) | null | undefined): Promise<void>Attaches a callback for only the rejection of the Promise.
catch(const constVoid: LazyArg<void>Returns no meaningful value when called.
When to use
Use when you need a thunk that is called only for its effect and has no
meaningful return value.
Example (Returning void from a thunk)
import { Function } from "effect"
import * as assert from "node:assert"
assert.deepStrictEqual(Function.constVoid(), undefined)
constVoid))
)
return import EffectEffect.const flatMap: {
<A, B, E1, R1>(
f: (a: A) => Effect<B, E1, R1>
): <E, R>(
self: Effect<A, E, R>
) => Effect<B, E1 | E, R1 | R>
<A, E, R, B, E1, R1>(
self: Effect<A, E, R>,
f: (a: A) => Effect<B, E1, R1>
): Effect<B, E | E1, R | R1>
}
import EffectEffect.const tryPromise: <
A,
E = Cause.UnknownError
>(
options:
| {
readonly try: (
signal: AbortSignal
) => PromiseLike<A>
readonly catch: (error: unknown) => E
}
| ((signal: AbortSignal) => PromiseLike<A>)
) => Effect<A, E>
Creates an Effect from an asynchronous computation that may throw or
reject, mapping failures into the error channel.
When to use
Use when you need to perform asynchronous operations that might fail, such
as fetching data from an API, and want thrown exceptions or rejected promises
captured as Effect errors.
Details
The promise thunk is evaluated when the effect runs. If it returns a promise
that resolves, the resolved value becomes the success value. If the thunk
throws before returning a promise, or if the returned promise rejects, the
thrown or rejected value is mapped into the error channel.
Passing the thunk directly maps failures to
Cause.UnknownError
.
Passing { try, catch } uses catch to map failures to an error of type
E.
The thunk receives an AbortSignal that is aborted if the effect is
interrupted. The underlying asynchronous operation only stops if it observes
that signal.
Gotchas
If catch throws while mapping the error, that thrown value is treated as a
defect. Return the error value you want in the error channel instead of
throwing it.
Example (Wrapping a fetch request that may fail)
import { Effect } from "effect"
const getTodo = (id: number) =>
// Will catch any errors and propagate them as UnknownError
Effect.tryPromise(() =>
fetch(`https://jsonplaceholder.typicode.com/todos/${id}`)
)
// ┌─── Effect<Response, UnknownError, never>
// ▼
const program = getTodo(1)
Example (Mapping Promise rejections to a tagged error)
import { Data, Effect } from "effect"
class TodoFetchError extends Data.TaggedError("TodoFetchError")<{ readonly cause: unknown }> {}
const getTodo = (id: number) =>
Effect.tryPromise({
try: () => fetch(`https://jsonplaceholder.typicode.com/todos/${id}`),
// remap the error
catch: (cause) => new TodoFetchError({ cause })
})
// ┌─── Effect<Response, TodoFetchError, never>
// ▼
const program = getTodo(1)
tryPromise({
try: (
signal: AbortSignal
) => PromiseLike<ReadableStreamReadResult<A>>
const reader: ReadableStreamDefaultReader<A>reader.ReadableStreamDefaultReader<A>.read(): Promise<ReadableStreamReadResult<A>>The read() method of the ReadableStreamDefaultReader interface returns a Promise providing access to the next chunk in the stream's internal queue.
read(),
catch: (error: unknown) => Ecatch: (reason: unknownreason) => options: {
readonly evaluate: LazyArg<ReadableStream<A>>
readonly onError: (error: unknown) => E
readonly releaseLockOnEnd?: boolean | undefined
}
onError: (error: unknown) => EonError(reason: unknownreason)
}),
({ done: booleandone, value: A | undefinedvalue }) => done: booleandone ? import CauseCause.const done: <A = void>(
value?: A
) => Effect.Effect<never, Done<A>>
Creates an Effect that fails with a Done error. Shorthand for
Effect.fail(Cause.Done(value)).
When to use
Use when you model stream or queue completion through the error channel.
Example (Failing with Done)
import { Cause, Effect } from "effect"
const program = Cause.done("finished")
Effect.runPromiseExit(program).then((exit) => {
console.log(exit._tag) // "Failure"
})
done() : import EffectEffect.const succeed: <A>(value: A) => Effect<A>Creates an Effect that always succeeds with a given value.
When to use
Use when an effect should complete successfully with a specific value without any errors
or external dependencies.
Example (Creating a successful effect)
import { Effect } from "effect"
// Creating an effect that represents a successful scenario
//
// ┌─── Effect<number, never, never>
// ▼
const success = Effect.succeed(42)
succeed(import ArrArr.const of: <A>(a: A) => NonEmptyArray<A>Wraps a single value in a NonEmptyArray.
Example (Creating a single-element array)
import { Array } from "effect"
console.log(Array.of(1)) // [1]
of(value: Avalue))
)
})))