<A, R>(
stream: Stream.Stream<A, never, R>,
wake: Effect.Effect<void>
): Effect.Effect<void, never, R | Scope.Scope>Wire a stream to a cadence control: every element runs wake (or any control effect), so an
external fact — a queue add event, a store change — ends the polling wait IMMEDIATELY
instead of waiting out the interval. Forked into the current scope.
Pair with adaptive: point it at proc.polling.resetCadence and arrivals snap the
drainer back to its active cadence.
export const const wakeOn: <A, R>(
stream: Stream.Stream<A, never, R>,
wake: Effect.Effect<void>
) => Effect.Effect<void, never, R | Scope.Scope>
Wire a stream to a cadence control: every element runs wake (or any control effect), so an
external fact — a queue add event, a store change — ends the polling wait IMMEDIATELY
instead of waiting out the interval. Forked into the current scope.
Pair with
adaptive
: point it at proc.polling.resetCadence and arrivals snap the
drainer back to its active cadence.
wakeOn = <function (type parameter) A in <A, R>(stream: Stream.Stream<A, never, R>, wake: Effect.Effect<void>): Effect.Effect<void, never, R | Scope.Scope>A, function (type parameter) R in <A, R>(stream: Stream.Stream<A, never, R>, wake: Effect.Effect<void>): Effect.Effect<void, never, R | Scope.Scope>R>(
stream: Stream.Stream<A, never, R>(parameter) stream: {
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; <…;
}
import StreamStream.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, R>(stream: Stream.Stream<A, never, R>, wake: Effect.Effect<void>): Effect.Effect<void, never, R | Scope.Scope>A, never, function (type parameter) R in <A, R>(stream: Stream.Stream<A, never, R>, wake: Effect.Effect<void>): Effect.Effect<void, never, R | Scope.Scope>R>,
wake: Effect.Effect<void, never, never>(parameter) wake: {
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;
}
import EffectEffect.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<void>
): import EffectEffect.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<void, never, function (type parameter) R in <A, R>(stream: Stream.Stream<A, never, R>, wake: Effect.Effect<void>): Effect.Effect<void, never, R | Scope.Scope>R | import ScopeScope.Scope> =>
import EffectEffect.const asVoid: <Fiber<void, never>, never, R | Scope.Scope>(self: Effect.Effect<Fiber<void, never>, never, R | Scope.Scope>) => Effect.Effect<void, never, R | Scope.Scope>Maps the success value of an Effect to void, preserving failures.
Example (Discarding success values)
import { Effect } from "effect"
const program = Effect.asVoid(Effect.succeed(42))
Effect.runPromise(program).then(console.log)
// undefined (void)
asVoid(import EffectEffect.const forkScoped: <Effect.Effect<void, never, R>>(effectOrOptions?: Effect.Effect<void, never, R> | undefined, options?: {
readonly startImmediately?: boolean | undefined;
readonly uninterruptible?: boolean | "inherit" | undefined;
} | undefined) => Effect.Effect<Fiber<void, never>, never, R | Scope.Scope>
Forks the fiber in a Scope, interrupting it when the scope is closed.
Example (Forking into the current scope)
import { Effect } from "effect"
const backgroundTask = Effect.gen(function*() {
yield* Effect.sleep("5 seconds")
yield* Effect.log("Background task completed")
return "result"
})
const program = Effect.scoped(
Effect.gen(function*() {
const fiber = yield* backgroundTask.pipe(Effect.forkScoped)
// or fork a fiber that starts immediately:
yield* backgroundTask.pipe(Effect.forkScoped({ startImmediately: true }))
yield* Effect.log("Task forked in scope")
yield* Effect.sleep("1 second")
// Fiber will be interrupted when scope closes
return "scope completed"
})
)
forkScoped(import StreamStream.const runForEach: <A, never, R, void, never, never>(self: Stream.Stream<A, never, R>, f: (a: A) => Effect.Effect<void, never, never>) => Effect.Effect<void, never, R> (+1 overload)Runs the provided effectful callback for each element of the stream.
Example (Running an effect for each value)
import { Console, Effect, Stream } from "effect"
const stream = Stream.make(1, 2, 3)
const program = Effect.gen(function*() {
yield* Stream.runForEach(stream, (n) => Console.log(`Processing: ${n}`))
})
Effect.runPromise(program)
// Processing: 1
// Processing: 2
// Processing: 3
runForEach(stream: Stream.Stream<A, never, R>(parameter) stream: {
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; <…;
}
wake: Effect.Effect<void, never, never>(parameter) wake: {
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;
}