<A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean>Interrupts the queue gracefully with the current fiber's interruption cause.
Details
If the queue still contains items, it enters the closing state so buffered items can be drained before consumers observe the interruption. If it is empty, it transitions directly to done. Returns false if the queue was already closing or done.
Example (Interrupting queues)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number>(10)
yield* TxQueue.offer(queue, 42)
// Interrupt gracefully - allows remaining items to be consumed
const result = yield* TxQueue.interrupt(queue)
console.log(result) // true
})combinators
Source effect/TxQueue.ts:11352 lines
export const const interrupt: <A, E>(
self: TxEnqueue<A, E>
) => Effect.Effect<boolean>
Interrupts the queue gracefully with the current fiber's interruption cause.
Details
If the queue still contains items, it enters the closing state so buffered items can be drained before consumers observe the interruption. If it is empty, it transitions directly to done. Returns false if the queue was already closing or done.
Example (Interrupting queues)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number>(10)
yield* TxQueue.offer(queue, 42)
// Interrupt gracefully - allows remaining items to be consumed
const result = yield* TxQueue.interrupt(queue)
console.log(result) // true
})
interrupt = <function (type parameter) A in <A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean>A, function (type parameter) E in <A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean>E>(self: TxEnqueue<A, E>(parameter) self: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
interface TxEnqueue<in A, in E = never>Namespace containing type definitions for TxEnqueue variance annotations.
A TxEnqueue represents the write-only interface of a transactional queue, providing
operations for adding elements (enqueue operations) and inspecting queue state.
Example (Offering values through enqueue handles)
import { Effect, TxQueue } from "effect"
import type { Cause } from "effect"
const program = Effect.gen(function*() {
// Queue without error channel
const queue = yield* TxQueue.bounded<number>(10)
const accepted = yield* TxQueue.offer(queue, 42)
// Queue with error channel for completion signaling
const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offerAll(faultTolerantQueue, [1, 2, 3])
yield* TxQueue.fail(faultTolerantQueue, "processing complete")
// Works with Done for clean completion
const completableQueue = yield* TxQueue.bounded<
string,
Cause.Done
>(5)
yield* TxQueue.offer(completableQueue, "task")
yield* TxQueue.end(completableQueue)
})
TxEnqueue<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean>A, function (type parameter) E in <A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean>E>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<boolean> =>
import EffectEffect.withFiber((fiber: Fiber<unknown, unknown>(parameter) fiber: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (context: Context.Context<never>) => void;
currentScheduler: Scheduler;
currentDispatcher: SchedulerDispatcher;
currentSpan: AnySpan | undefined;
currentLogLevel: LogLevel;
minimumLogLevel: LogLevel;
currentStackFrame: StackFrame | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
addObserver: (cb: (exit: Exit<A, E>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<A, E> | undefined;
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; <…;
}
const failCause: {
<E>(cause: Cause.Cause<E>): <A>(
self: TxEnqueue<A, E>
) => Effect.Effect<boolean>
<A, E>(
self: TxEnqueue<A, E>,
cause: Cause.Cause<E>
): Effect.Effect<boolean>
}
self: TxEnqueue<A, E>(parameter) self: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
import CauseCause.interrupt(fiber: Fiber<unknown, unknown>(parameter) fiber: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (context: Context.Context<never>) => void;
currentScheduler: Scheduler;
currentDispatcher: SchedulerDispatcher;
currentSpan: AnySpan | undefined;
currentLogLevel: LogLevel;
minimumLogLevel: LogLevel;
currentStackFrame: StackFrame | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
addObserver: (cb: (exit: Exit<A, E>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<A, E> | undefined;
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; <…;
}
Referenced by 1 symbols