1/**
2 * Passes values asynchronously between fibers.
3 *
4 * A `Queue<A, E>` accepts values, hands each value to one consumer in offer
5 * order, and can complete, fail, interrupt, or shut down. Queues can be bounded
6 * or unbounded, and bounded queues can suspend, drop, or slide values when
7 * producers are faster than consumers.
8 *
9 * @since 3.8.0
10 */
11import * as Arr from "./Array.ts"
12import type { Cause, Done } from "./Cause.ts"
13import type { Effect } from "./Effect.ts"
14import type { Exit, Failure } from "./Exit.ts"
15import { constant, constTrue, dual, identity } from "./Function.ts"
16import type { Inspectable } from "./Inspectable.ts"
17import * as core from "./internal/core.ts"
18import { PipeInspectableProto } from "./internal/core.ts"
19import * as Count from "./internal/count.ts"
20import * as internalEffect from "./internal/effect.ts"
21import * as MutableList from "./MutableList.ts"
22import * as Option from "./Option.ts"
23import { hasProperty } from "./Predicate.ts"
24import * as Pull from "./Pull.ts"
25import type { SchedulerDispatcher } from "./Scheduler.ts"
26import type * as Types from "./Types.ts"
27
28const TypeId = "~effect/Queue"
29const EnqueueTypeId = "~effect/Queue/Enqueue"
30const DequeueTypeId = "~effect/Queue/Dequeue"
31
32/**
33 * Type guard to check if a value is a Queue.
34 *
35 * **When to use**
36 *
37 * Use to narrow an unknown value to a full `Queue` before passing it to APIs
38 * that need both offering and taking capabilities.
39 *
40 * @see {@link isEnqueue} for checking values that only need write access
41 * @see {@link isDequeue} for checking values that only need read access
42 *
43 * @category guards
44 * @since 2.0.0
45 */
46export const isQueue = <A = unknown, E = unknown>(
47 u: unknown
48): u is Queue<A, E> => hasProperty(u, TypeId)
49
50/**
51 * Type guard to check if a value is an Enqueue.
52 *
53 * **When to use**
54 *
55 * Use to narrow an unknown value before calling queue operations that require
56 * write-side access.
57 *
58 * **Gotchas**
59 *
60 * A full `Queue` also satisfies this guard because every queue includes the
61 * enqueue side.
62 *
63 * @see {@link isQueue} for checking for a full read-write queue handle
64 * @see {@link isDequeue} for checking for the read side of a queue
65 * @see {@link asEnqueue} for narrowing an existing `Queue` to its write-only interface
66 *
67 * @category guards
68 * @since 2.0.0
69 */
70export const isEnqueue = <A = unknown, E = unknown>(
71 u: unknown
72): u is Enqueue<A, E> => hasProperty(u, EnqueueTypeId)
73
74/**
75 * Type guard to check if a value is a Dequeue.
76 *
77 * **When to use**
78 *
79 * Use to narrow an unknown value before passing it to read-side queue
80 * operations.
81 *
82 * @see {@link Dequeue} for the read-side queue handle checked by this guard
83 * @see {@link isQueue} for checking for a full read-write queue handle
84 * @see {@link isEnqueue} for checking for the write side of a queue
85 * @see {@link asDequeue} for narrowing an existing `Queue` to its read-only interface
86 *
87 * @category guards
88 * @since 2.0.0
89 */
90export const isDequeue = <A = unknown, E = unknown>(
91 u: unknown
92): u is Dequeue<A, E> => hasProperty(u, DequeueTypeId)
93
94/**
95 * Converts a `Queue` to its write-only `Enqueue` interface.
96 *
97 * **When to use**
98 *
99 * Use to expose only the producer side of a `Queue` to code that should offer
100 * values or signal queue lifecycle.
101 *
102 * **Gotchas**
103 *
104 * This is a type-level capability restriction. It returns the same queue
105 * object, so it does not hide read operations at runtime.
106 *
107 * @see {@link asDequeue} for exposing only the read side of a `Queue`
108 * @see {@link Enqueue} for the write-only queue handle returned by this conversion
109 *
110 * @category converting
111 * @since 4.0.0
112 */
113export const asEnqueue = <A, E>(self: Queue<A, E>): Enqueue<A, E> => self
114
115/**
116 * Narrows a `Queue` to a `Dequeue`, exposing the consumer side of the queue.
117 *
118 * **When to use**
119 *
120 * Use to pass a queue to code that should consume values while keeping
121 * producer-side operations out of that code's TypeScript type.
122 *
123 * **Gotchas**
124 *
125 * This is a type-level narrowing operation. It returns the same queue object
126 * and does not create a runtime wrapper.
127 *
128 * @see {@link asEnqueue} for narrowing a queue to its producer side
129 * @see {@link Dequeue} for the consumer-side queue handle returned by this function
130 *
131 * @category converting
132 * @since 4.0.0
133 */
134export const asDequeue: <A, E>(self: Queue<A, E>) => Dequeue<A, E> = identity
135
136/**
137 * An `Enqueue` is a queue that can be offered to.
138 *
139 * **Details**
140 *
141 * This interface represents the write-only part of a Queue, allowing you to offer
142 * elements to the queue but not take elements from it.
143 *
144 * **Example** (Offering through enqueue handles)
145 *
146 * ```ts import.meta.vitest
147 * import { Effect, Queue } from "effect"
148 *
149 * // Function that only needs write access to a queue
150 * const producer = (enqueue: Queue.Enqueue<string>) =>
151 * Effect.gen(function*() {
152 * yield* Queue.offer(enqueue, "hello")
153 * yield* Queue.offerAll(enqueue, ["world", "!"])
154 * })
155 *
156 * const program = Effect.gen(function*() {
157 * const queue = yield* Queue.bounded<string>(10)
158 * yield* producer(queue)
159 * return yield* Queue.takeAll(queue)
160 * })
161 *
162 * await Effect.runPromise(program) // => ["hello", "world", "!"]
163 * ```
164 *
165 * @category models
166 * @since 2.0.0
167 */
168export interface Enqueue<in A, in E = never> extends Inspectable {
169 readonly [EnqueueTypeId]: Enqueue.Variance<A, E>
170 readonly strategy: "suspend" | "dropping" | "sliding"
171 readonly dispatcher: SchedulerDispatcher
172 capacity: number
173 messages: MutableList.MutableList<any>
174 state: Queue.State<any, any>
175 scheduleRunning: boolean
176}
177
178/**
179 * Companion namespace containing type-level metadata for the `Enqueue`
180 * write-only queue interface.
181 *
182 * @since 2.0.0
183 */
184export declare namespace Enqueue {
185 /**
186 * Type-level variance marker for `Enqueue`.
187 *
188 * **Details**
189 *
190 * `Enqueue` is contravariant in both its offered value type `A` and failure
191 * type `E`, because values and failures flow into the queue through this
192 * handle.
193 *
194 * @category models
195 * @since 4.0.0
196 */
197 export interface Variance<A, E> {
198 _A: Types.Contravariant<A>
199 _E: Types.Contravariant<E>
200 }
201}
202
203/**
204 * A `Dequeue` is a queue that can be taken from.
205 *
206 * **Details**
207 *
208 * This interface represents the read-only part of a Queue, allowing you to take
209 * elements from the queue but not offer elements to it.
210 *
211 * **Example** (Taking through dequeue handles)
212 *
213 * ```ts import.meta.vitest
214 * import { Effect, Queue } from "effect"
215 *
216 * const program = Effect.gen(function*() {
217 * const queue = yield* Queue.bounded<string, never>(10)
218 *
219 * // A Dequeue can only take elements
220 * const dequeue: Queue.Dequeue<string> = queue
221 *
222 * // Pre-populate the queue
223 * yield* Queue.offerAll(queue, ["a", "b", "c"])
224 *
225 * // Take elements using dequeue interface
226 * const item = yield* Queue.take(dequeue)
227 * return item
228 * })
229 *
230 * await Effect.runPromise(program) // => "a"
231 * ```
232 *
233 * @category models
234 * @since 2.0.0
235 */
236export interface Dequeue<out A, out E = never> extends Inspectable {
237 readonly [DequeueTypeId]: Dequeue.Variance<A, E>
238 readonly strategy: "suspend" | "dropping" | "sliding"
239 readonly dispatcher: SchedulerDispatcher
240 capacity: number
241 messages: MutableList.MutableList<any>
242 state: Queue.State<any, any>
243 scheduleRunning: boolean
244}
245
246/**
247 * Companion namespace containing type-level metadata for the `Dequeue`
248 * read-only queue interface.
249 *
250 * @since 2.0.0
251 */
252export declare namespace Dequeue {
253 /**
254 * Type-level variance marker for `Dequeue`.
255 *
256 * **Details**
257 *
258 * `Dequeue` is covariant in both the taken value type `A` and failure type
259 * `E`, because values and failures are observed through this handle.
260 *
261 * @category models
262 * @since 4.0.0
263 */
264 export interface Variance<A, E> {
265 _A: Types.Covariant<A>
266 _E: Types.Covariant<E>
267 }
268}
269
270/**
271 * A `Queue` is an asynchronous queue that can be offered to and taken from.
272 *
273 * **Details**
274 *
275 * It also supports signaling that it is done or failed.
276 *
277 * **Example** (Offering and taking queue values)
278 *
279 * ```ts import.meta.vitest
280 * import { Effect, Queue } from "effect"
281 *
282 * const program = Effect.gen(function*() {
283 * // Create a bounded queue
284 * const queue = yield* Queue.bounded<string>(10)
285 *
286 * // Producer: offer items to the queue
287 * yield* Queue.offer(queue, "hello")
288 * yield* Queue.offerAll(queue, ["world", "!"])
289 *
290 * // Consumer: take items from the queue
291 * const item1 = yield* Queue.take(queue)
292 * const item2 = yield* Queue.take(queue)
293 * const item3 = yield* Queue.take(queue)
294 *
295 * return [item1, item2, item3]
296 * })
297 *
298 * await Effect.runPromise(program) // => ["hello", "world", "!"]
299 * ```
300 *
301 * @category models
302 * @since 2.0.0
303 */
304export interface Queue<in out A, in out E = never> extends Enqueue<A, E>, Dequeue<A, E> {
305 readonly [TypeId]: Queue.Variance<A, E>
306}
307
308/**
309 * Companion namespace containing type-level metadata and low-level state types
310 * for `Queue`.
311 *
312 * @since 2.0.0
313 */
314export declare namespace Queue {
315 /**
316 * Type-level variance marker for `Queue`.
317 *
318 * **Details**
319 *
320 * A full `Queue` is invariant in both `A` and `E` because the same handle can
321 * both produce and consume values and failures.
322 *
323 * @category models
324 * @since 4.0.0
325 */
326 export interface Variance<A, E> {
327 _A: Types.Invariant<A>
328 _E: Types.Invariant<E>
329 }
330
331 /**
332 * Tagged state of a `Queue`.
333 *
334 * **Details**
335 *
336 * `Open` queues can accept offers and takers, `Closing` queues are
337 * completing with a stored failure exit, and `Done` queues have finished.
338 * This is low-level metadata exposed by the queue model; most users should
339 * inspect queues through the public operations.
340 *
341 * @category models
342 * @since 4.0.0
343 */
344 export type State<A, E> =
345 | {
346 readonly _tag: "Open"
347 readonly takers: Set<Taker<E>>
348 readonly offers: Set<OfferEntry<A>>
349 readonly awaiters: Set<(_: Effect<void, E>) => void>
350 }
351 | {
352 readonly _tag: "Closing"
353 readonly takers: Set<Taker<E>>
354 readonly offers: Set<OfferEntry<A>>
355 readonly awaiters: Set<(_: Effect<void, E>) => void>
356 readonly exit: Failure<never, E>
357 }
358 | {
359 readonly _tag: "Done"
360 readonly exit: Failure<never, E>
361 }
362
363 /**
364 * Represents a suspended offer waiting to be admitted to a bounded queue.
365 *
366 * **Details**
367 *
368 * An entry is either a single message or a batch with an offset into its
369 * remaining messages, plus a resume callback that completes the suspended
370 * offer when the queue can accept more input.
371 *
372 * @category models
373 * @since 4.0.0
374 */
375 export type OfferEntry<A> =
376 | {
377 readonly _tag: "Array"
378 readonly remaining: Array<A>
379 offset: number
380 readonly resume: (_: Effect<Array<A>>) => void
381 }
382 | {
383 readonly _tag: "Single"
384 readonly message: A
385 readonly resume: (_: Effect<boolean>) => void
386 }
387
388 /**
389 * Represents a suspended take waiting for the queue to become readable.
390 *
391 * **Details**
392 *
393 * `ready` reports whether the take can now complete, so a batch take is
394 * only resumed once its minimum is available. `resume` completes the
395 * suspended take, with `void` to retry or with a failure exit when the queue
396 * is done.
397 *
398 * @category models
399 * @since 4.0.0
400 */
401 export interface Taker<E> {
402 readonly ready: () => boolean
403 readonly resume: (_: Effect<void, E>) => void
404 }
405}
406
407const variance = {
408 _A: identity,
409 _E: identity
410}
411const QueueProto = {
412 [TypeId]: variance,
413 [EnqueueTypeId]: variance,
414 [DequeueTypeId]: variance,
415 ...PipeInspectableProto,
416 toJSON(this: Queue<unknown, unknown>) {
417 return {
418 _id: "effect/Queue",
419 state: this.state._tag,
420 size: sizeUnsafe(this)
421 }
422 }
423}
424
425/**
426 * Creates a `Queue` with optional capacity and overflow strategy.
427 *
428 * **Details**
429 *
430 * By default the queue is unbounded and uses the `"suspend"` strategy. Provide
431 * `capacity` for a bounded queue and choose `"suspend"`, `"dropping"`, or
432 * `"sliding"` to control what happens when the queue is full. The returned
433 * queue can be offered to, taken from, failed, ended, interrupted, or shut down.
434 *
435 * **Example** (Creating queues)
436 *
437 * ```ts import.meta.vitest
438 * import { Cause, Effect, Queue } from "effect"
439 *
440 * const program = Effect.gen(function*() {
441 * const queue = yield* Queue.make<number, string | Cause.Done>()
442 *
443 * // add messages to the queue
444 * yield* Queue.offer(queue, 1)
445 * yield* Queue.offer(queue, 2)
446 * yield* Queue.offerAll(queue, [3, 4, 5])
447 *
448 * // take messages from the queue
449 * const messages = yield* Queue.takeAll(queue)
450 *
451 * // signal that the queue is done
452 * yield* Queue.end(queue)
453 * const done = yield* Effect.flip(Queue.take(queue))
454 *
455 * // signal that another queue has failed
456 * const failedQueue = yield* Queue.make<number, string>()
457 * const failed = yield* Queue.fail(failedQueue, "boom")
458 * return { messages, done, failed }
459 * })
460 *
461 * await Effect.runPromise(program) // => { messages: [1, 2, 3, 4, 5], done: Cause.Done(), failed: true }
462 * ```
463 *
464 * @category constructors
465 * @since 4.0.0
466 */
467export const make = <A, E = never>(
468 options?: {
469 readonly capacity?: number | undefined
470 readonly strategy?: "suspend" | "dropping" | "sliding" | undefined
471 } | undefined
472): Effect<Queue<A, E>> =>
473 core.withFiber((fiber) => {
474 const self = Object.create(QueueProto)
475 self.dispatcher = fiber.currentDispatcher
476 self.capacity = options?.capacity ?? Number.POSITIVE_INFINITY
477 self.strategy = options?.strategy ?? "suspend"
478 self.messages = MutableList.make()
479 self.scheduleRunning = false
480 self.state = {
481 _tag: "Open",
482 takers: new Set(),
483 offers: new Set(),
484 awaiters: new Set()
485 }
486 return internalEffect.succeed(self)
487 })
488
489/**
490 * Creates a bounded queue with the specified capacity that uses backpressure strategy.
491 *
492 * **Details**
493 *
494 * When the queue reaches capacity, producers will be suspended until space becomes available.
495 * This ensures all messages are processed but may slow down producers.
496 *
497 * **Example** (Creating bounded queues)
498 *
499 * ```ts import.meta.vitest
500 * import { Effect, Queue } from "effect"
501 *
502 * const program = Effect.gen(function*() {
503 * const queue = yield* Queue.bounded<string>(5)
504 *
505 * // This will succeed as queue has capacity
506 * yield* Queue.offer(queue, "first")
507 * yield* Queue.offer(queue, "second")
508 *
509 * const size = yield* Queue.size(queue)
510 * return size
511 * })
512 *
513 * await Effect.runPromise(program) // => 2
514 * ```
515 *
516 * @category constructors
517 * @since 2.0.0
518 */
519export const bounded = <A, E = never>(capacity: number): Effect<Queue<A, E>> => make({ capacity })
520
521/**
522 * Creates a bounded queue with sliding strategy. When the queue reaches capacity,
523 * new elements are added and the oldest elements are dropped.
524 *
525 * **When to use**
526 *
527 * Use when you need producer offers not to block and can accept dropping the
528 * oldest messages, such as when maintaining a rolling window of recent values.
529 *
530 * **Example** (Creating sliding queues)
531 *
532 * ```ts import.meta.vitest
533 * import { Effect, Queue } from "effect"
534 *
535 * const program = Effect.gen(function*() {
536 * const queue = yield* Queue.sliding<number>(3)
537 *
538 * // Fill the queue to capacity
539 * yield* Queue.offer(queue, 1)
540 * yield* Queue.offer(queue, 2)
541 * yield* Queue.offer(queue, 3)
542 *
543 * // This will succeed, dropping the oldest element (1)
544 * yield* Queue.offer(queue, 4)
545 *
546 * const all = yield* Queue.takeAll(queue)
547 * return all
548 * })
549 *
550 * await Effect.runPromise(program) // => [2, 3, 4]
551 * ```
552 *
553 * @category constructors
554 * @since 2.0.0
555 */
556export const sliding = <A, E = never>(capacity: number): Effect<Queue<A, E>> => make({ capacity, strategy: "sliding" })
557
558/**
559 * Creates a bounded queue with dropping strategy. When the queue reaches capacity,
560 * new elements are dropped and the offer operation returns false.
561 *
562 * **When to use**
563 *
564 * Use when you need producer offers not to block while preserving existing
565 * queued messages, even if new messages may be dropped when the queue is full.
566 *
567 * **Example** (Creating dropping queues)
568 *
569 * ```ts import.meta.vitest
570 * import { Effect, Queue } from "effect"
571 *
572 * const program = Effect.gen(function*() {
573 * const queue = yield* Queue.dropping<number>(2)
574 *
575 * // Fill the queue to capacity
576 * const success1 = yield* Queue.offer(queue, 1)
577 * const success2 = yield* Queue.offer(queue, 2)
578 *
579 * // This will be dropped
580 * const success3 = yield* Queue.offer(queue, 3)
581 *
582 * const all = yield* Queue.takeAll(queue)
583 * return [success1, success2, success3, all]
584 * })
585 *
586 * await Effect.runPromise(program) // => [true, true, false, [1, 2]]
587 * ```
588 *
589 * @category constructors
590 * @since 2.0.0
591 */
592export const dropping = <A, E = never>(capacity: number): Effect<Queue<A, E>> =>
593 make({ capacity, strategy: "dropping" })
594
595/**
596 * Creates an unbounded queue that can grow to any size without blocking producers.
597 *
598 * **When to use**
599 *
600 * Use when you need producers to add messages without backpressure and accept
601 * unbounded memory growth.
602 *
603 * **Example** (Creating unbounded queues)
604 *
605 * ```ts import.meta.vitest
606 * import { Effect, Queue } from "effect"
607 *
608 * const program = Effect.gen(function*() {
609 * const queue = yield* Queue.unbounded<string>()
610 *
611 * // Producers can always add messages without blocking
612 * yield* Queue.offer(queue, "message1")
613 * yield* Queue.offer(queue, "message2")
614 * yield* Queue.offerAll(queue, ["message3", "message4", "message5"])
615 *
616 * // Check current size
617 * const size = yield* Queue.size(queue)
618 *
619 * // Take all messages
620 * const messages = yield* Queue.takeAll(queue)
621 * return { size, messages }
622 * })
623 *
624 * await Effect.runPromise(program) // => { size: 5, messages: ["message1", "message2", "message3", "message4", "message5"] }
625 * ```
626 *
627 * @category constructors
628 * @since 2.0.0
629 */
630export const unbounded = <A, E = never>(): Effect<Queue<A, E>> => make()
631
632/**
633 * Adds a message to the queue. Returns `false` if the queue is done.
634 *
635 * **Details**
636 *
637 * For bounded queues, this operation may suspend if the queue is at capacity,
638 * depending on the backpressure strategy. For dropping/sliding queues, it may
639 * return false or succeed immediately by dropping/sliding existing messages.
640 *
641 * **Example** (Offering a value)
642 *
643 * ```ts import.meta.vitest
644 * import { Effect, Queue } from "effect"
645 *
646 * const program = Effect.gen(function*() {
647 * const queue = yield* Queue.bounded<number>(3)
648 *
649 * // Successfully add messages to queue
650 * const success1 = yield* Queue.offer(queue, 1)
651 * const success2 = yield* Queue.offer(queue, 2)
652 *
653 * // Queue state
654 * const size = yield* Queue.size(queue)
655 * return { offered: [success1, success2], size }
656 * })
657 *
658 * await Effect.runPromise(program) // => { offered: [true, true], size: 2 }
659 * ```
660 *
661 * @category offering
662 * @since 2.0.0
663 */
664export const offer: {
665 <A>(message: A): <E>(self: Enqueue<A, E>) => Effect<boolean>
666 <A, E>(self: Enqueue<A, E>, message: Types.NoInfer<A>): Effect<boolean>
667} = dual(
668 2,
669 <A, E>(self: Enqueue<A, E>, message: Types.NoInfer<A>): Effect<boolean> =>
670 internalEffect.suspend(() =>
671 offerUnsafe(self, message)
672 ? exitTrue
673 : self.state._tag === "Open" && self.strategy === "suspend"
674 ? offerOrWait(self, message)
675 : exitFalse
676 )
677)
678
679/**
680 * Adds a message to the queue synchronously. Returns `false` if the queue is done.
681 *
682 * **When to use**
683 *
684 * Use when you are already in synchronous queue internals or a performance
685 * boundary where wrapping the mutation in `Effect` is intentionally avoided.
686 *
687 * **Gotchas**
688 *
689 * This is an unsafe operation that directly modifies the queue without Effect wrapping.
690 * Use this only when you're certain about the synchronous nature of the operation.
691 *
692 * **Example** (Offering a value synchronously)
693 *
694 * ```ts import.meta.vitest
695 * import { Effect, Queue } from "effect"
696 *
697 * // Create a queue effect and extract the queue for unsafe operations
698 * const program = Effect.gen(function*() {
699 * const queue = yield* Queue.bounded<number>(3)
700 *
701 * // Add messages synchronously using unsafe API
702 * const success1 = Queue.offerUnsafe(queue, 1)
703 * const success2 = Queue.offerUnsafe(queue, 2)
704 *
705 * // Check current size
706 * const size = Queue.sizeUnsafe(queue)
707 * return { offered: [success1, success2], size }
708 * })
709 *
710 * await Effect.runPromise(program) // => { offered: [true, true], size: 2 }
711 * ```
712 *
713 * @category offering
714 * @since 4.0.0
715 */
716export const offerUnsafe = <A, E>(self: Enqueue<A, E>, message: Types.NoInfer<A>): boolean => {
717 if (self.state._tag !== "Open") {
718 return false
719 } else if (self.messages.length >= self.capacity) {
720 if (self.strategy === "sliding") {
721 MutableList.take(self.messages)
722 MutableList.append(self.messages, message)
723 return true
724 } else if (self.capacity <= 0 && self.state.takers.size > 0) {
725 MutableList.append(self.messages, message)
726 releaseTakers(self as Queue<A, E>)
727 return true
728 }
729 return false
730 }
731 MutableList.append(self.messages, message)
732 scheduleReleaseTaker(self as Queue<A, E>)
733 return true
734}
735
736/**
737 * Adds multiple messages to the queue. Returns the remaining messages that
738 * were not added.
739 *
740 * **When to use**
741 *
742 * Use when producers can submit a batch at once and need to know which messages
743 * did not fit under the queue's capacity strategy.
744 *
745 * **Details**
746 *
747 * For bounded queues, this operation may suspend if the queue doesn't have
748 * enough capacity. The operation returns an array of messages that couldn't
749 * be added (empty array means all messages were successfully added).
750 *
751 * **Example** (Offering multiple values)
752 *
753 * ```ts import.meta.vitest
754 * import { Effect, Queue } from "effect"
755 *
756 * const program = Effect.gen(function*() {
757 * const queue = yield* Queue.dropping<number>(3)
758 *
759 * // Try to add more messages than capacity without suspending
760 * const remaining1 = yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
761 * return remaining1
762 * })
763 *
764 * await Effect.runPromise(program) // => [4, 5]
765 * ```
766 *
767 * @category offering
768 * @since 2.0.0
769 */
770export const offerAll: {
771 <A>(messages: Iterable<A>): <E>(self: Enqueue<A, E>) => Effect<Array<A>>
772 <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>
773} = dual(2, <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>> =>
774 // Admitting what fits and registering the rest are one step, so no take can
775 // free capacity between them
776 internalEffect.callback<Array<A>>((resume) => {
777 const remaining = offerAllUnsafe(self, messages)
778 if (remaining.length === 0 || self.strategy === "dropping") {
779 return resume(core.exitSucceed(remaining))
780 }
781 return waitToOffer(self, { _tag: "Array", remaining, offset: 0, resume })
782 }))
783
784/**
785 * Adds multiple messages to the queue synchronously. Returns the remaining messages that
786 * were not added.
787 *
788 * **When to use**
789 *
790 * Use when queue internals or a performance boundary need a synchronous batch
791 * offer and can handle any messages that do not fit.
792 *
793 * **Gotchas**
794 *
795 * This is an unsafe operation that directly modifies the queue without Effect wrapping.
796 *
797 * **Example** (Offering multiple values synchronously)
798 *
799 * ```ts import.meta.vitest
800 * import { Effect, Queue } from "effect"
801 *
802 * // Create a bounded queue and use unsafe API
803 * const program = Effect.gen(function*() {
804 * const queue = yield* Queue.bounded<number>(3)
805 *
806 * // Try to add 5 messages to capacity-3 queue using unsafe API
807 * const remaining = Queue.offerAllUnsafe(queue, [1, 2, 3, 4, 5])
808 *
809 * // Check what's in the queue
810 * const size = Queue.sizeUnsafe(queue)
811 * return { remaining, size }
812 * })
813 *
814 * await Effect.runPromise(program) // => { remaining: [4, 5], size: 3 }
815 * ```
816 *
817 * @category offering
818 * @since 4.0.0
819 */
820export const offerAllUnsafe = <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Array<A> => {
821 if (self.state._tag !== "Open") {
822 return Arr.fromIterable(messages)
823 } else if (
824 self.capacity === Number.POSITIVE_INFINITY ||
825 self.strategy === "sliding"
826 ) {
827 MutableList.appendAll(self.messages, messages)
828 if (self.strategy === "sliding") {
829 MutableList.takeN(self.messages, self.messages.length - self.capacity)
830 }
831 scheduleReleaseTaker(self as Queue<A, E>)
832 return []
833 }
834 const free = self.capacity <= 0
835 ? self.state.takers.size
836 : self.capacity - self.messages.length
837 if (free === 0) {
838 return Arr.fromIterable(messages)
839 }
840 const remaining: Array<A> = []
841 let i = 0
842 for (const message of messages) {
843 if (i < free) {
844 MutableList.append(self.messages, message)
845 } else {
846 remaining.push(message)
847 }
848 i++
849 }
850 scheduleReleaseTaker(self as Queue<A, E>)
851 return remaining
852}
853
854/**
855 * Fails the queue with an error. If the queue is already done, `false` is
856 * returned.
857 *
858 * **Example** (Failing queues with an error)
859 *
860 * ```ts import.meta.vitest
861 * import { Effect, Exit, Queue } from "effect"
862 *
863 * const program = Effect.gen(function*() {
864 * const queue = yield* Queue.bounded<number, string>(10)
865 *
866 * // Fail the queue with an error
867 * const failed = yield* Queue.fail(queue, "Something went wrong")
868 *
869 * // Taking from the failed queue fails with the error
870 * const exit = yield* Effect.exit(Queue.take(queue))
871 * return [failed, exit]
872 * })
873 *
874 * await Effect.runPromise(program) // => [true, Exit.fail("Something went wrong")]
875 * ```
876 *
877 * @category completion
878 * @since 4.0.0
879 */
880export const fail: {
881 <E>(error: E): <A>(self: Enqueue<A, E>) => Effect<boolean>
882 <A, E>(self: Enqueue<A, E>, error: E): Effect<boolean>
883} = dual(2, <A, E>(self: Enqueue<A, E>, error: E): Effect<boolean> => failCause(self, core.causeFail(error)))
884
885/**
886 * Fails the queue with a cause. If the queue is already done, `false` is
887 * returned.
888 *
889 * **Example** (Failing queues with a cause)
890 *
891 * ```ts import.meta.vitest
892 * import { Cause, Effect, Exit, Queue } from "effect"
893 *
894 * const program = Effect.gen(function*() {
895 * const queue = yield* Queue.bounded<number, string>(10)
896 *
897 * // Create a cause and fail the queue
898 * const cause = Cause.fail("Queue processing failed")
899 * const failed = yield* Queue.failCause(queue, cause)
900 *
901 * // The queue is now done with the specified failure cause
902 * const exit = yield* Effect.exit(Queue.take(queue))
903 * return [failed, exit]
904 * })
905 *
906 * await Effect.runPromise(program) // => [true, Exit.failCause(Cause.fail("Queue processing failed"))]
907 * ```
908 *
909 * @category completion
910 * @since 4.0.0
911 */
912export const failCause: {
913 <E>(cause: Cause<E>): <A>(self: Enqueue<A, E>) => Effect<boolean>
914 <A, E>(self: Enqueue<A, E>, cause: Cause<E>): Effect<boolean>
915} = dual(
916 2,
917 <A, E>(self: Enqueue<A, E>, cause: Cause<E>): Effect<boolean> =>
918 internalEffect.sync(() => failCauseUnsafe(self, cause))
919)
920
921/**
922 * Fails the queue with a cause synchronously. If the queue is already done, `false` is
923 * returned.
924 *
925 * **When to use**
926 *
927 * Use when queue completion must be driven from synchronous internals while
928 * preserving the full failure `Cause`.
929 *
930 * **Gotchas**
931 *
932 * This is an unsafe operation that directly modifies the queue without Effect wrapping.
933 *
934 * **Example** (Failing queues with a cause synchronously)
935 *
936 * ```ts import.meta.vitest
937 * import { Cause, Effect, Exit, Queue } from "effect"
938 *
939 * const program = Effect.gen(function*() {
940 * const queue = yield* Queue.bounded<number, string>(10)
941 *
942 * // Create a cause and fail the queue synchronously
943 * const cause = Cause.fail("Processing error")
944 * const failed = Queue.failCauseUnsafe(queue, cause)
945 *
946 * // The queue is now done with the specified failure cause
947 * const exit = Queue.takeUnsafe(queue)
948 * return [failed, exit]
949 * })
950 *
951 * await Effect.runPromise(program) // => [true, Exit.failCause(Cause.fail("Processing error"))]
952 * ```
953 *
954 * @category completion
955 * @since 4.0.0
956 */
957export const failCauseUnsafe = <A, E>(self: Enqueue<A, E>, cause: Cause<E>): boolean => {
958 if (self.state._tag !== "Open") {
959 return false
960 }
961 const exit = core.exitFailCause(cause)
962 const fail = internalEffect.exitZipRight(exit, exitFailDone) as Failure<never, E>
963 if (
964 self.state.offers.size === 0 &&
965 self.messages.length === 0
966 ) {
967 finalize(self, fail)
968 return true
969 }
970 self.state = { ...self.state, _tag: "Closing", exit: fail }
971 // Batch takers waiting on a minimum can now drain the remainder.
972 scheduleReleaseTaker(self)
973 return true
974}
975
976/**
977 * Signals queue completion.
978 *
979 * **When to use**
980 *
981 * Use to stop accepting new offers while allowing already queued messages to be
982 * consumed.
983 *
984 * **Details**
985 *
986 * Returns `false` if the queue is already done.
987 *
988 * **Example** (Ending queues)
989 *
990 * ```ts import.meta.vitest
991 * import { Cause, Effect, Queue } from "effect"
992 *
993 * const program = Effect.gen(function*() {
994 * const queue = yield* Queue.bounded<number, Cause.Done>(10)
995 *
996 * // Add some messages
997 * yield* Queue.offer(queue, 1)
998 * yield* Queue.offer(queue, 2)
999 *
1000 * // Signal completion - no more messages will be accepted
1001 * const ended = yield* Queue.end(queue)
1002 *
1003 * // Trying to offer more messages will return false
1004 * const offerResult = yield* Queue.offer(queue, 3)
1005 *
1006 * // But we can still take existing messages
1007 * const message = yield* Queue.take(queue)
1008 * return [ended, offerResult, message]
1009 * })
1010 *
1011 * await Effect.runPromise(program) // => [true, false, 1]
1012 * ```
1013 *
1014 * @category completion
1015 * @since 4.0.0
1016 */
1017export const end = <A, E>(self: Enqueue<A, E | Done>): Effect<boolean> => failCause(self, core.causeFail(core.Done()))
1018
1019/**
1020 * Signals queue completion synchronously.
1021 *
1022 * **When to use**
1023 *
1024 * Use when implementing low-level queue integrations that must complete a queue
1025 * without wrapping the operation in `Effect`.
1026 *
1027 * **Details**
1028 *
1029 * Returns `false` if the queue is already done.
1030 *
1031 * **Gotchas**
1032 *
1033 * This is an unsafe operation that directly modifies the queue without Effect wrapping.
1034 *
1035 * **Example** (Ending queues synchronously)
1036 *
1037 * ```ts import.meta.vitest
1038 * import { Cause, Effect, Queue } from "effect"
1039 *
1040 * // Create a queue and use unsafe operations
1041 * const program = Effect.gen(function*() {
1042 * const queue = yield* Queue.bounded<number, Cause.Done>(10)
1043 *
1044 * // Add some messages
1045 * Queue.offerUnsafe(queue, 1)
1046 * Queue.offerUnsafe(queue, 2)
1047 *
1048 * // End the queue synchronously
1049 * const ended = Queue.endUnsafe(queue)
1050 *
1051 * // Existing messages can still be consumed while the queue is closing
1052 * const states = [queue.state._tag]
1053 *
1054 * Queue.takeUnsafe(queue)
1055 * Queue.takeUnsafe(queue)
1056 *
1057 * // After buffered messages are consumed, the queue is done
1058 * states.push(queue.state._tag)
1059 * return { ended, states }
1060 * })
1061 *
1062 * await Effect.runPromise(program) // => { ended: true, states: ["Closing", "Done"] }
1063 * ```
1064 *
1065 * @category completion
1066 * @since 4.0.0
1067 */
1068export const endUnsafe = <A, E>(self: Enqueue<A, E | Done>) => failCauseUnsafe(self, core.causeFail(core.Done()))
1069
1070/**
1071 * Interrupts the queue gracefully, transitioning it to a closing state.
1072 *
1073 * **Details**
1074 *
1075 * This operation stops accepting new offers but allows existing messages to be consumed.
1076 * Once all messages are drained, the queue transitions to the Done state with an interrupt cause.
1077 *
1078 * **Example** (Interrupting queues gracefully)
1079 *
1080 * ```ts import.meta.vitest
1081 * import { Cause, Effect, Queue } from "effect"
1082 *
1083 * const program = Effect.gen(function*() {
1084 * const queue = yield* Queue.bounded<number>(10)
1085 *
1086 * // Add some messages
1087 * yield* Queue.offer(queue, 1)
1088 * yield* Queue.offer(queue, 2)
1089 *
1090 * // Interrupt gracefully - no more offers accepted, but messages can be consumed
1091 * const interrupted = yield* Queue.interrupt(queue)
1092 *
1093 * // Trying to offer more messages will return false
1094 * const offerResult = yield* Queue.offer(queue, 3)
1095 *
1096 * // But we can still take existing messages
1097 * const message1 = yield* Queue.take(queue)
1098 *
1099 * const message2 = yield* Queue.take(queue)
1100 *
1101 * // After all messages are consumed, queue is done
1102 * const isDone = queue.state._tag === "Done"
1103 * return { interrupted, offerResult, messages: [message1, message2], isDone }
1104 * })
1105 *
1106 * await Effect.runPromise(program) // => { interrupted: true, offerResult: false, messages: [1, 2], isDone: true }
1107 * ```
1108 *
1109 * @category completion
1110 * @since 4.0.0
1111 */
1112export const interrupt = <A, E>(self: Enqueue<A, E>): Effect<boolean> =>
1113 core.withFiber((fiber) => failCause(self, internalEffect.causeInterrupt(fiber.id)))
1114
1115/**
1116 * Shuts down the queue immediately, discarding buffered messages and resuming
1117 * pending operations.
1118 *
1119 * **Details**
1120 *
1121 * Returns `true` when the queue is shut down by this call, or `false` when it
1122 * has already been shut down or completed.
1123 *
1124 * **Example** (Shutting down queues)
1125 *
1126 * ```ts import.meta.vitest
1127 * import { Effect, Queue } from "effect"
1128 *
1129 * const program = Effect.gen(function*() {
1130 * const queue = yield* Queue.bounded<number>(2)
1131 *
1132 * // Add messages
1133 * yield* Queue.offer(queue, 1)
1134 * yield* Queue.offer(queue, 2)
1135 *
1136 * // Shutdown clears buffered messages and prevents further offers
1137 * const wasShutdown = yield* Queue.shutdown(queue)
1138 *
1139 * // Queue is now done and cleared
1140 * const size = yield* Queue.size(queue)
1141 * return { wasShutdown, size }
1142 * })
1143 *
1144 * await Effect.runPromise(program) // => { wasShutdown: true, size: 0 }
1145 * ```
1146 *
1147 * @see {@link shutdownUnsafe} for synchronous shutdown
1148 * @category completion
1149 * @since 2.0.0
1150 */
1151export const shutdown = <A, E>(self: Enqueue<A, E>): Effect<boolean> => internalEffect.sync(() => shutdownUnsafe(self))
1152
1153/**
1154 * Shuts down the queue synchronously, discarding buffered messages and resuming
1155 * pending operations.
1156 *
1157 * **When to use**
1158 *
1159 * Use when a synchronous callback must discard buffered messages and settle
1160 * pending queue operations before returning.
1161 *
1162 * **Details**
1163 *
1164 * An open queue completes with an interruption. A queue already closing retains
1165 * its completion cause. Call `failCauseUnsafe` first to shut down with a specific
1166 * failure. Returns `true` when the queue is shut down by this call, or `false`
1167 * when it has already been shut down or completed.
1168 *
1169 * @see {@link shutdown} for the effectful variant
1170 * @see {@link failCauseUnsafe} to set a failure before discarding buffered messages
1171 * @category completion
1172 * @since 4.0.0
1173 */
1174export const shutdownUnsafe = <A, E>(self: Enqueue<A, E>): boolean => {
1175 if (self.state._tag === "Done") {
1176 return false
1177 }
1178 MutableList.clear(self.messages)
1179 const offers = self.state.offers
1180 finalize(self, self.state._tag === "Open" ? exitInterrupt : self.state.exit)
1181 for (const entry of offers) {
1182 resumeUnoffered(entry)
1183 }
1184 return true
1185}
1186
1187/**
1188 * Takes and returns all currently buffered messages without waiting for more.
1189 *
1190 * **Details**
1191 *
1192 * Returns an empty array when the queue is empty or has completed normally. If
1193 * the queue has failed, the effect fails with the queue's error.
1194 *
1195 * **Example** (Clearing queued values)
1196 *
1197 * ```ts import.meta.vitest
1198 * import { Effect, Queue } from "effect"
1199 *
1200 * const program = Effect.gen(function*() {
1201 * const queue = yield* Queue.bounded<number>(10)
1202 *
1203 * // Add several messages
1204 * yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
1205 *
1206 * // Clear all messages from the queue
1207 * const messages = yield* Queue.clear(queue)
1208 *
1209 * // Queue is now empty
1210 * const size = yield* Queue.size(queue)
1211 *
1212 * // Clearing empty queue returns empty array
1213 * const empty = yield* Queue.clear(queue)
1214 * return { messages, size, empty }
1215 * })
1216 *
1217 * await Effect.runPromise(program) // => { messages: [1, 2, 3, 4, 5], size: 0, empty: [] }
1218 * ```
1219 *
1220 * @category taking
1221 * @since 4.0.0
1222 */
1223export const clear = <A, E>(self: Dequeue<A, E>): Effect<Array<A>, Pull.ExcludeDone<E>> =>
1224 internalEffect.suspend(() => {
1225 if (self.state._tag === "Done") {
1226 if (Pull.isDoneCause(self.state.exit.cause)) {
1227 return internalEffect.succeed([])
1228 }
1229 return self.state.exit
1230 }
1231 const messages = takeAllUnsafe(self)
1232 releaseCapacity(self)
1233 return internalEffect.succeed(messages)
1234 })
1235
1236/**
1237 * Takes all currently available messages, waiting until at least one message
1238 * is available when the queue is empty.
1239 *
1240 * **When to use**
1241 *
1242 * Use when consumers should process the next non-empty batch of buffered
1243 * messages instead of repeatedly taking one message at a time.
1244 *
1245 * **Details**
1246 *
1247 * Returns a non-empty array. If the queue completes or fails before a message
1248 * can be taken, the effect fails with the queue's terminal error.
1249 *
1250 * **Example** (Taking all available values)
1251 *
1252 * ```ts import.meta.vitest
1253 * import { Cause, Effect, Queue } from "effect"
1254 *
1255 * const program = Effect.gen(function*() {
1256 * const queue = yield* Queue.bounded<number, Cause.Done>(5)
1257 *
1258 * // Add several messages
1259 * yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
1260 *
1261 * // Take all available messages
1262 * const messages1 = yield* Queue.takeAll(queue)
1263 * return messages1
1264 * })
1265 *
1266 * await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
1267 * ```
1268 *
1269 * @category taking
1270 * @since 2.0.0
1271 */
1272export const takeAll = <A, E>(self: Dequeue<A, E>): Effect<Arr.NonEmptyArray<A>, E> =>
1273 takeBetween(self, 1, Number.POSITIVE_INFINITY) as any
1274
1275/**
1276 * Takes all messages from the queue, until the queue has errored or is done.
1277 *
1278 * **Example** (Collecting values until completion)
1279 *
1280 * ```ts import.meta.vitest
1281 * import { Cause, Effect, Queue } from "effect"
1282 *
1283 * const program = Effect.gen(function*() {
1284 * const queue = yield* Queue.bounded<number, Cause.Done>(5)
1285 *
1286 * // Add several messages
1287 * yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
1288 * yield* Queue.end(queue)
1289 *
1290 * // Collect all available messages
1291 * return yield* Queue.collect(queue)
1292 * })
1293 *
1294 * await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
1295 * ```
1296 *
1297 * @category taking
1298 * @since 4.0.0
1299 */
1300export const collect = <A, E>(self: Dequeue<A, E | Done>): Effect<Array<A>, Pull.ExcludeDone<E>> =>
1301 internalEffect.suspend(() => {
1302 const out = Arr.empty<A>()
1303 return internalEffect.as(
1304 Pull.catchDone(
1305 internalEffect.whileLoop({
1306 while: constTrue,
1307 body: constant(takeAll(self)),
1308 step(items: Arr.NonEmptyArray<A>) {
1309 for (let i = 0; i < items.length; i++) {
1310 out.push(items[i])
1311 }
1312 }
1313 }),
1314 () => internalEffect.void
1315 ),
1316 out
1317 )
1318 }) as any
1319
1320/**
1321 * Takes up to `n` messages from the queue.
1322 *
1323 * **Details**
1324 *
1325 * The operation may wait until enough messages are available to satisfy the
1326 * queue's batching rules. Finite fractional values of `n` are rounded down.
1327 * If `n` is `NaN` or non-positive, it succeeds with an empty array. If the
1328 * queue is closing, drains the currently available messages even when fewer
1329 * than `n` are available. Once the queue is done, the effect fails with the
1330 * queue's terminal error.
1331 *
1332 * **Example** (Taking a fixed number of values)
1333 *
1334 * ```ts import.meta.vitest
1335 * import { Cause, Effect, Queue } from "effect"
1336 *
1337 * const program = Effect.gen(function*() {
1338 * const queue = yield* Queue.bounded<number, Cause.Done>(10)
1339 *
1340 * // Add several messages
1341 * yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7])
1342 *
1343 * // Take exactly 3 messages
1344 * const first3 = yield* Queue.takeN(queue, 3)
1345 *
1346 * // Take exactly 2 more messages
1347 * const next2 = yield* Queue.takeN(queue, 2)
1348 *
1349 * // Take remaining messages
1350 * const remaining = yield* Queue.takeN(queue, 2)
1351 * return [first3, next2, remaining]
1352 * })
1353 *
1354 * await Effect.runPromise(program) // => [[1, 2, 3], [4, 5], [6, 7]]
1355 * ```
1356 *
1357 * @category taking
1358 * @since 2.0.0
1359 */
1360export const takeN: {
1361 (n: number): <A, E>(self: Dequeue<A, E>) => Effect<Array<A>, E>
1362 <A, E>(self: Dequeue<A, E>, n: number): Effect<Array<A>, E>
1363} = dual(2, <A, E>(self: Dequeue<A, E>, n: number): Effect<Array<A>, E> => takeBetween(self, n, n))
1364
1365/**
1366 * Takes between `min` and `max` messages from the queue.
1367 *
1368 * **Details**
1369 *
1370 * The operation waits when fewer than the required minimum messages are
1371 * available. It returns at most `max` messages. Finite fractional bounds are
1372 * rounded down, while `NaN` and non-positive bounds are treated as `0`. If the
1373 * queue is closing, drains the currently available messages even when fewer
1374 * than `min` are available. Once the queue is done, the effect fails with the
1375 * queue's terminal error.
1376 *
1377 * **Example** (Taking a bounded batch of values)
1378 *
1379 * ```ts import.meta.vitest
1380 * import { Cause, Effect, Queue } from "effect"
1381 *
1382 * const program = Effect.gen(function*() {
1383 * const queue = yield* Queue.bounded<number>(10)
1384 *
1385 * // Add several messages
1386 * yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8])
1387 *
1388 * // Take between 2 and 5 messages
1389 * const batch1 = yield* Queue.takeBetween(queue, 2, 5)
1390 *
1391 * // Take between 1 and 10 messages (but only 3 remain)
1392 * const batch2 = yield* Queue.takeBetween(queue, 1, 10)
1393 *
1394 * // No more messages available, will wait or return done
1395 * // const batch3 = yield* Queue.takeBetween(queue, 1, 3)
1396 * return [batch1, batch2]
1397 * })
1398 *
1399 * await Effect.runPromise(program) // => [[1, 2, 3, 4, 5], [6, 7, 8]]
1400 * ```
1401 *
1402 * @category taking
1403 * @since 2.0.0
1404 */
1405export const takeBetween: {
1406 (min: number, max: number): <A, E>(self: Dequeue<A, E>) => Effect<Array<A>, E>
1407 <A, E>(self: Dequeue<A, E>, min: number, max: number): Effect<Array<A>, E>
1408} = dual(3, <A, E>(self: Dequeue<A, E>, min: number, max: number): Effect<Array<A>, E> => {
1409 min = Count.normalize(min)
1410 max = Count.normalize(max)
1411 return internalEffect.suspend(() =>
1412 takeBetweenUnsafe(self, min, max) ??
1413 internalEffect.andThen(awaitTake(self, () => canTake(self, min)), takeBetween(self, min, max))
1414 )
1415})
1416
1417/**
1418 * Takes a single message from the queue, or wait for a message to be
1419 * available.
1420 *
1421 * **Details**
1422 *
1423 * If the queue is done, it will fail with `Done`. If the
1424 * queue fails, the Effect will fail with the error.
1425 *
1426 * **Example** (Taking one value)
1427 *
1428 * ```ts import.meta.vitest
1429 * import { Cause, Effect, Exit, Queue } from "effect"
1430 *
1431 * const program = Effect.gen(function*() {
1432 * const queue = yield* Queue.bounded<string, Cause.Done>(3)
1433 *
1434 * // Add some messages
1435 * yield* Queue.offer(queue, "first")
1436 * yield* Queue.offer(queue, "second")
1437 *
1438 * // Take messages one by one
1439 * const msg1 = yield* Queue.take(queue)
1440 * const msg2 = yield* Queue.take(queue)
1441 *
1442 * // End the queue
1443 * yield* Queue.end(queue)
1444 *
1445 * // Taking from an ended queue fails with Done
1446 * const result = yield* Effect.exit(Queue.take(queue))
1447 * return [[msg1, msg2], result]
1448 * })
1449 *
1450 * await Effect.runPromise(program) // => [["first", "second"], Exit.fail(Cause.Done())]
1451 * ```
1452 *
1453 * @category taking
1454 * @since 2.0.0
1455 */
1456export const take = <A, E>(self: Dequeue<A, E>): Effect<A, E> =>
1457 internalEffect.suspend(() =>
1458 takeUnsafe(self) ?? internalEffect.andThen(awaitTake(self, () => canTake(self, 1)), take(self))
1459 )
1460
1461/**
1462 * Attempts to take one item from the queue without waiting.
1463 *
1464 * **Details**
1465 *
1466 * Returns `Option.some` when an item is immediately available. Returns
1467 * `Option.none` when no item is available, when the queue is done, or when the
1468 * immediate take observes a queue failure.
1469 *
1470 * **Example** (Polling without blocking)
1471 *
1472 * ```ts import.meta.vitest
1473 * import { Effect, Option, Queue } from "effect"
1474 *
1475 * const program = Effect.gen(function*() {
1476 * const queue = yield* Queue.bounded<number>(10)
1477 *
1478 * // Poll returns Option.none if empty
1479 * const maybe1 = yield* Queue.poll(queue)
1480 *
1481 * // Add an item
1482 * yield* Queue.offer(queue, 42)
1483 *
1484 * // Poll returns Option.some with the item
1485 * const maybe2 = yield* Queue.poll(queue)
1486 * return [maybe1, maybe2]
1487 * })
1488 *
1489 * await Effect.runPromise(program) // => [Option.none(), Option.some(42)]
1490 * ```
1491 *
1492 * @category taking
1493 * @since 2.0.0
1494 */
1495export const poll = <A, E>(self: Dequeue<A, E>): Effect<Option.Option<A>> =>
1496 internalEffect.suspend(() => {
1497 const result = takeUnsafe(self)
1498 if (result === undefined) {
1499 return internalEffect.succeed(Option.none())
1500 }
1501 if (result._tag === "Success") {
1502 return internalEffect.succeed(Option.some(result.value))
1503 }
1504 return internalEffect.succeed(Option.none())
1505 })
1506
1507/**
1508 * Peeks at the next item without removing it.
1509 *
1510 * **Details**
1511 *
1512 * Blocks until an item is available. If the queue is done or fails, the error is propagated.
1513 *
1514 * **Example** (Peeking at the next value)
1515 *
1516 * ```ts import.meta.vitest
1517 * import { Cause, Effect, Queue } from "effect"
1518 *
1519 * const program = Effect.gen(function*() {
1520 * const queue = yield* Queue.bounded<number>(10)
1521 * yield* Queue.offer(queue, 42)
1522 *
1523 * // Peek at the next item without removing it
1524 * const item = yield* Queue.peek(queue)
1525 * return item
1526 * })
1527 *
1528 * await Effect.runPromise(program) // => 42
1529 * ```
1530 *
1531 * @category taking
1532 * @since 4.0.0
1533 */
1534export const peek = <A, E>(self: Dequeue<A, E>): Effect<A, E> =>
1535 internalEffect.suspend(() => {
1536 if (self.state._tag === "Done") {
1537 return self.state.exit
1538 }
1539 if (self.messages.length > 0 && self.messages.head) {
1540 return internalEffect.succeed(self.messages.head.array[self.messages.head.offset])
1541 }
1542 // Pending offers on a rendezvous queue are not visible to peek.
1543 return internalEffect.andThen(awaitTake(self, () => self.messages.length > 0), peek(self))
1544 })
1545
1546/**
1547 * Attempts to take one message from the queue synchronously.
1548 *
1549 * **When to use**
1550 *
1551 * Use when polling queue internals must not suspend or register a waiting taker,
1552 * and `undefined` is an acceptable result for an empty queue.
1553 *
1554 * **Details**
1555 *
1556 * Returns an `Exit` for an immediately available message or for the queue's
1557 * terminal state. Returns `undefined` when no message is immediately available.
1558 * This operation does not wait or register a taker.
1559 *
1560 * **Example** (Taking one value synchronously)
1561 *
1562 * ```ts import.meta.vitest
1563 * import { Effect, Exit, Queue } from "effect"
1564 *
1565 * // Create a queue and use unsafe operations
1566 * const program = Effect.gen(function*() {
1567 * const queue = yield* Queue.bounded<number>(10)
1568 *
1569 * // Add some messages
1570 * Queue.offerUnsafe(queue, 1)
1571 * Queue.offerUnsafe(queue, 2)
1572 *
1573 * // Take a message synchronously
1574 * const result1 = Queue.takeUnsafe(queue)
1575 *
1576 * const result2 = Queue.takeUnsafe(queue)
1577 *
1578 * // No more messages - returns undefined
1579 * const result3 = Queue.takeUnsafe(queue)
1580 * return [result1, result2, result3]
1581 * })
1582 *
1583 * await Effect.runPromise(program) // => [Exit.succeed(1), Exit.succeed(2), undefined]
1584 * ```
1585 *
1586 * @category taking
1587 * @since 4.0.0
1588 */
1589export const takeUnsafe = <A, E>(self: Dequeue<A, E>): Exit<A, E> | undefined => {
1590 if (self.state._tag === "Done") {
1591 return self.state.exit
1592 }
1593 if (self.messages.length > 0) {
1594 const message = MutableList.take(self.messages)!
1595 releaseCapacity(self)
1596 return core.exitSucceed(message)
1597 } else if (self.capacity <= 0 && self.state.offers.size > 0) {
1598 const message = takeOfferUnsafe(self.state.offers)
1599 releaseCapacity(self)
1600 return core.exitSucceed(message)
1601 }
1602 return undefined
1603}
1604
1605/**
1606 * Manually releases current queue takers synchronously.
1607 *
1608 * **When to use**
1609 *
1610 * Use when synchronous offers should release waiting consumers immediately
1611 * instead of waiting for the scheduled release task.
1612 *
1613 * **Details**
1614 *
1615 * This immediately runs the queue's taker-release pass instead of waiting for
1616 * its scheduled task. It does not complete the queue or resume fibers waiting
1617 * on `Queue.await`.
1618 *
1619 * **Example** (Releasing a waiting taker synchronously)
1620 *
1621 * ```ts import.meta.vitest
1622 * import { Effect, Fiber, Queue } from "effect"
1623 *
1624 * const program = Effect.gen(function*() {
1625 * const queue = yield* Queue.unbounded<number>()
1626 * const taker = yield* Queue.take(queue).pipe(Effect.forkChild)
1627 * yield* Effect.yieldNow
1628 *
1629 * Queue.offerUnsafe(queue, 1)
1630 * Queue.flushUnsafe(queue)
1631 *
1632 * return yield* Fiber.join(taker)
1633 * })
1634 *
1635 * await Effect.runPromise(program) // => 1
1636 * ```
1637 *
1638 * @category offering
1639 * @since 4.0.0
1640 */
1641export const flushUnsafe = <A, E>(self: Enqueue<A, E>): void => releaseTakers(self)
1642
1643/**
1644 * Manually releases current queue takers.
1645 *
1646 * **When to use**
1647 *
1648 * Use when synchronous offers should release waiting consumers through an
1649 * `Effect` instead of waiting for the scheduled release task.
1650 *
1651 * **Details**
1652 *
1653 * This immediately runs the queue's taker-release pass instead of waiting for
1654 * its scheduled task. It does not complete the queue or resume fibers waiting
1655 * on `Queue.await`.
1656 *
1657 * **Example** (Releasing a waiting taker)
1658 *
1659 * ```ts import.meta.vitest
1660 * import { Effect, Fiber, Queue } from "effect"
1661 *
1662 * const program = Effect.gen(function*() {
1663 * const queue = yield* Queue.unbounded<number>()
1664 * const taker = yield* Queue.take(queue).pipe(Effect.forkChild)
1665 * yield* Effect.yieldNow
1666 *
1667 * Queue.offerUnsafe(queue, 1)
1668 * yield* Queue.flush(queue)
1669 *
1670 * return yield* Fiber.join(taker)
1671 * })
1672 *
1673 * await Effect.runPromise(program) // => 1
1674 * ```
1675 *
1676 * @category offering
1677 * @since 4.0.0
1678 */
1679export const flush = <A, E>(self: Enqueue<A, E>): Effect<void> => internalEffect.sync(() => flushUnsafe(self))
1680
1681const await_ = <A, E>(self: Dequeue<A, E>): Effect<void, Exclude<E, Done>> =>
1682 internalEffect.callback<void, Exclude<E, Done>>((resume) => {
1683 const awaiter = (effect: Effect<void, E>) => resume(Pull.catchDone(effect, () => internalEffect.exitVoid))
1684 if (self.state._tag === "Done") {
1685 return awaiter(self.state.exit)
1686 }
1687 self.state.awaiters.add(awaiter)
1688 return internalEffect.sync(() => {
1689 if (self.state._tag !== "Done") {
1690 self.state.awaiters.delete(awaiter)
1691 }
1692 })
1693 })
1694
1695export {
1696 /**
1697 * Waits until a queue reaches the `Done` state.
1698 *
1699 * **When to use**
1700 *
1701 * Use to suspend a fiber until no further values can be taken from the queue
1702 * and its terminal outcome is known.
1703 *
1704 * **Details**
1705 *
1706 * The effect succeeds with `void` for normal `Done` completion. Other
1707 * terminal causes are preserved, so failures and interruptions complete this
1708 * effect with the same terminal outcome.
1709 *
1710 * **Gotchas**
1711 *
1712 * A queue can be closing before it is done. `await` resumes at `Done`, not at
1713 * the first completion signal, so buffered messages may need to be drained
1714 * first.
1715 *
1716 * @see {@link end} for signaling normal completion while preserving buffered messages for consumers
1717 * @see {@link fail} for signaling an error while preserving buffered messages for consumers
1718 * @see {@link interrupt} for graceful interruption after buffered messages are drained
1719 * @see {@link shutdown} for immediately discarding buffered messages and resuming pending operations
1720 *
1721 * @category completion
1722 * @since 4.0.0
1723 */
1724 await_ as await
1725}
1726
1727/**
1728 * Returns the current number of buffered messages in the queue.
1729 *
1730 * **Details**
1731 *
1732 * After `end`, a queue remains `Closing` while buffered messages are drained,
1733 * and its size continues to include those messages. A `Done` queue reports a
1734 * size of `0`.
1735 *
1736 * **Example** (Checking queue size)
1737 *
1738 * ```ts import.meta.vitest
1739 * import { Cause, Effect, Queue } from "effect"
1740 *
1741 * const program = Effect.gen(function*() {
1742 * const queue = yield* Queue.bounded<number, Cause.Done>(10)
1743 *
1744 * // Check size of empty queue
1745 * const size1 = yield* Queue.size(queue)
1746 *
1747 * // Add some messages
1748 * yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
1749 *
1750 * // Check size after adding messages
1751 * const size2 = yield* Queue.size(queue)
1752 *
1753 * // End the queue
1754 * yield* Queue.end(queue)
1755 *
1756 * // Ending retains the buffered size while the queue is Closing
1757 * const size3 = yield* Queue.size(queue)
1758 * return [size1, size2, size3]
1759 * })
1760 *
1761 * await Effect.runPromise(program) // => [0, 5, 5]
1762 * ```
1763 *
1764 * @category sizes
1765 * @since 2.0.0
1766 */
1767export const size = <A, E>(self: Dequeue<A, E>): Effect<number> => internalEffect.sync(() => sizeUnsafe(self))
1768
1769/**
1770 * Checks whether the queue is full.
1771 *
1772 * **Example** (Checking if queues are full)
1773 *
1774 * ```ts import.meta.vitest
1775 * import { Cause, Effect, Queue } from "effect"
1776 *
1777 * const program = Effect.gen(function*() {
1778 * const queue = yield* Queue.bounded<number, Cause.Done>(3)
1779 *
1780 * const before = yield* Queue.isFull(queue)
1781 *
1782 * // Add some messages
1783 * yield* Queue.offerAll(queue, [1, 2, 3])
1784 *
1785 * const after = yield* Queue.isFull(queue)
1786 * return [before, after]
1787 * })
1788 *
1789 * await Effect.runPromise(program) // => [false, true]
1790 * ```
1791 *
1792 * @category predicates
1793 * @since 2.0.0
1794 */
1795export const isFull = <A, E>(self: Dequeue<A, E>): Effect<boolean> => internalEffect.sync(() => isFullUnsafe(self))
1796
1797/**
1798 * Returns the current number of buffered messages in the queue synchronously.
1799 *
1800 * **When to use**
1801 *
1802 * Use when you need an immediate `Queue` size snapshot for diagnostics or
1803 * internals and do not need the read wrapped in `Effect`.
1804 *
1805 * **Details**
1806 *
1807 * After `endUnsafe`, a queue remains `Closing` while buffered messages are
1808 * drained, and its size continues to include those messages. A `Done` queue
1809 * reports a size of `0`. This unsafe operation reads the queue state directly
1810 * without Effect wrapping.
1811 *
1812 * **Example** (Checking queue size synchronously)
1813 *
1814 * ```ts import.meta.vitest
1815 * import { Cause, Effect, Queue } from "effect"
1816 *
1817 * const program = Effect.gen(function*() {
1818 * const queue = yield* Queue.bounded<number, Cause.Done>(10)
1819 *
1820 * // Check size of empty queue
1821 * const size1 = Queue.sizeUnsafe(queue)
1822 *
1823 * // Add some messages
1824 * Queue.offerUnsafe(queue, 1)
1825 * Queue.offerUnsafe(queue, 2)
1826 * Queue.offerUnsafe(queue, 3)
1827 *
1828 * // Check size after adding messages
1829 * const size2 = Queue.sizeUnsafe(queue)
1830 *
1831 * // End the queue
1832 * Queue.endUnsafe(queue)
1833 *
1834 * // Ending retains the buffered size while the queue is Closing
1835 * const size3 = Queue.sizeUnsafe(queue)
1836 * return [size1, size2, size3]
1837 * })
1838 *
1839 * await Effect.runPromise(program) // => [0, 3, 3]
1840 * ```
1841 *
1842 * @category sizes
1843 * @since 4.0.0
1844 */
1845export const sizeUnsafe = <A, E>(self: Dequeue<A, E>): number => self.state._tag === "Done" ? 0 : self.messages.length
1846
1847/**
1848 * Checks whether the queue is full synchronously.
1849 *
1850 * **When to use**
1851 *
1852 * Use when an immediate `Queue` capacity snapshot is needed outside effectful
1853 * code and racing queue changes are acceptable.
1854 *
1855 * **Example** (Checking fullness synchronously)
1856 *
1857 * ```ts import.meta.vitest
1858 * import { Cause, Effect, Queue } from "effect"
1859 *
1860 * const program = Effect.gen(function*() {
1861 * const queue = yield* Queue.bounded<number, Cause.Done>(3)
1862 *
1863 * const before = Queue.isFullUnsafe(queue)
1864 *
1865 * // Add some messages
1866 * yield* Queue.offerAll(queue, [1, 2, 3])
1867 *
1868 * const after = Queue.isFullUnsafe(queue)
1869 * return [before, after]
1870 * })
1871 *
1872 * await Effect.runPromise(program) // => [false, true]
1873 * ```
1874 *
1875 * @category predicates
1876 * @since 4.0.0
1877 */
1878export const isFullUnsafe = <A, E>(self: Dequeue<A, E>): boolean => sizeUnsafe(self) === self.capacity
1879
1880/**
1881 * Runs an `Effect` into a `Queue`, where success ends the queue and failure
1882 * fails the queue.
1883 *
1884 * **Example** (Running effects into queues)
1885 *
1886 * ```ts import.meta.vitest
1887 * import { Cause, Effect, Exit, Queue } from "effect"
1888 *
1889 * const program = Effect.gen(function*() {
1890 * const queue = yield* Queue.bounded<number, Cause.Done>(10)
1891 *
1892 * // Create an effect that succeeds
1893 * const dataProcessing = Effect.gen(function*() {
1894 * yield* Effect.yieldNow
1895 * return "Processing completed successfully"
1896 * })
1897 *
1898 * // Pipe the effect into the queue
1899 * // If dataProcessing succeeds, queue ends successfully
1900 * // If dataProcessing fails, queue fails with the error
1901 * const effectIntoQueue = Queue.into(queue)(dataProcessing)
1902 *
1903 * const wasCompleted = yield* effectIntoQueue
1904 * const exit = yield* Effect.exit(Queue.take(queue))
1905 * return [wasCompleted, exit]
1906 * })
1907 *
1908 * await Effect.runPromise(program) // => [true, Exit.fail(Cause.Done())]
1909 * ```
1910 *
1911 * @category completion
1912 * @since 4.0.0
1913 */
1914export const into: {
1915 <A, E>(
1916 self: Enqueue<A, E | Done>
1917 ): <AX, EX extends E, RX>(
1918 effect: Effect<AX, EX, RX>
1919 ) => Effect<boolean, never, RX>
1920 <AX, E, EX extends E, RX, A>(
1921 effect: Effect<AX, EX, RX>,
1922 self: Enqueue<A, E | Done>
1923 ): Effect<boolean, never, RX>
1924} = dual(
1925 2,
1926 <AX, E, EX extends E, RX, A>(
1927 effect: Effect<AX, EX, RX>,
1928 self: Enqueue<A, E | Done>
1929 ): Effect<boolean, never, RX> =>
1930 internalEffect.uninterruptibleMask((restore) =>
1931 internalEffect.matchCauseEffect(restore(effect), {
1932 onFailure: (cause) => failCause(self, cause),
1933 onSuccess: (_) => end(self)
1934 })
1935 )
1936)
1937
1938// -----------------------------------------------------------------------------
1939// internals
1940// -----------------------------------------------------------------------------
1941//
1942
1943const exitFalse = core.exitSucceed(false)
1944const exitTrue = core.exitSucceed(true)
1945const exitFailDone = core.exitFail(core.Done()) as Failure<never, Done>
1946const exitInterrupt = internalEffect.exitInterrupt() as Failure<never, never>
1947
1948const releaseTakers = <A, E>(self: Enqueue<A, E>) => {
1949 if (self.state._tag === "Done" || self.state.takers.size === 0) {
1950 return
1951 }
1952 for (const taker of self.state.takers) {
1953 if (!taker.ready()) continue
1954 self.state.takers.delete(taker)
1955 taker.resume(internalEffect.exitVoid)
1956 if (self.messages.length === 0) {
1957 break
1958 }
1959 }
1960}
1961
1962const scheduleReleaseTaker = <A, E>(self: Enqueue<A, E>) => {
1963 if (self.scheduleRunning || self.state._tag === "Done" || self.state.takers.size === 0) {
1964 return
1965 }
1966 self.scheduleRunning = true
1967 self.dispatcher.scheduleTask(() => {
1968 self.scheduleRunning = false
1969 releaseTakers(self)
1970 }, 0)
1971}
1972
1973const takeBetweenUnsafe = <A, E>(
1974 self: Dequeue<A, E>,
1975 min: number,
1976 max: number
1977): Exit<Array<A>, E> | undefined => {
1978 if (self.state._tag === "Done") {
1979 return self.state.exit
1980 } else if (max <= 0 || min <= 0) {
1981 return core.exitSucceed([])
1982 } else if (!canTake(self, min)) {
1983 return undefined
1984 }
1985 const messages = self.messages.length > 0
1986 ? MutableList.takeN(self.messages, max)
1987 : [takeOfferUnsafe(self.state.offers)]
1988 releaseCapacity(self)
1989 return core.exitSucceed(messages)
1990}
1991
1992// Whether a take of at least `min` messages can complete without waiting.
1993// A closing queue receives no more messages, so any remainder satisfies `min`.
1994const canTake = <A, E>(self: Dequeue<A, E>, min: number): boolean =>
1995 self.messages.length >= (self.state._tag === "Closing" ? 1 : Math.min(min, self.capacity || 1)) ||
1996 (self.capacity <= 0 && self.state._tag !== "Done" && self.state.offers.size > 0)
1997
1998// The readiness check and the taker registration run in one step, so no
1999// message can arrive between them. Wake-ups resume with void and the caller
2000// retries, which keeps the fiber stack flat across spurious wake-ups.
2001const awaitTake = <A, E>(self: Dequeue<A, E>, ready: () => boolean) =>
2002 internalEffect.callback<void, E>((resume) => {
2003 if (self.state._tag === "Done") return resume(self.state.exit)
2004 if (ready()) return resume(internalEffect.exitVoid)
2005 const taker = { ready, resume }
2006 self.state.takers.add(taker)
2007 return internalEffect.sync(() => {
2008 if (self.state._tag !== "Done") self.state.takers.delete(taker)
2009 })
2010 })
2011
2012const offerOrWait = <A, E>(self: Enqueue<A, E>, message: A) =>
2013 internalEffect.callback<boolean>((resume) =>
2014 offerUnsafe(self, message) ? resume(exitTrue) : waitToOffer(self, { _tag: "Single", message, resume })
2015 )
2016
2017const waitToOffer = <A, E>(self: Enqueue<A, E>, entry: Queue.OfferEntry<A>) => {
2018 if (self.state._tag !== "Open") return resumeUnoffered(entry)
2019 const offers = self.state.offers
2020 offers.add(entry)
2021 return internalEffect.sync(() => {
2022 if (self.state._tag === "Done") return
2023 offers.delete(entry)
2024 if (self.state._tag === "Closing" && offers.size === 0 && self.messages.length === 0) {
2025 finalize(self, self.state.exit)
2026 }
2027 })
2028}
2029
2030const resumeUnoffered = <A>(entry: Queue.OfferEntry<A>) =>
2031 entry._tag === "Single"
2032 ? entry.resume(exitFalse)
2033 : entry.resume(core.exitSucceed(entry.remaining.slice(entry.offset)))
2034
2035// Reserve a pending message for the consumer before the producer can reenter.
2036const takeOfferUnsafe = <A>(offers: Set<Queue.OfferEntry<A>>): A => {
2037 const entry = offers.values().next().value!
2038 if (entry._tag === "Single") {
2039 offers.delete(entry)
2040 entry.resume(exitTrue)
2041 return entry.message
2042 }
2043 const message = entry.remaining[entry.offset++]
2044 if (entry.offset === entry.remaining.length) {
2045 offers.delete(entry)
2046 entry.resume(core.exitSucceed([]))
2047 }
2048 return message
2049}
2050
2051const releaseCapacity = <A, E>(self: Dequeue<A, E>): boolean => {
2052 if (self.state._tag === "Done") {
2053 return Pull.isDoneCause(self.state.exit.cause)
2054 } else if (self.state.offers.size === 0) {
2055 if (
2056 self.state._tag === "Closing" &&
2057 self.messages.length === 0
2058 ) {
2059 finalize(self, self.state.exit)
2060 return Pull.isDoneCause(self.state.exit.cause)
2061 }
2062 return false
2063 }
2064 // Resuming a producer can synchronously take, offer, or shut down this queue.
2065 for (const entry of self.state.offers) {
2066 let n = self.capacity - self.messages.length
2067 if (n <= 0) break
2068 else if (entry._tag === "Single") {
2069 MutableList.append(self.messages, entry.message)
2070 self.state.offers.delete(entry)
2071 entry.resume(exitTrue)
2072 } else {
2073 for (; entry.offset < entry.remaining.length; entry.offset++) {
2074 if (n === 0) return false
2075 MutableList.append(self.messages, entry.remaining[entry.offset])
2076 n--
2077 }
2078 self.state.offers.delete(entry)
2079 entry.resume(core.exitSucceed([]))
2080 }
2081 }
2082 return false
2083}
2084
2085const takeAllUnsafe = <A, E>(self: Dequeue<A, E>) => {
2086 if (self.messages.length > 0) {
2087 const messages = MutableList.takeAll(self.messages)
2088 releaseCapacity(self)
2089 return messages
2090 } else if (self.state._tag !== "Done" && self.state.offers.size > 0) {
2091 const messages = [takeOfferUnsafe(self.state.offers)]
2092 releaseCapacity(self)
2093 return messages
2094 }
2095 return []
2096}
2097
2098const finalize = <A, E>(self: Enqueue<A, E> | Dequeue<A, E>, exit: Failure<never, E>) => {
2099 if (self.state._tag === "Done") {
2100 return
2101 }
2102 const openState = self.state
2103 self.state = { _tag: "Done", exit }
2104 for (const taker of openState.takers) {
2105 taker.resume(exit)
2106 }
2107 openState.takers.clear()
2108 for (const awaiter of openState.awaiters) {
2109 awaiter(exit)
2110 }
2111 openState.awaiters.clear()
2112}