<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>): import EffectEffect<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)))