<A = unknown>(
target: EventListener<A>,
type: string,
options?:
| boolean
| {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
| undefined
): Stream<A>Creates a stream from an event listener.
Example (Creating a stream from an event listener)
import { Effect, Stream } from "effect"
class NumberTarget implements Stream.EventListener<number> {
addEventListener(event: string, f: (event: number) => void) {
if (event === "data") {
f(1)
f(2)
f(3)
}
}
removeEventListener(_event: string, _f: (event: number) => void) {}
}
Effect.runPromise(Effect.gen(function*() {
const stream = Stream.fromEventListener(new NumberTarget(), "data").pipe(
Stream.take(3)
)
const values = yield* Stream.runCollect(stream)
yield* Effect.sync(() => console.log(values))
}))
// [ 1, 2, 3 ]export const const fromEventListener: <A = unknown>(
target: EventListener<A>,
type: string,
options?:
| boolean
| {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
| undefined
) => Stream<A>
Creates a stream from an event listener.
Example (Creating a stream from an event listener)
import { Effect, Stream } from "effect"
class NumberTarget implements Stream.EventListener<number> {
addEventListener(event: string, f: (event: number) => void) {
if (event === "data") {
f(1)
f(2)
f(3)
}
}
removeEventListener(_event: string, _f: (event: number) => void) {}
}
Effect.runPromise(Effect.gen(function*() {
const stream = Stream.fromEventListener(new NumberTarget(), "data").pipe(
Stream.take(3)
)
const values = yield* Stream.runCollect(stream)
yield* Effect.sync(() => console.log(values))
}))
// [ 1, 2, 3 ]
fromEventListener = <function (type parameter) A in <A = unknown>(target: EventListener<A>, type: string, options?: boolean | {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly bufferSize?: number | undefined;
} | undefined): Stream<A>
target: EventListener<A>(parameter) target: {
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;
}
interface EventListener<A>Interface representing an event listener target.
EventListener<function (type parameter) A in <A = unknown>(target: EventListener<A>, type: string, options?: boolean | {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly bufferSize?: number | undefined;
} | undefined): Stream<A>
type: stringtype: string,
options: | boolean
| {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
| undefined
capture?: boolean | undefinedcapture?: boolean
readonly passive?: boolean | undefinedpassive?: boolean
readonly once?: boolean | undefinedonce?: boolean
readonly bufferSize?: number | undefinedbufferSize?: number | undefined
} | 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 = unknown>(target: EventListener<A>, type: string, options?: boolean | {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly bufferSize?: number | undefined;
} | undefined): Stream<A>
const callback: <A, E = never, R = never>(
f: (
queue: Queue.Queue<A, E | Cause.Done>
) => Effect.Effect<unknown, E, R | Scope.Scope>,
options?: {
readonly bufferSize?: number | undefined
readonly strategy?:
| "sliding"
| "dropping"
| "suspend"
| undefined
}
) => Stream<A, E, Exclude<R, Scope.Scope>>
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)
import { Console, Effect, Queue, Stream } from "effect"
const stream = Stream.callback<number>((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)
})
)
const program = Effect.gen(function*() {
const values = yield* stream.pipe(Stream.runCollect)
yield* Console.log(values)
// [ 1, 2, 3 ]
})
Effect.runPromise(program)
callback<function (type parameter) A in <A = unknown>(target: EventListener<A>, type: string, options?: boolean | {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly bufferSize?: number | undefined;
} | undefined): Stream<A>
queue: Queue.Queue<A, Cause.Done<void>>(parameter) queue: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
function (local function) emit(event: A): voidemit(event: A = unknownevent: function (type parameter) A in <A = unknown>(target: EventListener<A>, type: string, options?: boolean | {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly bufferSize?: number | undefined;
} | undefined): Stream<A>
import QueueQueue.const offerUnsafe: <A, E>(
self: Enqueue<A, E>,
message: Types.NoInfer<A>
) => boolean
Adds a message to the queue synchronously. Returns false if the queue is done.
When to use
Use when you are already in synchronous queue internals or a performance
boundary where wrapping the mutation in Effect is intentionally avoided.
Gotchas
This is an unsafe operation that directly modifies the queue without Effect wrapping.
Use this only when you're certain about the synchronous nature of the operation.
Example (Offering a value synchronously)
import { Cause, Effect, Queue } from "effect"
// Create a queue effect and extract the queue for unsafe operations
const program = Effect.gen(function*() {
const queue = yield* Queue.bounded<number>(3)
// Add messages synchronously using unsafe API
const success1 = Queue.offerUnsafe(queue, 1)
const success2 = Queue.offerUnsafe(queue, 2)
console.log(success1, success2) // true, true
// Check current size
const size = Queue.sizeUnsafe(queue)
console.log(size) // 2
})
offerUnsafe(queue: Queue.Queue<A, Cause.Done<void>>(parameter) queue: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
event: A = unknownevent)
}
return import EffectEffect.const acquireRelease: <A, E, R, R2>(
acquire: Effect<A, E, R>,
release: (
a: A,
exit: Exit.Exit<unknown, unknown>
) => Effect<unknown, never, R2>,
options?: { readonly interruptible?: boolean }
) => Effect<A, E, R | R2 | Scope>
Constructs a scoped resource from an acquisition effect and a release
finalizer.
When to use
Use to acquire a scoped resource with an explicit release finalizer.
Details
If acquisition succeeds, the release finalizer is added to the current scope
and is guaranteed to run when that scope closes. The finalizer receives the
Exit value used to close the scope.
By default, acquisition is protected by an uninterruptible region. Pass
{ interruptible: true } to allow the acquisition effect to be interrupted.
Example (Acquiring and releasing a resource)
import { Console, Effect, Exit } from "effect"
// Simulate a resource that needs cleanup
interface FileHandle {
readonly path: string
readonly content: string
}
// Acquire a file handle
const acquire = Effect.gen(function*() {
yield* Console.log("Opening file")
return { path: "/tmp/file.txt", content: "file content" }
})
// Release the file handle
const release = (handle: FileHandle, exit: Exit.Exit<unknown, unknown>) =>
Console.log(
`Closing file ${handle.path} with exit: ${
Exit.isSuccess(exit) ? "success" : "failure"
}`
)
// Create a scoped resource
const resource = Effect.acquireRelease(acquire, release)
// Use the resource within a scope
const program = Effect.scoped(
Effect.gen(function*() {
const handle = yield* resource
yield* Console.log(`Using file: ${handle.path}`)
return handle.content
})
)
acquireRelease(
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(() => target: EventListener<A>(parameter) target: {
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;
}
EventListener<A>.addEventListener(event: string, f: (event: A) => void, options?: {
readonly capture?: boolean;
readonly passive?: boolean;
readonly once?: boolean;
readonly signal?: AbortSignal;
} | boolean): void
type: stringtype, function (local function) emit(event: A): voidemit, options: | boolean
| {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
| undefined
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(() => target: EventListener<A>(parameter) target: {
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;
}
EventListener<A>.removeEventListener(event: string, f: (event: A) => void, options?: {
readonly capture?: boolean;
} | boolean): void
type: stringtype, function (local function) emit(event: A): voidemit, options: | boolean
| {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
| undefined
bufferSize?: number | undefinedbufferSize: typeof options: | boolean
| {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
| undefined
options: {
readonly capture?: boolean
readonly passive?: boolean
readonly once?: boolean
readonly bufferSize?: number | undefined
}
bufferSize?: number | undefinedbufferSize : var undefinedundefined })