<A, E = never>(
options?:
| {
readonly capacity?: number | undefined
readonly strategy?: "suspend" | "dropping" | "sliding" | undefined
}
| undefined
): Effect<Queue<A, E>>Creates a Queue with optional capacity and overflow strategy.
Details
By default the queue is unbounded and uses the "suspend" strategy. Provide
capacity for a bounded queue and choose "suspend", "dropping", or
"sliding" to control what happens when the queue is full. The returned
queue can be offered to, taken from, failed, ended, interrupted, or shut down.
Example (Creating queues)
import { Cause, Effect, Queue } from "effect"
Effect.gen(function*() {
const queue = yield* Queue.make<number, string | Cause.Done>()
// add messages to the queue
yield* Queue.offer(queue, 1)
yield* Queue.offer(queue, 2)
yield* Queue.offerAll(queue, [3, 4, 5])
// take messages from the queue
const messages = yield* Queue.takeAll(queue)
console.log(messages) // [1, 2, 3, 4, 5]
// signal that the queue is done
yield* Queue.end(queue)
const done = yield* Effect.flip(Queue.take(queue))
console.log(Cause.isDone(done)) // true
// signal that another queue has failed
const failedQueue = yield* Queue.make<number, string>()
const failed = yield* Queue.fail(failedQueue, "boom")
console.log(failed) // true
})constructors
Source effect/Queue.ts:44121 lines
export const const make: <A, E = never>(
options?:
| {
readonly capacity?: number | undefined
readonly strategy?:
| "suspend"
| "dropping"
| "sliding"
| undefined
}
| undefined
) => Effect<Queue<A, E>>
Creates a Queue with optional capacity and overflow strategy.
Details
By default the queue is unbounded and uses the "suspend" strategy. Provide
capacity for a bounded queue and choose "suspend", "dropping", or
"sliding" to control what happens when the queue is full. The returned
queue can be offered to, taken from, failed, ended, interrupted, or shut down.
Example (Creating queues)
import { Cause, Effect, Queue } from "effect"
Effect.gen(function*() {
const queue = yield* Queue.make<number, string | Cause.Done>()
// add messages to the queue
yield* Queue.offer(queue, 1)
yield* Queue.offer(queue, 2)
yield* Queue.offerAll(queue, [3, 4, 5])
// take messages from the queue
const messages = yield* Queue.takeAll(queue)
console.log(messages) // [1, 2, 3, 4, 5]
// signal that the queue is done
yield* Queue.end(queue)
const done = yield* Effect.flip(Queue.take(queue))
console.log(Cause.isDone(done)) // true
// signal that another queue has failed
const failedQueue = yield* Queue.make<number, string>()
const failed = yield* Queue.fail(failedQueue, "boom")
console.log(failed) // true
})
make = <function (type parameter) A in <A, E = never>(options?: {
readonly capacity?: number | undefined;
readonly strategy?: "suspend" | "dropping" | "sliding" | undefined;
} | undefined): Effect<Queue<A, E>>
function (type parameter) E in <A, E = never>(options?: {
readonly capacity?: number | undefined;
readonly strategy?: "suspend" | "dropping" | "sliding" | undefined;
} | undefined): Effect<Queue<A, E>>
options: | {
readonly capacity?: number | undefined
readonly strategy?:
| "suspend"
| "dropping"
| "sliding"
| undefined
}
| undefined
capacity?: number | undefinedcapacity?: number | undefined
readonly strategy?: | "suspend"
| "dropping"
| "sliding"
| undefined
import EffectEffect<interface Queue<in out A, in out E = never>A Queue is an asynchronous queue that can be offered to and taken from.
Details
It also supports signaling that it is done or failed.
Example (Offering and taking queue values)
import { Effect, Queue } from "effect"
const program = Effect.gen(function*() {
// Create a bounded queue
const queue = yield* Queue.bounded<string>(10)
// Producer: offer items to the queue
yield* Queue.offer(queue, "hello")
yield* Queue.offerAll(queue, ["world", "!"])
// Consumer: take items from the queue
const item1 = yield* Queue.take(queue)
const item2 = yield* Queue.take(queue)
const item3 = yield* Queue.take(queue)
console.log([item1, item2, item3]) // ["hello", "world", "!"]
})
Companion namespace containing type-level metadata and low-level state types
for Queue.
Queue<function (type parameter) A in <A, E = never>(options?: {
readonly capacity?: number | undefined;
readonly strategy?: "suspend" | "dropping" | "sliding" | undefined;
} | undefined): Effect<Queue<A, E>>
function (type parameter) E in <A, E = never>(options?: {
readonly capacity?: number | undefined;
readonly strategy?: "suspend" | "dropping" | "sliding" | undefined;
} | undefined): Effect<Queue<A, E>>
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 self: anyself = var Object: ObjectConstructorProvides functionality common to all JavaScript objects.
Object.ObjectConstructor.create(o: object | null): any (+1 overload)Creates an object that has the specified prototype or that has null prototype.
create(const QueueProto: {
toJSON(this: Queue<unknown, unknown>): {
_id: string
state: "Done" | "Open" | "Closing"
size: number
}
pipe(): unknown
toString(): string
[NodeInspectSymbol](): any
"~effect/Queue": {
_A: <A>(a: A) => A
_E: <A>(a: A) => A
}
"~effect/Queue/Enqueue": {
_A: <A>(a: A) => A
_E: <A>(a: A) => A
}
"~effect/Queue/Dequeue": {
_A: <A>(a: A) => A
_E: <A>(a: A) => A
}
}
const self: anyself.dispatcher = 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>.currentDispatcher: SchedulerDispatcher(property) FiberImpl<unknown, unknown>.currentDispatcher: {
scheduleTask: (task: () => void, priority: number) => void;
flush: () => void;
}
const self: anyself.capacity = options: | {
readonly capacity?: number | undefined
readonly strategy?:
| "suspend"
| "dropping"
| "sliding"
| undefined
}
| undefined
capacity?: number | undefinedcapacity ?? var Number: NumberConstructorAn object that represents a number of any kind. All JavaScript numbers are 64-bit floating-point numbers.
Number.NumberConstructor.POSITIVE_INFINITY: numberA value greater than the largest number that can be represented in JavaScript.
JavaScript displays POSITIVE_INFINITY values as infinity.
POSITIVE_INFINITY
const self: anyself.strategy = options: | {
readonly capacity?: number | undefined
readonly strategy?:
| "suspend"
| "dropping"
| "sliding"
| undefined
}
| undefined
strategy?: | "suspend"
| "dropping"
| "sliding"
| undefined
const self: anyself.messages = import MutableListMutableList.make()
const self: anyself.scheduleRunning = false
const self: anyself.state = {
_tag: string_tag: "Open",
takers: Set<unknown>takers: new var Set: SetConstructor
new <unknown>(iterable?: Iterable<unknown> | null | undefined) => Set<unknown> (+1 overload)
offers: Set<unknown>offers: new var Set: SetConstructor
new <unknown>(iterable?: Iterable<unknown> | null | undefined) => Set<unknown> (+1 overload)
awaiters: Set<unknown>awaiters: new var Set: SetConstructor
new <unknown>(iterable?: Iterable<unknown> | null | undefined) => Set<unknown> (+1 overload)
import internalEffectinternalEffect.const succeed: <A>(
value: A
) => Effect.Effect<A>
const self: anyself)
})Referenced by 10 symbols