1/**
2 * Transactional queues whose state changes participate in Effect transactions.
3 *
4 * A `TxQueue<A, E>` stores values of type `A`, exposes write-only `TxEnqueue`
5 * and read-only `TxDequeue` handles, and can complete, fail, or shut down with
6 * causes observed by consumers. Queue operations can retry transactionally when
7 * they cannot proceed, such as taking from an empty open queue or offering to a
8 * full bounded queue. This makes the queue useful for coordinating producers
9 * and consumers alongside other transactional state changes.
10 *
11 * @since 4.0.0
12 */
13import type * as Arr from "./Array.ts"
14import * as Cause from "./Cause.ts"
15import * as Chunk from "./Chunk.ts"
16import * as Effect from "./Effect.ts"
17import { dual } from "./Function.ts"
18import type { Inspectable } from "./Inspectable.ts"
19import { NodeInspectSymbol, toJson } from "./Inspectable.ts"
20import * as Count from "./internal/count.ts"
21import * as Option from "./Option.ts"
22import { hasProperty } from "./Predicate.ts"
23import { type ExcludeDone, isDoneCause } from "./Pull.ts"
24import * as TxChunk from "./TxChunk.ts"
25import * as TxRef from "./TxRef.ts"
26import type * as Types from "./Types.ts"
27
28/**
29 * Represents the state of a transactional queue with sophisticated lifecycle management.
30 *
31 * **Details**
32 *
33 * The queue progresses through three states:
34 * - **Open**: Accepting offers and serving takes normally
35 * - **Closing**: No new offers accepted, serving remaining items until empty
36 * - **Done**: Terminal state with completion cause, no further operations possible
37 *
38 * **Example** (Inspecting queue lifecycle states)
39 *
40 * ```ts import.meta.vitest
41 * import type { TxQueue } from "effect"
42 *
43 * const state: TxQueue.State<string, Error> = { _tag: "Open" }
44 * state._tag // => "Open"
45 * ```
46 *
47 * @category models
48 * @since 4.0.0
49 */
50export type State<_A, E> =
51 | {
52 readonly _tag: "Open"
53 }
54 | {
55 readonly _tag: "Closing"
56 readonly cause: Cause.Cause<E>
57 }
58 | {
59 readonly _tag: "Done"
60 readonly cause: Cause.Cause<E>
61 }
62
63const EnqueueTypeId = "~effect/TxQueue/Enqueue"
64const DequeueTypeId = "~effect/TxQueue/Dequeue"
65const TypeId = "~effect/TxQueue"
66
67/**
68 * Namespace containing type definitions for TxEnqueue variance annotations.
69 *
70 * @since 4.0.0
71 */
72export declare namespace TxEnqueue {
73 /**
74 * Variance annotation interface for TxEnqueue contravariance.
75 *
76 * @category models
77 * @since 4.0.0
78 */
79 export interface Variance<in A, in E> {
80 readonly _A: Types.Contravariant<A>
81 readonly _E: Types.Contravariant<E>
82 }
83}
84
85/**
86 * Namespace containing type definitions for TxDequeue variance annotations.
87 *
88 * @since 4.0.0
89 */
90export declare namespace TxDequeue {
91 /**
92 * Variance annotation interface for TxDequeue covariance.
93 *
94 * @category models
95 * @since 4.0.0
96 */
97 export interface Variance<out A, out E> {
98 readonly _A: Types.Covariant<A>
99 readonly _E: Types.Covariant<E>
100 }
101}
102
103/**
104 * Namespace containing type definitions for TxQueue variance annotations.
105 *
106 * @since 4.0.0
107 */
108export declare namespace TxQueue {
109 /**
110 * Variance annotation interface for TxQueue invariance.
111 *
112 * @category models
113 * @since 4.0.0
114 */
115 export interface Variance<in out A, in out E> {
116 readonly _A: Types.Invariant<A>
117 readonly _E: Types.Invariant<E>
118 }
119}
120
121/**
122 * Represents the shared state of a transactional queue that can be inspected.
123 * This interface contains the core properties needed for queue state inspection
124 * operations like size, capacity, and completion status.
125 *
126 * @category models
127 * @since 4.0.0
128 */
129export interface TxQueueState extends Inspectable {
130 readonly strategy: "bounded" | "unbounded" | "dropping" | "sliding"
131 readonly capacity: number
132 readonly items: TxChunk.TxChunk<any>
133 readonly stateRef: TxRef.TxRef<State<any, any>>
134}
135
136/**
137 * A TxEnqueue represents the write-only interface of a transactional queue, providing
138 * operations for adding elements (enqueue operations) and inspecting queue state.
139 *
140 * **Example** (Offering values through enqueue handles)
141 *
142 * ```ts import.meta.vitest
143 * import { Effect, TxQueue } from "effect"
144 * import type { Cause } from "effect"
145 *
146 * const program = Effect.gen(function*() {
147 * // Queue without error channel
148 * const queue = yield* TxQueue.bounded<number>(10)
149 * const accepted = yield* TxQueue.offer(queue, 42)
150 *
151 * // Queue with error channel for completion signaling
152 * const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
153 * yield* TxQueue.offerAll(faultTolerantQueue, [1, 2, 3])
154 * yield* TxQueue.fail(faultTolerantQueue, "processing complete")
155 *
156 * // Works with Done for clean completion
157 * const completableQueue = yield* TxQueue.bounded<
158 * string,
159 * Cause.Done
160 * >(5)
161 * yield* TxQueue.offer(completableQueue, "task")
162 * yield* TxQueue.end(completableQueue)
163 *
164 * return accepted
165 * })
166 *
167 * await Effect.runPromise(program) // => true
168 * ```
169 *
170 * @category models
171 * @since 4.0.0
172 */
173export interface TxEnqueue<in A, in E = never> extends TxQueueState {
174 readonly [EnqueueTypeId]: TxEnqueue.Variance<A, E>
175}
176
177/**
178 * A TxDequeue represents the read-only interface of a transactional queue, providing
179 * operations for consuming elements (dequeue operations) and inspecting queue state.
180 *
181 * **Example** (Taking values through dequeue handles)
182 *
183 * ```ts import.meta.vitest
184 * import { Effect, TxQueue } from "effect"
185 *
186 * const program = Effect.gen(function*() {
187 * // Queue without error channel
188 * const queue = yield* TxQueue.bounded<number>(10)
189 * yield* TxQueue.offer(queue, 42)
190 * const item = yield* TxQueue.take(queue)
191 *
192 * // Queue with error channel - errors propagate through E-channel
193 * const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
194 * yield* TxQueue.fail(faultTolerantQueue, "processing failed")
195 *
196 * // All dequeue operations now fail with the error directly
197 * const takeResult = yield* Effect.flip(TxQueue.take(faultTolerantQueue)) // "processing failed"
198 * const peekResult = yield* Effect.flip(TxQueue.peek(faultTolerantQueue)) // "processing failed"
199 * return [item, takeResult, peekResult] as const
200 * })
201 *
202 * await Effect.runPromise(program) // => [42, "processing failed", "processing failed"]
203 * ```
204 *
205 * @category models
206 * @since 4.0.0
207 */
208export interface TxDequeue<out A, out E = never> extends TxQueueState {
209 readonly [DequeueTypeId]: TxDequeue.Variance<A, E>
210}
211
212/**
213 * A TxQueue represents a transactional queue data structure that provides both
214 * enqueue and dequeue operations with Software Transactional Memory (STM) semantics.
215 *
216 * **Example** (Combining enqueue and dequeue operations)
217 *
218 * ```ts import.meta.vitest
219 * import { Effect, TxQueue } from "effect"
220 *
221 * const program = Effect.gen(function*() {
222 * // Create a bounded transactional queue (E defaults to never)
223 * const queue = yield* TxQueue.bounded<number>(10)
224 *
225 * // Single operations - automatically transactional
226 * const accepted = yield* TxQueue.offer(queue, 42)
227 * const item = yield* TxQueue.take(queue) // Effect<number, never>
228 *
229 * // Queue with error channel
230 * const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
231 *
232 * // Operations can handle queue-level failures
233 * yield* TxQueue.fail(faultTolerantQueue, "queue failed")
234 * const result = yield* Effect.flip(TxQueue.take(faultTolerantQueue))
235 * return [accepted, item, result] as const
236 * })
237 *
238 * await Effect.runPromise(program) // => [true, 42, "queue failed"]
239 * ```
240 *
241 * @category models
242 * @since 4.0.0
243 */
244export interface TxQueue<in out A, in out E = never> extends TxEnqueue<A, E>, TxDequeue<A, E> {
245 readonly [TypeId]: TxQueue.Variance<A, E>
246}
247
248/**
249 * Checks whether the given value is a TxEnqueue.
250 *
251 * **Example** (Checking enqueue handles)
252 *
253 * ```ts import.meta.vitest
254 * import { TxQueue } from "effect"
255 *
256 * const someValue: unknown = {}
257 * TxQueue.isTxEnqueue(someValue) // => false
258 * ```
259 *
260 * @category guards
261 * @since 4.0.0
262 */
263export const isTxEnqueue = <A = unknown, E = unknown>(u: unknown): u is TxEnqueue<A, E> => hasProperty(u, EnqueueTypeId)
264
265/**
266 * Checks whether the given value is a TxDequeue.
267 *
268 * **Example** (Checking dequeue handles)
269 *
270 * ```ts import.meta.vitest
271 * import { TxQueue } from "effect"
272 *
273 * const someValue: unknown = {}
274 * TxQueue.isTxDequeue(someValue) // => false
275 * ```
276 *
277 * @category guards
278 * @since 4.0.0
279 */
280export const isTxDequeue = <A = unknown, E = unknown>(u: unknown): u is TxDequeue<A, E> => hasProperty(u, DequeueTypeId)
281
282/**
283 * Checks whether the given value is a TxQueue.
284 *
285 * **Example** (Checking queue handles)
286 *
287 * ```ts import.meta.vitest
288 * import { TxQueue } from "effect"
289 *
290 * const someValue: unknown = {}
291 * TxQueue.isTxQueue(someValue) // => false
292 * ```
293 *
294 * @category guards
295 * @since 4.0.0
296 */
297export const isTxQueue = <A = unknown, E = unknown>(u: unknown): u is TxQueue<A, E> => hasProperty(u, TypeId)
298
299// =============================================================================
300// Proto
301// =============================================================================
302
303const TxQueueProto = {
304 [EnqueueTypeId]: { _A: (_: never) => _, _E: (_: never) => _ },
305 [DequeueTypeId]: { _A: (_: never) => _, _E: (_: never) => _ },
306 [TypeId]: { _A: (_: never) => _, _E: (_: never) => _ },
307 [NodeInspectSymbol](this: TxQueue<unknown, unknown>) {
308 return toJson(this)
309 },
310 toString(this: TxQueue<unknown, unknown>) {
311 return `TxQueue(${this.strategy}, ${this.capacity})`
312 },
313 toJSON(this: TxQueue<unknown, unknown>) {
314 return {
315 _id: "TxQueue",
316 strategy: this.strategy,
317 capacity: this.capacity
318 }
319 }
320}
321
322// =============================================================================
323// Constructors
324// =============================================================================
325
326/**
327 * Creates a new bounded `TxQueue` with the specified capacity.
328 *
329 * **Details**
330 *
331 * This function returns a new TxQueue reference with the specified capacity. No existing TxQueue instances are modified.
332 *
333 * **Example** (Creating bounded queues)
334 *
335 * ```ts import.meta.vitest
336 * import { Effect, TxQueue } from "effect"
337 *
338 * const program = Effect.gen(function*() {
339 * // Create a bounded queue (E defaults to never)
340 * const queue = yield* TxQueue.bounded<number>(10)
341 *
342 * // Create a bounded queue with error channel
343 * const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
344 *
345 * // Offer items - will succeed until capacity is reached
346 * yield* TxQueue.offer(queue, 1)
347 * yield* TxQueue.offer(queue, 2)
348 *
349 * return yield* TxQueue.take(queue)
350 * })
351 *
352 * await Effect.runPromise(program) // => 1
353 * ```
354 *
355 * @category constructors
356 * @since 2.0.0
357 */
358export const bounded = <A = never, E = never>(
359 capacity: number
360): Effect.Effect<TxQueue<A, E>> =>
361 Effect.gen(function*() {
362 const items = yield* TxChunk.empty<A>()
363 const stateRef = yield* TxRef.make<State<A, E>>({ _tag: "Open" })
364
365 const txQueue = Object.create(TxQueueProto)
366 txQueue.strategy = "bounded"
367 txQueue.capacity = capacity
368 txQueue.items = items
369 txQueue.stateRef = stateRef
370 return txQueue
371 }).pipe(Effect.tx)
372
373/**
374 * Creates a new unbounded `TxQueue` with unlimited capacity.
375 *
376 * **Details**
377 *
378 * This function returns a new TxQueue reference with unlimited capacity. No existing TxQueue instances are modified.
379 *
380 * **Example** (Creating unbounded queues)
381 *
382 * ```ts import.meta.vitest
383 * import { Effect, TxQueue } from "effect"
384 *
385 * const program = Effect.gen(function*() {
386 * // Create an unbounded queue (E defaults to never)
387 * const queue = yield* TxQueue.unbounded<string>()
388 *
389 * // Create an unbounded queue with error channel
390 * const faultTolerantQueue = yield* TxQueue.unbounded<string, Error>()
391 *
392 * // Can offer unlimited items
393 * yield* TxQueue.offer(queue, "hello")
394 * yield* TxQueue.offer(queue, "world")
395 *
396 * return yield* TxQueue.size(queue)
397 * })
398 *
399 * await Effect.runPromise(program) // => 2
400 * ```
401 *
402 * @category constructors
403 * @since 2.0.0
404 */
405export const unbounded = <A = never, E = never>(): Effect.Effect<TxQueue<A, E>> =>
406 Effect.gen(function*() {
407 const items = yield* TxChunk.empty<A>()
408 const stateRef = yield* TxRef.make<State<A, E>>({ _tag: "Open" })
409
410 const txQueue = Object.create(TxQueueProto)
411 txQueue.strategy = "unbounded"
412 txQueue.capacity = Number.POSITIVE_INFINITY
413 txQueue.items = items
414 txQueue.stateRef = stateRef
415 return txQueue
416 }).pipe(Effect.tx)
417
418/**
419 * Creates a new dropping `TxQueue` with the specified capacity that drops new items when full.
420 *
421 * **Details**
422 *
423 * This function returns a new TxQueue reference with dropping strategy. No existing TxQueue instances are modified.
424 *
425 * **Example** (Creating dropping queues)
426 *
427 * ```ts import.meta.vitest
428 * import { Effect, TxQueue } from "effect"
429 *
430 * const program = Effect.gen(function*() {
431 * // Create a dropping queue with capacity 2
432 * const queue = yield* TxQueue.dropping<number>(2)
433 *
434 * // Fill to capacity
435 * yield* TxQueue.offer(queue, 1)
436 * yield* TxQueue.offer(queue, 2)
437 *
438 * // This will be dropped (returns false)
439 * return yield* TxQueue.offer(queue, 3)
440 * })
441 *
442 * await Effect.runPromise(program) // => false
443 * ```
444 *
445 * @category constructors
446 * @since 2.0.0
447 */
448export const dropping = <A = never, E = never>(
449 capacity: number
450): Effect.Effect<TxQueue<A, E>> =>
451 Effect.gen(function*() {
452 const items = yield* TxChunk.empty<A>()
453 const stateRef = yield* TxRef.make<State<A, E>>({ _tag: "Open" })
454
455 const txQueue = Object.create(TxQueueProto)
456 txQueue.strategy = "dropping"
457 txQueue.capacity = capacity
458 txQueue.items = items
459 txQueue.stateRef = stateRef
460 return txQueue
461 }).pipe(Effect.tx)
462
463/**
464 * Creates a new sliding `TxQueue` with the specified capacity that evicts old items when full.
465 *
466 * **Details**
467 *
468 * This function returns a new TxQueue reference with sliding strategy. No existing TxQueue instances are modified.
469 *
470 * **Example** (Creating sliding queues)
471 *
472 * ```ts import.meta.vitest
473 * import { Effect, TxQueue } from "effect"
474 *
475 * const program = Effect.gen(function*() {
476 * // Create a sliding queue with capacity 2
477 * const queue = yield* TxQueue.sliding<number>(2)
478 *
479 * // Fill to capacity
480 * yield* TxQueue.offer(queue, 1)
481 * yield* TxQueue.offer(queue, 2)
482 *
483 * // This will evict item 1 and add 3
484 * yield* TxQueue.offer(queue, 3)
485 *
486 * return yield* TxQueue.take(queue)
487 * })
488 *
489 * await Effect.runPromise(program) // => 2
490 * ```
491 *
492 * @category constructors
493 * @since 2.0.0
494 */
495export const sliding = <A = never, E = never>(
496 capacity: number
497): Effect.Effect<TxQueue<A, E>> =>
498 Effect.gen(function*() {
499 const items = yield* TxChunk.empty<A>()
500 const stateRef = yield* TxRef.make<State<A, E>>({ _tag: "Open" })
501
502 const txQueue = Object.create(TxQueueProto)
503 txQueue.strategy = "sliding"
504 txQueue.capacity = capacity
505 txQueue.items = items
506 txQueue.stateRef = stateRef
507 return txQueue
508 }).pipe(Effect.tx)
509
510// =============================================================================
511// Core Queue Operations
512// =============================================================================
513
514/**
515 * Offers an item to the queue and returns whether it was accepted.
516 *
517 * **Details**
518 *
519 * Open unbounded queues always accept; open bounded queues retry while full; dropping queues return `false` when full; sliding queues evict the oldest item when full. Closing or done queues return `false`. This function mutates the original TxQueue by adding the item according to the queue's strategy. It does not return a new TxQueue reference.
520 *
521 * **Example** (Offering a value)
522 *
523 * ```ts import.meta.vitest
524 * import { Effect, TxQueue } from "effect"
525 *
526 * const program = Effect.gen(function*() {
527 * const queue = yield* TxQueue.bounded<number>(10)
528 *
529 * // Offer an item - returns true if accepted
530 * return yield* TxQueue.offer(queue, 42)
531 * })
532 *
533 * await Effect.runPromise(program) // => true
534 * ```
535 *
536 * @category combinators
537 * @since 2.0.0
538 */
539export const offer: {
540 <A, E>(value: A): (self: TxEnqueue<A, E>) => Effect.Effect<boolean>
541 <A, E>(self: TxEnqueue<A, E>, value: A): Effect.Effect<boolean>
542} = dual(
543 2,
544 <A, E>(self: TxEnqueue<A, E>, value: A): Effect.Effect<boolean> =>
545 Effect.gen(function*() {
546 const state = yield* TxRef.get(self.stateRef)
547 if (state._tag === "Done" || state._tag === "Closing") {
548 return false
549 }
550
551 const currentSize = yield* size(self)
552
553 // Unbounded - always accept
554 if (self.strategy === "unbounded") {
555 yield* TxChunk.append(self.items, value)
556 return true
557 }
558
559 // For bounded queues, check capacity
560 if (currentSize < self.capacity) {
561 yield* TxChunk.append(self.items, value)
562 return true
563 }
564
565 // Queue is at capacity, strategy-specific behavior
566 if (self.strategy === "dropping") {
567 return false // Drop the new item
568 }
569
570 if (self.strategy === "sliding") {
571 yield* TxChunk.drop(self.items, 1) // Remove oldest item
572 yield* TxChunk.append(self.items, value) // Add new item
573 return true
574 }
575
576 // bounded strategy - block until space is available
577 return yield* Effect.txRetry
578 }).pipe(Effect.tx)
579)
580
581/**
582 * Offers multiple items to the queue, returning the items that were not
583 * accepted.
584 *
585 * **Details**
586 *
587 * Each item follows `offer` semantics: bounded queues retry while full, dropping queues reject new items when full, sliding queues evict old items to accept new items, and closing or done queues reject all items. This function mutates the original TxQueue by adding items according to the queue's strategy. It does not return a new TxQueue reference.
588 *
589 * **Example** (Offering multiple values)
590 *
591 * ```ts import.meta.vitest
592 * import { Effect, TxQueue } from "effect"
593 *
594 * const program = Effect.gen(function*() {
595 * const queue = yield* TxQueue.bounded<number>(10)
596 *
597 * // Offer multiple items - returns rejected items as array
598 * return yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
599 * })
600 *
601 * await Effect.runPromise(program) // => []
602 * ```
603 *
604 * @category combinators
605 * @since 2.0.0
606 */
607export const offerAll: {
608 <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>
609 <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>
610} = dual(
611 2,
612 <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>> => {
613 const valuesArray = Array.from(values)
614
615 return Effect.gen(function*() {
616 const rejected: Array<A> = []
617
618 for (const value of valuesArray) {
619 const accepted = yield* offer(self, value)
620 if (!accepted) {
621 rejected.push(value)
622 }
623 }
624
625 return rejected
626 }).pipe(Effect.tx)
627 }
628)
629
630/**
631 * Takes the next item from the queue, retrying the transaction while the queue
632 * is empty.
633 *
634 * **Details**
635 *
636 * If the queue is done, the effect fails with the queue's completion cause. This function mutates the original TxQueue by removing the first item. It does not return a new TxQueue reference.
637 *
638 * **Example** (Taking a value)
639 *
640 * ```ts import.meta.vitest
641 * import { Effect, Exit, TxQueue } from "effect"
642 *
643 * const program = Effect.gen(function*() {
644 * const queue = yield* TxQueue.bounded<number, string>(10)
645 * yield* TxQueue.offer(queue, 42)
646 *
647 * // Take an item - blocks if empty
648 * const item = yield* TxQueue.take(queue)
649 *
650 * // When queue fails, take fails with the same error
651 * yield* TxQueue.fail(queue, "queue error")
652 * const result = yield* Effect.exit(TxQueue.take(queue))
653 * return [item, result] as const
654 * })
655 *
656 * await Effect.runPromise(program) // => [42, Exit.fail("queue error")]
657 * ```
658 *
659 * @category combinators
660 * @since 2.0.0
661 */
662export const take = <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E> =>
663 Effect.gen(function*() {
664 const state = yield* TxRef.get(self.stateRef)
665
666 // Check if queue is done - forward the cause directly
667 if (state._tag === "Done") {
668 return yield* Effect.failCause(state.cause)
669 }
670
671 // If no items available, retry transaction
672 if (yield* isEmpty(self)) {
673 return yield* Effect.txRetry
674 }
675
676 // Take item from queue
677 const chunk = yield* TxChunk.get(self.items)
678 const head = Chunk.head(chunk)
679 if (Option.isNone(head)) {
680 return yield* Effect.txRetry
681 }
682
683 yield* TxChunk.drop(self.items, 1)
684
685 // Check if we need to transition Closing → Done
686 if (state._tag === "Closing" && (yield* isEmpty(self))) {
687 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
688 }
689
690 return head.value
691 }).pipe(Effect.tx)
692
693/**
694 * Tries to take an item from the queue without blocking.
695 *
696 * **Example** (Polling without blocking)
697 *
698 * ```ts import.meta.vitest
699 * import { Effect, Option, TxQueue } from "effect"
700 *
701 * const program = Effect.gen(function*() {
702 * const queue = yield* TxQueue.bounded<number>(10)
703 *
704 * // Poll returns Option.none if empty
705 * const maybe = yield* TxQueue.poll(queue)
706 *
707 * yield* TxQueue.offer(queue, 42)
708 * const item = yield* TxQueue.poll(queue)
709 * return [maybe, item] as const
710 * })
711 *
712 * await Effect.runPromise(program) // => [Option.none(), Option.some(42)]
713 * ```
714 *
715 * @category combinators
716 * @since 2.0.0
717 */
718export const poll = <A, E>(self: TxDequeue<A, E>): Effect.Effect<Option.Option<A>> =>
719 Effect.gen(function*() {
720 const state = yield* TxRef.get(self.stateRef)
721 if (state._tag === "Done") {
722 return Option.none()
723 }
724
725 const chunk = yield* TxChunk.get(self.items)
726 const head = Chunk.head(chunk)
727 if (Option.isNone(head)) {
728 return Option.none()
729 }
730
731 yield* TxChunk.drop(self.items, 1)
732
733 if (state._tag === "Closing" && (yield* isEmpty(self))) {
734 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
735 }
736
737 return Option.some(head.value)
738 }).pipe(Effect.tx)
739
740/**
741 * Takes all items from the queue. Blocks if the queue is empty.
742 *
743 * **Details**
744 *
745 * If the queue is already in a failed state, the error is propagated through the E-channel. This follows the same patterns as `take` and waits when there are no elements. It returns a non-empty array because it blocks until at least one item is available. This function mutates the original TxQueue by removing all items. It does not return a new TxQueue reference.
746 *
747 * **Example** (Taking all queued values)
748 *
749 * ```ts import.meta.vitest
750 * import { Effect, Exit, TxQueue } from "effect"
751 *
752 * const program = Effect.gen(function*() {
753 * const queue = yield* TxQueue.bounded<number, string>(10)
754 * yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
755 *
756 * // Take all items atomically - returns NonEmptyArray
757 * return yield* TxQueue.takeAll(queue)
758 * })
759 *
760 * // Error propagation example
761 * const errorExample = Effect.gen(function*() {
762 * const queue = yield* TxQueue.bounded<number, string>(5)
763 * yield* TxQueue.offerAll(queue, [1, 2])
764 * yield* TxQueue.fail(queue, "processing error")
765 *
766 * // takeAll() propagates the queue error through E-channel
767 * return yield* Effect.exit(TxQueue.takeAll(queue))
768 * })
769 *
770 * await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
771 * await Effect.runPromise(errorExample) // => Exit.fail("processing error")
772 * ```
773 *
774 * @category combinators
775 * @since 2.0.0
776 */
777export const takeAll = <A, E>(self: TxDequeue<A, E>): Effect.Effect<Arr.NonEmptyArray<A>, E> =>
778 Effect.gen(function*() {
779 const state = yield* TxRef.get(self.stateRef)
780
781 // Handle done queue
782 if (state._tag === "Done") {
783 return yield* Effect.failCause(state.cause)
784 }
785
786 // Wait if empty - same pattern as take()
787 if (yield* isEmpty(self)) {
788 return yield* Effect.txRetry
789 }
790
791 const chunk = yield* TxChunk.get(self.items)
792
793 // Take all items (guaranteed non-empty due to isEmpty check above)
794 const items = Chunk.toArray(chunk) as Arr.NonEmptyArray<A>
795 yield* TxChunk.set(self.items, Chunk.empty())
796
797 // Check if we need to transition Closing → Done
798 if (state._tag === "Closing") {
799 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
800 }
801
802 return items
803 }).pipe(Effect.tx)
804
805/**
806 * Takes up to `n` items from the queue in a single transaction.
807 *
808 * **Details**
809 *
810 * For an open queue, waits until `min(n, capacity)` items are available, then removes that many items. Finite fractional values of `n` are rounded down. If `n` is `NaN` or non-positive, returns an empty array without modifying the queue. If the queue is closing, drains the currently available items and transitions to `Done`. If the queue is already done, the effect fails with the queue's completion cause. This function mutates the original TxQueue by removing the taken items. It does not return a new TxQueue reference.
811 *
812 * **Example** (Taking a fixed number of values)
813 *
814 * ```ts import.meta.vitest
815 * import { Effect, TxQueue } from "effect"
816 *
817 * const program = Effect.gen(function*() {
818 * const queue = yield* TxQueue.bounded<number>(5)
819 * yield* TxQueue.offerAll(queue, [1, 2, 3, 4])
820 *
821 * const items = yield* TxQueue.takeN(queue, 4)
822 *
823 * // This requests more than capacity (5), so takes all available (up to 5)
824 * yield* TxQueue.offerAll(queue, [5, 6, 7, 8, 9])
825 * const all = yield* TxQueue.takeN(queue, 10)
826 * return [items, all] as const
827 * })
828 *
829 * await Effect.runPromise(program) // => [[1, 2, 3, 4], [5, 6, 7, 8, 9]]
830 * ```
831 *
832 * @category combinators
833 * @since 2.0.0
834 */
835export const takeN: {
836 (n: number): <A, E>(self: TxDequeue<A, E>) => Effect.Effect<Array<A>, E>
837 <A, E>(self: TxDequeue<A, E>, n: number): Effect.Effect<Array<A>, E>
838} = dual(
839 2,
840 <A, E>(self: TxDequeue<A, E>, n: number): Effect.Effect<Array<A>, E> => {
841 const requestedCount = Count.normalize(n)
842 return Effect.gen(function*() {
843 const state = yield* TxRef.get(self.stateRef)
844
845 // Check if queue is done - forward the cause directly
846 if (state._tag === "Done") {
847 return yield* Effect.failCause(state.cause)
848 }
849
850 const currentSize = yield* size(self)
851
852 // Determine how many items we can/should take
853 const maxPossible = Math.min(requestedCount, self.capacity)
854
855 // If we can't get the requested amount due to capacity constraints,
856 // take what the capacity allows. Otherwise, wait for the full amount.
857 const shouldWaitForFull = requestedCount <= self.capacity
858 const minimumRequired = shouldWaitForFull ? requestedCount : maxPossible
859
860 // If we don't have enough items available
861 if (currentSize < minimumRequired) {
862 // If queue is closing, transition to done and return what we have
863 if (state._tag === "Closing") {
864 if (yield* isEmpty(self)) {
865 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
866 return []
867 }
868 // Take all remaining items when closing
869 const chunk = yield* TxChunk.get(self.items)
870 const taken = Chunk.toArray(chunk)
871 yield* TxChunk.set(self.items, Chunk.empty())
872 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
873 return taken
874 }
875
876 // Queue is still open but not enough items - retry transaction
877 return yield* Effect.txRetry
878 }
879
880 // Take the determined number of items
881 const toTake = minimumRequired
882 const chunk = yield* TxChunk.get(self.items)
883 const taken = Chunk.take(chunk, toTake)
884 yield* TxChunk.drop(self.items, toTake)
885
886 // Check if we need to transition Closing → Done
887 if (state._tag === "Closing" && (yield* isEmpty(self))) {
888 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
889 }
890
891 return Chunk.toArray(taken)
892 }).pipe(Effect.tx)
893 }
894)
895
896/**
897 * Takes between `min` and `max` currently available items, waiting for `min` on
898 * an open queue.
899 *
900 * **Details**
901 *
902 * Finite fractional bounds are rounded down, while `NaN` and non-positive bounds are treated as `0`. If the queue is closing, drains the currently available items even when fewer than `min` are available and transitions to `Done`. Invalid normalized ranges (`min <= 0`, `max <= 0`, or `min > max`) return an empty array. If the queue is already done, the effect fails with the queue's completion cause.
903 *
904 * **Example** (Taking batches within bounds)
905 *
906 * ```ts import.meta.vitest
907 * import { Effect, TxQueue } from "effect"
908 *
909 * const program = Effect.gen(function*() {
910 * const queue = yield* TxQueue.bounded<number>(10)
911 * yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8])
912 *
913 * // Take between 2 and 5 items
914 * const batch1 = yield* TxQueue.takeBetween(queue, 2, 5)
915 *
916 * // Take between 1 and 10 items (but only 3 remain)
917 * const batch2 = yield* TxQueue.takeBetween(queue, 1, 10)
918 *
919 * // Would wait for at least 1 item to be available
920 * // const batch3 = yield* TxQueue.takeBetween(queue, 1, 3)
921 * return [batch1, batch2] as const
922 * })
923 *
924 * await Effect.runPromise(program) // => [[1, 2, 3, 4, 5], [6, 7, 8]]
925 * ```
926 *
927 * @category taking
928 * @since 2.0.0
929 */
930export const takeBetween: {
931 (min: number, max: number): <A, E>(self: TxDequeue<A, E>) => Effect.Effect<Array<A>, E>
932 <A, E>(self: TxDequeue<A, E>, min: number, max: number): Effect.Effect<Array<A>, E>
933} = dual(
934 3,
935 <A, E>(self: TxDequeue<A, E>, min: number, max: number): Effect.Effect<Array<A>, E> => {
936 const minimum = Count.normalize(min)
937 const maximum = Count.normalize(max)
938 return Effect.gen(function*() {
939 const state = yield* TxRef.get(self.stateRef)
940
941 // Check if queue is done - forward the cause directly
942 if (state._tag === "Done") {
943 return yield* Effect.failCause(state.cause)
944 }
945
946 // Validate parameters
947 if (minimum <= 0 || maximum <= 0 || minimum > maximum) {
948 return []
949 }
950
951 const currentSize = yield* size(self)
952
953 // If we have less than minimum required items
954 if (currentSize < minimum) {
955 // If queue is closing, transition to done and return what we have
956 if (state._tag === "Closing") {
957 if (yield* isEmpty(self)) {
958 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
959 return []
960 }
961 // Take all remaining items when closing (if >= min or all available)
962 const chunk = yield* TxChunk.get(self.items)
963 const taken = Chunk.toArray(chunk)
964 yield* TxChunk.set(self.items, Chunk.empty())
965 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
966 return taken
967 }
968
969 // Queue is still open but not enough items - retry transaction
970 return yield* Effect.txRetry
971 }
972
973 // We have at least the minimum, take up to the maximum
974 const toTake = Math.min(currentSize, maximum)
975 const chunk = yield* TxChunk.get(self.items)
976 const taken = Chunk.take(chunk, toTake)
977 yield* TxChunk.drop(self.items, toTake)
978
979 // Check if we need to transition Closing → Done
980 if (state._tag === "Closing" && (yield* isEmpty(self))) {
981 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
982 }
983
984 return Chunk.toArray(taken)
985 }).pipe(Effect.tx)
986 }
987)
988
989/**
990 * Waits transactionally for the next item and returns it without removing it.
991 *
992 * **Details**
993 *
994 * If the queue is open but empty, the transaction retries until an item is available or the queue completes. If the queue is done, the queue's completion cause is propagated through the error channel.
995 *
996 * **Example** (Peeking without removing values)
997 *
998 * ```ts import.meta.vitest
999 * import { Effect, Exit, TxQueue } from "effect"
1000 *
1001 * const program = Effect.gen(function*() {
1002 * const queue = yield* TxQueue.bounded<number, string>(10)
1003 * yield* TxQueue.offer(queue, 42)
1004 *
1005 * // Peek at the next item without removing it
1006 * const item = yield* TxQueue.peek(queue)
1007 *
1008 * // Item is still in the queue
1009 * const size = yield* TxQueue.size(queue)
1010 * return [item, size] as const
1011 * })
1012 *
1013 * // Error handling example
1014 * const errorExample = Effect.gen(function*() {
1015 * const queue = yield* TxQueue.bounded<number, string>(5)
1016 * yield* TxQueue.fail(queue, "queue failed")
1017 *
1018 * // peek() propagates the queue error through E-channel
1019 * return yield* Effect.exit(TxQueue.peek(queue))
1020 * })
1021 *
1022 * await Effect.runPromise(program) // => [42, 1]
1023 * await Effect.runPromise(errorExample) // => Exit.fail("queue failed")
1024 * ```
1025 *
1026 * @category combinators
1027 * @since 2.0.0
1028 */
1029export const peek = <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E> =>
1030 Effect.gen(function*() {
1031 const state = yield* TxRef.get(self.stateRef)
1032 if (state._tag === "Done") {
1033 return yield* Effect.failCause(state.cause)
1034 }
1035
1036 const chunk = yield* TxChunk.get(self.items)
1037 const head = Chunk.head(chunk)
1038 if (Option.isNone(head)) {
1039 return yield* Effect.txRetry
1040 }
1041
1042 return head.value
1043 }).pipe(Effect.tx)
1044
1045/**
1046 * Gets the current size of the queue.
1047 *
1048 * **Example** (Reading queue size)
1049 *
1050 * ```ts import.meta.vitest
1051 * import { Effect, TxQueue } from "effect"
1052 *
1053 * const program = Effect.gen(function*() {
1054 * const queue = yield* TxQueue.bounded<number>(10)
1055 * yield* TxQueue.offerAll(queue, [1, 2, 3])
1056 *
1057 * return yield* TxQueue.size(queue)
1058 * })
1059 *
1060 * await Effect.runPromise(program) // => 3
1061 * ```
1062 *
1063 * @category combinators
1064 * @since 2.0.0
1065 */
1066export const size = (self: TxQueueState): Effect.Effect<number> => TxChunk.size(self.items)
1067
1068/**
1069 * Checks whether the queue is empty.
1070 *
1071 * **Example** (Checking whether a queue is empty)
1072 *
1073 * ```ts import.meta.vitest
1074 * import { Effect, TxQueue } from "effect"
1075 *
1076 * const program = Effect.gen(function*() {
1077 * const queue = yield* TxQueue.bounded<number>(10)
1078 *
1079 * const empty = yield* TxQueue.isEmpty(queue)
1080 *
1081 * yield* TxQueue.offer(queue, 42)
1082 * const stillEmpty = yield* TxQueue.isEmpty(queue)
1083 * return [empty, stillEmpty] as const
1084 * })
1085 *
1086 * await Effect.runPromise(program) // => [true, false]
1087 * ```
1088 *
1089 * @category predicates
1090 * @since 2.0.0
1091 */
1092export const isEmpty = (self: TxQueueState): Effect.Effect<boolean> => TxChunk.isEmpty(self.items)
1093
1094/**
1095 * Checks whether the queue is non-empty.
1096 *
1097 * **Example** (Checking whether a queue is non-empty)
1098 *
1099 * ```ts import.meta.vitest
1100 * import { Effect, TxQueue } from "effect"
1101 *
1102 * const program = Effect.gen(function*() {
1103 * const queue = yield* TxQueue.bounded<number>(10)
1104 *
1105 * const empty = yield* TxQueue.isNonEmpty(queue)
1106 *
1107 * yield* TxQueue.offer(queue, 42)
1108 * const nonEmpty = yield* TxQueue.isNonEmpty(queue)
1109 * return [empty, nonEmpty] as const
1110 * })
1111 *
1112 * await Effect.runPromise(program) // => [false, true]
1113 * ```
1114 *
1115 * @category predicates
1116 * @since 4.0.0
1117 */
1118export const isNonEmpty = (self: TxQueueState): Effect.Effect<boolean> => TxChunk.isNonEmpty(self.items)
1119
1120/**
1121 * Checks whether the queue is at capacity.
1122 *
1123 * **Example** (Checking whether a queue is full)
1124 *
1125 * ```ts import.meta.vitest
1126 * import { Effect, TxQueue } from "effect"
1127 *
1128 * const program = Effect.gen(function*() {
1129 * const queue = yield* TxQueue.bounded<number>(2)
1130 *
1131 * const full = yield* TxQueue.isFull(queue)
1132 *
1133 * yield* TxQueue.offerAll(queue, [1, 2])
1134 * const nowFull = yield* TxQueue.isFull(queue)
1135 * return [full, nowFull] as const
1136 * })
1137 *
1138 * await Effect.runPromise(program) // => [false, true]
1139 * ```
1140 *
1141 * @category predicates
1142 * @since 2.0.0
1143 */
1144export const isFull = (self: TxQueueState): Effect.Effect<boolean> =>
1145 self.capacity === Number.POSITIVE_INFINITY
1146 ? Effect.succeed(false)
1147 : Effect.map(size(self), (currentSize) => currentSize >= self.capacity)
1148
1149/**
1150 * Interrupts the queue gracefully with the current fiber's interruption cause.
1151 *
1152 * **Details**
1153 *
1154 * 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.
1155 *
1156 * **Example** (Interrupting queues)
1157 *
1158 * ```ts import.meta.vitest
1159 * import { Effect, TxQueue } from "effect"
1160 *
1161 * const program = Effect.gen(function*() {
1162 * const queue = yield* TxQueue.bounded<number>(10)
1163 * yield* TxQueue.offer(queue, 42)
1164 *
1165 * // Interrupt gracefully - allows remaining items to be consumed
1166 * return yield* TxQueue.interrupt(queue)
1167 * })
1168 *
1169 * await Effect.runPromise(program) // => true
1170 * ```
1171 *
1172 * @category combinators
1173 * @since 4.0.0
1174 */
1175export const interrupt = <A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean> =>
1176 Effect.withFiber((fiber) => failCause(self, Cause.interrupt(fiber.id)))
1177
1178/**
1179 * Fails the queue with the specified error, discarding any buffered items.
1180 *
1181 * **Details**
1182 *
1183 * The queue transitions directly to done with `Cause.fail(error)`. Returns `false` if the queue was already closing or done.
1184 *
1185 * **Example** (Failing queues)
1186 *
1187 * ```ts import.meta.vitest
1188 * import { Effect, TxQueue } from "effect"
1189 *
1190 * const program = Effect.gen(function*() {
1191 * const queue = yield* TxQueue.bounded<number, string>(10)
1192 *
1193 * // Fail the queue with an error
1194 * return yield* TxQueue.fail(queue, "connection lost")
1195 * })
1196 *
1197 * await Effect.runPromise(program) // => true
1198 * ```
1199 *
1200 * @category combinators
1201 * @since 4.0.0
1202 */
1203export const fail: {
1204 <E>(error: E): <A>(self: TxEnqueue<A, E>) => Effect.Effect<boolean>
1205 <A, E>(self: TxEnqueue<A, E>, error: E): Effect.Effect<boolean>
1206} = dual(
1207 2,
1208 <A, E>(self: TxEnqueue<A, E>, error: E): Effect.Effect<boolean> =>
1209 Effect.gen(function*() {
1210 const state = yield* TxRef.get(self.stateRef)
1211
1212 if (state._tag !== "Open") {
1213 return false // Already closing/done
1214 }
1215
1216 // Fail transitions directly to Done, clearing items
1217 yield* TxChunk.set(self.items, Chunk.empty())
1218 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: Cause.fail(error) })
1219
1220 return true
1221 }).pipe(Effect.tx)
1222)
1223
1224/**
1225 * Completes the queue with the specified cause.
1226 *
1227 * **Details**
1228 *
1229 * If the queue is empty, it transitions directly to done. If it still contains items, it enters the closing state so buffered items can be drained before the cause is observed. Returns `false` if the queue was already closing or done.
1230 *
1231 * **Example** (Failing queues with causes)
1232 *
1233 * ```ts import.meta.vitest
1234 * import { Cause, Effect, TxQueue } from "effect"
1235 *
1236 * const program = Effect.gen(function*() {
1237 * const queue = yield* TxQueue.bounded<number>(10)
1238 *
1239 * // Complete with specific cause
1240 * const cause = Cause.interrupt()
1241 * const result = yield* TxQueue.failCause(queue, cause)
1242 * return [cause, result] as const
1243 * })
1244 *
1245 * await Effect.runPromise(program) // => [Cause.interrupt(), true]
1246 * ```
1247 *
1248 * @category combinators
1249 * @since 4.0.0
1250 */
1251export const failCause: {
1252 <E>(cause: Cause.Cause<E>): <A>(self: TxEnqueue<A, E>) => Effect.Effect<boolean>
1253 <A, E>(self: TxEnqueue<A, E>, cause: Cause.Cause<E>): Effect.Effect<boolean>
1254} = dual(
1255 2,
1256 <A, E>(self: TxEnqueue<A, E>, cause: Cause.Cause<E>): Effect.Effect<boolean> =>
1257 Effect.gen(function*() {
1258 const state = yield* TxRef.get(self.stateRef)
1259
1260 if (state._tag !== "Open") {
1261 return false // Already closing/done
1262 }
1263
1264 if (yield* isEmpty(self)) {
1265 // Can transition directly to Done
1266 yield* TxRef.set(self.stateRef, { _tag: "Done", cause })
1267 } else {
1268 // Need to go through Closing state
1269 yield* TxRef.set(self.stateRef, { _tag: "Closing", cause })
1270 }
1271
1272 return true
1273 }).pipe(Effect.tx)
1274)
1275
1276/**
1277 * Ends a queue by signaling completion with a `Cause.Done` error.
1278 *
1279 * **Details**
1280 *
1281 * This is a convenience wrapper around `failCause` for queues whose error channel can contain `Cause.Done`. If buffered items remain, the queue enters the closing state and those items may still be consumed before later `take` or `peek` operations fail with `Cause.Done`.
1282 *
1283 * **Example** (Ending queues)
1284 *
1285 * ```ts import.meta.vitest
1286 * import { Cause, Effect, Exit, TxQueue } from "effect"
1287 *
1288 * const program = Effect.gen(function*() {
1289 * const queue = yield* TxQueue.bounded<number, Cause.Done>(10)
1290 *
1291 * // Signal the end of the queue
1292 * const result = yield* TxQueue.end(queue)
1293 *
1294 * // All operations will now fail with Done
1295 * const takeResult = yield* Effect.exit(TxQueue.take(queue))
1296 *
1297 * const peekResult = yield* Effect.exit(TxQueue.peek(queue))
1298 * return [result, takeResult, peekResult] as const
1299 * })
1300 *
1301 * await Effect.runPromise(program) // => [true, Exit.fail(Cause.Done()), Exit.fail(Cause.Done())]
1302 * ```
1303 *
1304 * @category combinators
1305 * @since 4.0.0
1306 */
1307export const end = <A, E>(self: TxEnqueue<A, E | Cause.Done>): Effect.Effect<boolean> =>
1308 failCause(self, Cause.fail(Cause.Done()))
1309
1310/**
1311 * Removes and returns all currently buffered elements.
1312 *
1313 * **Details**
1314 *
1315 * If the queue is closing, draining its buffered elements transitions it to done. If the queue is already done with a `Cause.Done` error, returns an empty array. If the queue is done for any other cause, including interruption or failure, that cause is propagated.
1316 *
1317 * **Example** (Clearing queues)
1318 *
1319 * ```ts import.meta.vitest
1320 * import { Effect, TxQueue } from "effect"
1321 *
1322 * const program = Effect.gen(function*() {
1323 * const queue = yield* TxQueue.bounded<number>(10)
1324 * yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
1325 *
1326 * const sizeBefore = yield* TxQueue.size(queue)
1327 *
1328 * const cleared = yield* TxQueue.clear(queue)
1329 *
1330 * const sizeAfter = yield* TxQueue.size(queue)
1331 * return [sizeBefore, cleared, sizeAfter] as const
1332 * })
1333 *
1334 * await Effect.runPromise(program) // => [5, [1, 2, 3, 4, 5], 0]
1335 * ```
1336 *
1337 * @category combinators
1338 * @since 4.0.0
1339 */
1340export const clear = <A, E>(self: TxEnqueue<A, E>): Effect.Effect<Array<A>, ExcludeDone<E>> =>
1341 Effect.gen(function*() {
1342 const state = yield* TxRef.get(self.stateRef)
1343 if (state._tag === "Done") {
1344 // Return empty array only for halt causes (like Cause.Done)
1345 if (isDoneCause(state.cause)) {
1346 return []
1347 }
1348 return yield* Effect.failCause(state.cause)
1349 }
1350 const chunk = yield* TxChunk.get(self.items)
1351 yield* TxChunk.set(self.items, Chunk.empty())
1352 if (state._tag === "Closing") {
1353 yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
1354 }
1355 return Chunk.toArray(chunk)
1356 }).pipe(Effect.tx)
1357
1358/**
1359 * Shuts down the queue immediately by clearing all items and interrupting it (legacy compatibility).
1360 *
1361 * **Details**
1362 *
1363 * This operation clears all items from the queue using `clear`, then interrupts the queue using `interrupt`. This function mutates the original TxQueue by clearing its contents and marking it as shutdown. It does not return a new TxQueue reference.
1364 *
1365 * **Example** (Shutting down queues)
1366 *
1367 * ```ts import.meta.vitest
1368 * import { Effect, TxQueue } from "effect"
1369 *
1370 * const program = Effect.gen(function*() {
1371 * const queue = yield* TxQueue.bounded<number>(10)
1372 * yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
1373 *
1374 * const sizeBefore = yield* TxQueue.size(queue)
1375 *
1376 * yield* TxQueue.shutdown(queue)
1377 *
1378 * const sizeAfter = yield* TxQueue.size(queue)
1379 *
1380 * const isShutdown = yield* TxQueue.isShutdown(queue)
1381 * return [sizeBefore, sizeAfter, isShutdown] as const
1382 * })
1383 *
1384 * await Effect.runPromise(program) // => [5, 0, true]
1385 * ```
1386 *
1387 * @category combinators
1388 * @since 2.0.0
1389 */
1390export const shutdown = <A, E>(self: TxEnqueue<A, E>): Effect.Effect<boolean> =>
1391 Effect.gen(function*() {
1392 yield* Effect.ignoreCause(clear(self))
1393 return yield* interrupt(self)
1394 }).pipe(Effect.tx)
1395
1396/**
1397 * Checks whether the queue is in the open state.
1398 *
1399 * **Example** (Checking open state)
1400 *
1401 * ```ts import.meta.vitest
1402 * import { Effect, TxQueue } from "effect"
1403 *
1404 * const program = Effect.gen(function*() {
1405 * const queue = yield* TxQueue.bounded<number>(10)
1406 *
1407 * const open = yield* TxQueue.isOpen(queue)
1408 *
1409 * yield* TxQueue.interrupt(queue)
1410 * const stillOpen = yield* TxQueue.isOpen(queue)
1411 * return [open, stillOpen] as const
1412 * })
1413 *
1414 * await Effect.runPromise(program) // => [true, false]
1415 * ```
1416 *
1417 * @category predicates
1418 * @since 4.0.0
1419 */
1420export const isOpen = (self: TxQueueState): Effect.Effect<boolean> =>
1421 Effect.map(TxRef.get(self.stateRef), (state) => state._tag === "Open")
1422
1423/**
1424 * Checks whether the queue is in the closing state.
1425 *
1426 * **Example** (Checking closing state)
1427 *
1428 * ```ts import.meta.vitest
1429 * import { Effect, TxQueue } from "effect"
1430 *
1431 * const program = Effect.gen(function*() {
1432 * const queue = yield* TxQueue.bounded<number>(10)
1433 * yield* TxQueue.offer(queue, 42)
1434 *
1435 * const closing = yield* TxQueue.isClosing(queue)
1436 *
1437 * yield* TxQueue.interrupt(queue)
1438 * const nowClosing = yield* TxQueue.isClosing(queue)
1439 * return [closing, nowClosing] as const
1440 * })
1441 *
1442 * await Effect.runPromise(program) // => [false, true]
1443 * ```
1444 *
1445 * @category predicates
1446 * @since 4.0.0
1447 */
1448export const isClosing = (self: TxQueueState): Effect.Effect<boolean> =>
1449 Effect.map(TxRef.get(self.stateRef), (state) => state._tag === "Closing")
1450
1451/**
1452 * Checks whether the queue is done (completed or failed).
1453 *
1454 * **Example** (Checking done state)
1455 *
1456 * ```ts import.meta.vitest
1457 * import { Effect, TxQueue } from "effect"
1458 *
1459 * const program = Effect.gen(function*() {
1460 * const queue = yield* TxQueue.bounded<number>(10)
1461 *
1462 * const done = yield* TxQueue.isDone(queue)
1463 *
1464 * yield* TxQueue.interrupt(queue)
1465 * const nowDone = yield* TxQueue.isDone(queue)
1466 * return [done, nowDone] as const
1467 * })
1468 *
1469 * await Effect.runPromise(program) // => [false, true]
1470 * ```
1471 *
1472 * @category predicates
1473 * @since 4.0.0
1474 */
1475export const isDone = (self: TxQueueState): Effect.Effect<boolean> =>
1476 Effect.map(TxRef.get(self.stateRef), (state) => state._tag === "Done")
1477
1478/**
1479 * Checks whether the queue is shutdown (legacy compatibility).
1480 *
1481 * **Example** (Checking shutdown state)
1482 *
1483 * ```ts import.meta.vitest
1484 * import { Effect, TxQueue } from "effect"
1485 *
1486 * const program = Effect.gen(function*() {
1487 * const queue = yield* TxQueue.bounded<number>(10)
1488 *
1489 * const isShutdown = yield* TxQueue.isShutdown(queue)
1490 *
1491 * yield* TxQueue.shutdown(queue)
1492 * const nowShutdown = yield* TxQueue.isShutdown(queue)
1493 * return [isShutdown, nowShutdown] as const
1494 * })
1495 *
1496 * await Effect.runPromise(program) // => [false, true]
1497 * ```
1498 *
1499 * @category predicates
1500 * @since 2.0.0
1501 */
1502export const isShutdown = (self: TxQueueState): Effect.Effect<boolean> => isDone(self)
1503
1504/**
1505 * Waits for the queue to complete (either successfully or with failure).
1506 *
1507 * **Example** (Awaiting queue completion)
1508 *
1509 * ```ts import.meta.vitest
1510 * import { Effect, Fiber, TxQueue } from "effect"
1511 *
1512 * const program = Effect.gen(function*() {
1513 * const queue = yield* TxQueue.bounded<number, string>(10)
1514 *
1515 * const waiter = yield* Effect.forkChild(TxQueue.awaitCompletion(queue))
1516 * yield* TxQueue.interrupt(queue)
1517 *
1518 * yield* Fiber.join(waiter)
1519 * return "Queue completed successfully"
1520 * })
1521 *
1522 * await Effect.runPromise(program) // => "Queue completed successfully"
1523 * ```
1524 *
1525 * @category combinators
1526 * @since 4.0.0
1527 */
1528export const awaitCompletion = (self: TxQueueState): Effect.Effect<void> =>
1529 Effect.gen(function*() {
1530 const state = yield* TxRef.get(self.stateRef)
1531
1532 if (state._tag === "Done") {
1533 return void 0
1534 }
1535
1536 // Not done yet, retry transaction
1537 return yield* Effect.txRetry
1538 }).pipe(Effect.tx)