<A, E>(self: Enqueue<A, E>): Effect<boolean>Interrupts the queue gracefully, transitioning it to a closing state.
Details
This operation stops accepting new offers but allows existing messages to be consumed. Once all messages are drained, the queue transitions to the Done state with an interrupt cause.
Example (Interrupting queues gracefully)
import { Cause, Effect, Queue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* Queue.bounded<number>(10)
// Add some messages
yield* Queue.offer(queue, 1)
yield* Queue.offer(queue, 2)
// Interrupt gracefully - no more offers accepted, but messages can be consumed
const interrupted = yield* Queue.interrupt(queue)
console.log(interrupted) // true
// Trying to offer more messages will return false
const offerResult = yield* Queue.offer(queue, 3)
console.log(offerResult) // false
// But we can still take existing messages
const message1 = yield* Queue.take(queue)
console.log(message1) // 1
const message2 = yield* Queue.take(queue)
console.log(message2) // 2
// After all messages are consumed, queue is done
const isDone = queue.state._tag === "Done"
console.log(isDone) // true
})completion
Source effect/Queue.ts:10772 lines
export const const interrupt: <A, E>(
self: Enqueue<A, E>
) => Effect<boolean>
Interrupts the queue gracefully, transitioning it to a closing state.
Details
This operation stops accepting new offers but allows existing messages to be consumed.
Once all messages are drained, the queue transitions to the Done state with an interrupt cause.
Example (Interrupting queues gracefully)
import { Cause, Effect, Queue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* Queue.bounded<number>(10)
// Add some messages
yield* Queue.offer(queue, 1)
yield* Queue.offer(queue, 2)
// Interrupt gracefully - no more offers accepted, but messages can be consumed
const interrupted = yield* Queue.interrupt(queue)
console.log(interrupted) // true
// Trying to offer more messages will return false
const offerResult = yield* Queue.offer(queue, 3)
console.log(offerResult) // false
// But we can still take existing messages
const message1 = yield* Queue.take(queue)
console.log(message1) // 1
const message2 = yield* Queue.take(queue)
console.log(message2) // 2
// After all messages are consumed, queue is done
const isDone = queue.state._tag === "Done"
console.log(isDone) // true
})
interrupt = <function (type parameter) A in <A, E>(self: Enqueue<A, E>): Effect<boolean>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>): Effect<boolean>E>(self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
interface Enqueue<in A, in E = never>An Enqueue is a queue that can be offered to.
Details
This interface represents the write-only part of a Queue, allowing you to offer
elements to the queue but not take elements from it.
Example (Offering through enqueue handles)
import { Effect, Queue } from "effect"
// Function that only needs write access to a queue
const producer = (enqueue: Queue.Enqueue<string>) =>
Effect.gen(function*() {
yield* Queue.offer(enqueue, "hello")
yield* Queue.offerAll(enqueue, ["world", "!"])
})
const program = Effect.gen(function*() {
const queue = yield* Queue.bounded<string>(10)
yield* producer(queue)
})
Companion namespace containing type-level metadata for the Enqueue
write-only queue interface.
Enqueue<function (type parameter) A in <A, E>(self: Enqueue<A, E>): Effect<boolean>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>): Effect<boolean>E>): 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<boolean> =>
import corecore.const withFiber: <
A,
E = never,
R = never
>(
evaluate: (
fiber: FiberImpl<unknown, unknown>
) => Effect.Effect<A, E, R>
) => Effect.Effect<A, E, R>
fiber: internalEffect.FiberImpl<
unknown,
unknown
>
(parameter) fiber: {
id: number;
interruptible: boolean;
currentOpCount: number;
currentLoopCount: number;
_stack: Array<Primitive>;
_observers: Array<(exit: Exit.Exit<A, E>) => void>;
_exit: Exit.Exit<A, E> | undefined;
_currentExit: Exit.Exit<A, E> | undefined;
_children: Set<FiberImpl<any, any>> | undefined;
_interruptedCause: Cause.Cause<never> | undefined;
_yielded: Exit.Exit<any, any> | (() => void) | undefined;
context: Context.Context<never>;
currentScheduler: Scheduler.Scheduler;
currentTracerContext: Tracer.Tracer["context"];
currentSpan: Tracer.AnySpan | undefined;
currentLogLevel: LogLevel.LogLevel;
minimumLogLevel: LogLevel.LogLevel;
currentStackFrame: StackFrame | undefined;
runtimeMetrics: Metric.FiberRuntimeMetricsService | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
_dispatcher: Scheduler.SchedulerDispatcher | undefined;
currentDispatcher: SchedulerDispatcher;
getRef: <X>(ref: Context.Reference<X>) => X;
addObserver: (cb: (exit: Exit<unknown, unknown>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<unknown, unknown> | undefined;
evaluate: (effect: core.Primitive) => void;
runLoop: (effect: core.Primitive) => typeof core.Yield | Exit<unknown, unknown>;
getCont: <S extends core.contA | core.contE>(symbol: S) => (core.Primitive & Record<S, (value: any, fiber: internalEffect.FiberImpl) => core.Primitive>) | undefined;
yieldWith: (value: Exit.Exit<any, any> | (() => void)) => Yield;
children: () => Set<Fiber.Fiber<any, any>>;
pipe: () => unknown;
setContext: (context: Context.Context<never>) => void;
currentSpanLocal: Span | undefined;
}
const failCause: {
<E>(cause: Cause<E>): <A>(
self: Enqueue<A, E>
) => Effect<boolean>
<A, E>(
self: Enqueue<A, E>,
cause: Cause<E>
): Effect<boolean>
}
self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
import internalEffectinternalEffect.const causeInterrupt: (
fiberId?: number | undefined
) => Cause.Cause<never>
fiber: internalEffect.FiberImpl<
unknown,
unknown
>
(parameter) fiber: {
id: number;
interruptible: boolean;
currentOpCount: number;
currentLoopCount: number;
_stack: Array<Primitive>;
_observers: Array<(exit: Exit.Exit<A, E>) => void>;
_exit: Exit.Exit<A, E> | undefined;
_currentExit: Exit.Exit<A, E> | undefined;
_children: Set<FiberImpl<any, any>> | undefined;
_interruptedCause: Cause.Cause<never> | undefined;
_yielded: Exit.Exit<any, any> | (() => void) | undefined;
context: Context.Context<never>;
currentScheduler: Scheduler.Scheduler;
currentTracerContext: Tracer.Tracer["context"];
currentSpan: Tracer.AnySpan | undefined;
currentLogLevel: LogLevel.LogLevel;
minimumLogLevel: LogLevel.LogLevel;
currentStackFrame: StackFrame | undefined;
runtimeMetrics: Metric.FiberRuntimeMetricsService | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
_dispatcher: Scheduler.SchedulerDispatcher | undefined;
currentDispatcher: SchedulerDispatcher;
getRef: <X>(ref: Context.Reference<X>) => X;
addObserver: (cb: (exit: Exit<unknown, unknown>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<unknown, unknown> | undefined;
evaluate: (effect: core.Primitive) => void;
runLoop: (effect: core.Primitive) => typeof core.Yield | Exit<unknown, unknown>;
getCont: <S extends core.contA | core.contE>(symbol: S) => (core.Primitive & Record<S, (value: any, fiber: internalEffect.FiberImpl) => core.Primitive>) | undefined;
yieldWith: (value: Exit.Exit<any, any> | (() => void)) => Yield;
children: () => Set<Fiber.Fiber<any, any>>;
pipe: () => unknown;
setContext: (context: Context.Context<never>) => void;
currentSpanLocal: Span | undefined;
}
FiberImpl<unknown, unknown>.id: numberid)))