TxQueue

Transactional queues whose state changes participate in Effect transactions.

A TxQueue<A, E> stores values of type A, exposes write-only TxEnqueue and read-only TxDequeue handles, and can complete, fail, or shut down with causes observed by consumers. Queue operations can retry transactionally when they cannot proceed, such as taking from an empty open queue or offering to a full bounded queue. This makes the queue useful for coordinating producers and consumers alongside other transactional state changes.

38 exports Since v2.0.0 Source

Combinators

awaitCompletion

unstable

Since v4.0.0

Waits for the queue to complete (either successfully or with failure).

Signature

declare function awaitCompletion(self: TxQueueState): Effect<void>

Example Awaiting queue completion

import { Effect, Fiber, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)

  const waiter = yield* Effect.forkChild(TxQueue.awaitCompletion(queue))
  yield* TxQueue.interrupt(queue)

  yield* Fiber.join(waiter)
  return "Queue completed successfully"
})

await Effect.runPromise(program) // => "Queue completed successfully"

clear

unstable

Since v4.0.0

Removes and returns all currently buffered elements.

Details

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.

Signature

declare function clear<A, E>(self: TxEnqueue<A, E>): Effect<Array<A>, ExcludeDone<E>>

Example Clearing queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])

  const sizeBefore = yield* TxQueue.size(queue)

  const cleared = yield* TxQueue.clear(queue)

  const sizeAfter = yield* TxQueue.size(queue)
  return [sizeBefore, cleared, sizeAfter] as const
})

await Effect.runPromise(program) // => [5, [1, 2, 3, 4, 5], 0]

end

unstable

Since v4.0.0

Ends a queue by signaling completion with a Cause.Done error.

Details

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.

Signature

declare function end<A, E>(self: TxEnqueue<A, Done<void> | E>): Effect<boolean>

Example Ending queues

import { Cause, Effect, Exit, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, Cause.Done>(10)

  // Signal the end of the queue
  const result = yield* TxQueue.end(queue)

  // All operations will now fail with Done
  const takeResult = yield* Effect.exit(TxQueue.take(queue))

  const peekResult = yield* Effect.exit(TxQueue.peek(queue))
  return [result, takeResult, peekResult] as const
})

await Effect.runPromise(program) // => [true, Exit.fail(Cause.Done()), Exit.fail(Cause.Done())]

fail

unstable

Since v4.0.0

Fails the queue with the specified error, discarding any buffered items.

Details

The queue transitions directly to done with Cause.fail(error). Returns false if the queue was already closing or done.

Signature

declare const fail: {
  <E>(error: E): <A>(self: TxEnqueue<A, E>) => Effect<boolean>;
  <A, E>(self: TxEnqueue<A, E>, error: E): Effect<boolean>;
}

Example Failing queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)

  // Fail the queue with an error
  return yield* TxQueue.fail(queue, "connection lost")
})

await Effect.runPromise(program) // => true

failCause

unstable

Since v4.0.0

Completes the queue with the specified cause.

Details

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.

Signature

declare const failCause: {
  <E>(cause: Cause<E>): <A>(self: TxEnqueue<A, E>) => Effect<boolean>;
  <A, E>(self: TxEnqueue<A, E>, cause: Cause<E>): Effect<boolean>;
}

Example Failing queues with causes

import { Cause, Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  // Complete with specific cause
  const cause = Cause.interrupt()
  const result = yield* TxQueue.failCause(queue, cause)
  return [cause, result] as const
})

await Effect.runPromise(program) // => [Cause.interrupt(), true]

interrupt

unstable

Since v4.0.0

Interrupts the queue gracefully with the current fiber's interruption cause.

Details

If the queue still contains items, it enters the closing state so buffered items can be drained before consumers observe the interruption. If it is empty, it transitions directly to done. Returns false if the queue was already closing or done.

Signature

declare function interrupt<A, E>(self: TxEnqueue<A, E>): Effect<boolean>

Example Interrupting queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offer(queue, 42)

  // Interrupt gracefully - allows remaining items to be consumed
  return yield* TxQueue.interrupt(queue)
})

await Effect.runPromise(program) // => true

offer

unstable

Since v2.0.0

Offers an item to the queue and returns whether it was accepted.

Details

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.

Signature

declare const offer: {
  <A, E>(value: A): (self: TxEnqueue<A, E>) => Effect<boolean>;
  <A, E>(self: TxEnqueue<A, E>, value: A): Effect<boolean>;
}

Example Offering a value

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  // Offer an item - returns true if accepted
  return yield* TxQueue.offer(queue, 42)
})

await Effect.runPromise(program) // => true

offerAll

unstable

Since v2.0.0

Offers multiple items to the queue, returning the items that were not accepted.

Details

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.

Signature

declare const offerAll: {
  <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect<Array<A>>;
  <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect<Array<A>>;
}

Example Offering multiple values

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  // Offer multiple items - returns rejected items as array
  return yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
})

await Effect.runPromise(program) // => []

peek

unstable

Since v2.0.0

Waits transactionally for the next item and returns it without removing it.

Details

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.

Signature

declare function peek<A, E>(self: TxDequeue<A, E>): Effect<A, E>

Example Peeking without removing values

import { Effect, Exit, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.offer(queue, 42)

  // Peek at the next item without removing it
  const item = yield* TxQueue.peek(queue)

  // Item is still in the queue
  const size = yield* TxQueue.size(queue)
  return [item, size] as const
})

// Error handling example
const errorExample = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(5)
  yield* TxQueue.fail(queue, "queue failed")

  // peek() propagates the queue error through E-channel
  return yield* Effect.exit(TxQueue.peek(queue))
})

await Effect.runPromise(program) // => [42, 1]
await Effect.runPromise(errorExample) // => Exit.fail("queue failed")

poll

unstable

Since v2.0.0

Tries to take an item from the queue without blocking.

Signature

declare function poll<A, E>(self: TxDequeue<A, E>): Effect<Option<A>>

Example Polling without blocking

import { Effect, Option, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  // Poll returns Option.none if empty
  const maybe = yield* TxQueue.poll(queue)

  yield* TxQueue.offer(queue, 42)
  const item = yield* TxQueue.poll(queue)
  return [maybe, item] as const
})

await Effect.runPromise(program) // => [Option.none(), Option.some(42)]

shutdown

unstable

Since v2.0.0

Shuts down the queue immediately by clearing all items and interrupting it (legacy compatibility).

Details

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.

Signature

declare function shutdown<A, E>(self: TxEnqueue<A, E>): Effect<boolean>

Example Shutting down queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])

  const sizeBefore = yield* TxQueue.size(queue)

  yield* TxQueue.shutdown(queue)

  const sizeAfter = yield* TxQueue.size(queue)

  const isShutdown = yield* TxQueue.isShutdown(queue)
  return [sizeBefore, sizeAfter, isShutdown] as const
})

await Effect.runPromise(program) // => [5, 0, true]

size

unstable

Since v2.0.0

Gets the current size of the queue.

Signature

declare function size(self: TxQueueState): Effect<number>

Example Reading queue size

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offerAll(queue, [1, 2, 3])

  return yield* TxQueue.size(queue)
})

await Effect.runPromise(program) // => 3

take

unstable

Since v2.0.0

Takes the next item from the queue, retrying the transaction while the queue is empty.

Details

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.

Signature

declare function take<A, E>(self: TxDequeue<A, E>): Effect<A, E>

Example Taking a value

import { Effect, Exit, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.offer(queue, 42)

  // Take an item - blocks if empty
  const item = yield* TxQueue.take(queue)

  // When queue fails, take fails with the same error
  yield* TxQueue.fail(queue, "queue error")
  const result = yield* Effect.exit(TxQueue.take(queue))
  return [item, result] as const
})

await Effect.runPromise(program) // => [42, Exit.fail("queue error")]

takeAll

unstable

Since v2.0.0

Takes all items from the queue. Blocks if the queue is empty.

Details

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.

Signature

declare function takeAll<A, E>(self: TxDequeue<A, E>): Effect<[A, ...Array<A>], E>

Example Taking all queued values

import { Effect, Exit, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])

  // Take all items atomically - returns NonEmptyArray
  return yield* TxQueue.takeAll(queue)
})

// Error propagation example
const errorExample = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(5)
  yield* TxQueue.offerAll(queue, [1, 2])
  yield* TxQueue.fail(queue, "processing error")

  // takeAll() propagates the queue error through E-channel
  return yield* Effect.exit(TxQueue.takeAll(queue))
})

await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
await Effect.runPromise(errorExample) // => Exit.fail("processing error")

takeN

unstable

Since v2.0.0

Takes up to n items from the queue in a single transaction.

Details

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.

Signature

declare const takeN: {
  (n: number): <A, E>(self: TxDequeue<A, E>) => Effect<Array<A>, E>;
  <A, E>(self: TxDequeue<A, E>, n: number): Effect<Array<A>, E>;
}

Example Taking a fixed number of values

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(5)
  yield* TxQueue.offerAll(queue, [1, 2, 3, 4])

  const items = yield* TxQueue.takeN(queue, 4)

  // This requests more than capacity (5), so takes all available (up to 5)
  yield* TxQueue.offerAll(queue, [5, 6, 7, 8, 9])
  const all = yield* TxQueue.takeN(queue, 10)
  return [items, all] as const
})

await Effect.runPromise(program) // => [[1, 2, 3, 4], [5, 6, 7, 8, 9]]

Constructors

bounded

unstable

Since v2.0.0

Creates a new bounded TxQueue with the specified capacity.

Details

This function returns a new TxQueue reference with the specified capacity. No existing TxQueue instances are modified.

Signature

declare function bounded<A = never, E = never>(capacity: number): Effect<TxQueue<A, E>>

Example Creating bounded queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  // Create a bounded queue (E defaults to never)
  const queue = yield* TxQueue.bounded<number>(10)

  // Create a bounded queue with error channel
  const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)

  // Offer items - will succeed until capacity is reached
  yield* TxQueue.offer(queue, 1)
  yield* TxQueue.offer(queue, 2)

  return yield* TxQueue.take(queue)
})

await Effect.runPromise(program) // => 1

dropping

unstable

Since v2.0.0

Creates a new dropping TxQueue with the specified capacity that drops new items when full.

Details

This function returns a new TxQueue reference with dropping strategy. No existing TxQueue instances are modified.

Signature

declare function dropping<A = never, E = never>(capacity: number): Effect<TxQueue<A, E>>

Example Creating dropping queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  // Create a dropping queue with capacity 2
  const queue = yield* TxQueue.dropping<number>(2)

  // Fill to capacity
  yield* TxQueue.offer(queue, 1)
  yield* TxQueue.offer(queue, 2)

  // This will be dropped (returns false)
  return yield* TxQueue.offer(queue, 3)
})

await Effect.runPromise(program) // => false

sliding

unstable

Since v2.0.0

Creates a new sliding TxQueue with the specified capacity that evicts old items when full.

Details

This function returns a new TxQueue reference with sliding strategy. No existing TxQueue instances are modified.

Signature

declare function sliding<A = never, E = never>(capacity: number): Effect<TxQueue<A, E>>

Example Creating sliding queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  // Create a sliding queue with capacity 2
  const queue = yield* TxQueue.sliding<number>(2)

  // Fill to capacity
  yield* TxQueue.offer(queue, 1)
  yield* TxQueue.offer(queue, 2)

  // This will evict item 1 and add 3
  yield* TxQueue.offer(queue, 3)

  return yield* TxQueue.take(queue)
})

await Effect.runPromise(program) // => 2

unbounded

unstable

Since v2.0.0

Creates a new unbounded TxQueue with unlimited capacity.

Details

This function returns a new TxQueue reference with unlimited capacity. No existing TxQueue instances are modified.

Signature

declare function unbounded<A = never, E = never>(): Effect<TxQueue<A, E>>

Example Creating unbounded queues

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  // Create an unbounded queue (E defaults to never)
  const queue = yield* TxQueue.unbounded<string>()

  // Create an unbounded queue with error channel
  const faultTolerantQueue = yield* TxQueue.unbounded<string, Error>()

  // Can offer unlimited items
  yield* TxQueue.offer(queue, "hello")
  yield* TxQueue.offer(queue, "world")

  return yield* TxQueue.size(queue)
})

await Effect.runPromise(program) // => 2

Guards

isTxDequeue

unstable

Since v4.0.0

Checks whether the given value is a TxDequeue.

Signature

declare function isTxDequeue<A = unknown, E = unknown>(u: unknown): u is TxDequeue<A, E>

Example Checking dequeue handles

import { TxQueue } from "effect"

const someValue: unknown = {}
TxQueue.isTxDequeue(someValue) // => false

isTxEnqueue

unstable

Since v4.0.0

Checks whether the given value is a TxEnqueue.

Signature

declare function isTxEnqueue<A = unknown, E = unknown>(u: unknown): u is TxEnqueue<A, E>

Example Checking enqueue handles

import { TxQueue } from "effect"

const someValue: unknown = {}
TxQueue.isTxEnqueue(someValue) // => false

isTxQueue

unstable

Since v4.0.0

Checks whether the given value is a TxQueue.

Signature

declare function isTxQueue<A = unknown, E = unknown>(u: unknown): u is TxQueue<A, E>

Example Checking queue handles

import { TxQueue } from "effect"

const someValue: unknown = {}
TxQueue.isTxQueue(someValue) // => false

Models

State

type

Since v4.0.0

Represents the state of a transactional queue with sophisticated lifecycle management.

Details

The queue progresses through three states:

Open: Accepting offers and serving takes normally
Closing: No new offers accepted, serving remaining items until empty
Done: Terminal state with completion cause, no further operations possible

Signature

type State<_A, E> = {
  readonly _tag: "Open";
} | {
  readonly _tag: "Closing";
  readonly cause: Cause.Cause<E>;
} | {
  readonly _tag: "Done";
  readonly cause: Cause.Cause<E>;
}

Example Inspecting queue lifecycle states

import type { TxQueue } from "effect"

const state: TxQueue.State<string, Error> = { _tag: "Open" }
state._tag // => "Open"

TxDequeue

interface unstable

Since v4.0.0

A TxDequeue represents the read-only interface of a transactional queue, providing operations for consuming elements (dequeue operations) and inspecting queue state.

Signature

interface TxDequeue<out A, out E = never> extends TxQueueState {
  readonly "~effect/TxQueue/Dequeue": Variance<A, E>;
}

Example Taking values through dequeue handles

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  // Queue without error channel
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offer(queue, 42)
  const item = yield* TxQueue.take(queue)

  // Queue with error channel - errors propagate through E-channel
  const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.fail(faultTolerantQueue, "processing failed")

  // All dequeue operations now fail with the error directly
  const takeResult = yield* Effect.flip(TxQueue.take(faultTolerantQueue)) // "processing failed"
  const peekResult = yield* Effect.flip(TxQueue.peek(faultTolerantQueue)) // "processing failed"
  return [item, takeResult, peekResult] as const
})

await Effect.runPromise(program) // => [42, "processing failed", "processing failed"]

TxEnqueue

interface unstable

Since v4.0.0

A TxEnqueue represents the write-only interface of a transactional queue, providing operations for adding elements (enqueue operations) and inspecting queue state.

Signature

interface TxEnqueue<in A, in E = never> extends TxQueueState {
  readonly "~effect/TxQueue/Enqueue": Variance<A, E>;
}

Example Offering values through enqueue handles

import { Effect, TxQueue } from "effect"
import type { Cause } from "effect"

const program = Effect.gen(function*() {
  // Queue without error channel
  const queue = yield* TxQueue.bounded<number>(10)
  const accepted = yield* TxQueue.offer(queue, 42)

  // Queue with error channel for completion signaling
  const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.offerAll(faultTolerantQueue, [1, 2, 3])
  yield* TxQueue.fail(faultTolerantQueue, "processing complete")

  // Works with Done for clean completion
  const completableQueue = yield* TxQueue.bounded<
    string,
    Cause.Done
  >(5)
  yield* TxQueue.offer(completableQueue, "task")
  yield* TxQueue.end(completableQueue)

  return accepted
})

await Effect.runPromise(program) // => true

TxQueue

interface unstable

Since v4.0.0

A TxQueue represents a transactional queue data structure that provides both enqueue and dequeue operations with Software Transactional Memory (STM) semantics.

Signature

interface TxQueue<in out A, in out E = never> extends TxEnqueue<A, E>, TxDequeue<A, E> {
  readonly "~effect/TxQueue": Variance<A, E>;
}

Example Combining enqueue and dequeue operations

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  // Create a bounded transactional queue (E defaults to never)
  const queue = yield* TxQueue.bounded<number>(10)

  // Single operations - automatically transactional
  const accepted = yield* TxQueue.offer(queue, 42)
  const item = yield* TxQueue.take(queue) // Effect<number, never>

  // Queue with error channel
  const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)

  // Operations can handle queue-level failures
  yield* TxQueue.fail(faultTolerantQueue, "queue failed")
  const result = yield* Effect.flip(TxQueue.take(faultTolerantQueue))
  return [accepted, item, result] as const
})

await Effect.runPromise(program) // => [true, 42, "queue failed"]

TxQueueState

interface unstable

Since v4.0.0

Represents the shared state of a transactional queue that can be inspected. This interface contains the core properties needed for queue state inspection operations like size, capacity, and completion status.

Signature

interface TxQueueState extends Inspectable {
  readonly capacity: number;
  readonly items: TxChunk<any>;
  readonly stateRef: TxRef<State<any, any>>;
  readonly strategy: "sliding" | "dropping" | "unbounded" | "bounded";
}

Other

TxDequeue

Since v4.0.0

Namespace containing type definitions for TxDequeue variance annotations.

TxEnqueue

Since v4.0.0

Namespace containing type definitions for TxEnqueue variance annotations.

TxQueue

Since v4.0.0

Namespace containing type definitions for TxQueue variance annotations.

Predicates

isClosing

unstable

Since v4.0.0

Checks whether the queue is in the closing state.

Signature

declare function isClosing(self: TxQueueState): Effect<boolean>

Example Checking closing state

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offer(queue, 42)

  const closing = yield* TxQueue.isClosing(queue)

  yield* TxQueue.interrupt(queue)
  const nowClosing = yield* TxQueue.isClosing(queue)
  return [closing, nowClosing] as const
})

await Effect.runPromise(program) // => [false, true]

isDone

unstable

Since v4.0.0

Checks whether the queue is done (completed or failed).

Signature

declare function isDone(self: TxQueueState): Effect<boolean>

Example Checking done state

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  const done = yield* TxQueue.isDone(queue)

  yield* TxQueue.interrupt(queue)
  const nowDone = yield* TxQueue.isDone(queue)
  return [done, nowDone] as const
})

await Effect.runPromise(program) // => [false, true]

isEmpty

unstable

Since v2.0.0

Checks whether the queue is empty.

Signature

declare function isEmpty(self: TxQueueState): Effect<boolean>

Example Checking whether a queue is empty

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  const empty = yield* TxQueue.isEmpty(queue)

  yield* TxQueue.offer(queue, 42)
  const stillEmpty = yield* TxQueue.isEmpty(queue)
  return [empty, stillEmpty] as const
})

await Effect.runPromise(program) // => [true, false]

isFull

unstable

Since v2.0.0

Checks whether the queue is at capacity.

Signature

declare function isFull(self: TxQueueState): Effect<boolean>

Example Checking whether a queue is full

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(2)

  const full = yield* TxQueue.isFull(queue)

  yield* TxQueue.offerAll(queue, [1, 2])
  const nowFull = yield* TxQueue.isFull(queue)
  return [full, nowFull] as const
})

await Effect.runPromise(program) // => [false, true]

isNonEmpty

unstable

Since v4.0.0

Checks whether the queue is non-empty.

Signature

declare function isNonEmpty(self: TxQueueState): Effect<boolean>

Example Checking whether a queue is non-empty

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  const empty = yield* TxQueue.isNonEmpty(queue)

  yield* TxQueue.offer(queue, 42)
  const nonEmpty = yield* TxQueue.isNonEmpty(queue)
  return [empty, nonEmpty] as const
})

await Effect.runPromise(program) // => [false, true]

isOpen

unstable

Since v4.0.0

Checks whether the queue is in the open state.

Signature

declare function isOpen(self: TxQueueState): Effect<boolean>

Example Checking open state

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  const open = yield* TxQueue.isOpen(queue)

  yield* TxQueue.interrupt(queue)
  const stillOpen = yield* TxQueue.isOpen(queue)
  return [open, stillOpen] as const
})

await Effect.runPromise(program) // => [true, false]

isShutdown

unstable

Since v2.0.0

Checks whether the queue is shutdown (legacy compatibility).

Signature

declare function isShutdown(self: TxQueueState): Effect<boolean>

Example Checking shutdown state

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)

  const isShutdown = yield* TxQueue.isShutdown(queue)

  yield* TxQueue.shutdown(queue)
  const nowShutdown = yield* TxQueue.isShutdown(queue)
  return [isShutdown, nowShutdown] as const
})

await Effect.runPromise(program) // => [false, true]

Taking

takeBetween

unstable

Since v2.0.0

Takes between min and max currently available items, waiting for min on an open queue.

Details

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.

Signature

declare const takeBetween: {
  (min: number, max: number): <A, E>(self: TxDequeue<A, E>) => Effect<Array<A>, E>;
  <A, E>(self: TxDequeue<A, E>, min: number, max: number): Effect<Array<A>, E>;
}

Example Taking batches within bounds

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number>(10)
  yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8])

  // Take between 2 and 5 items
  const batch1 = yield* TxQueue.takeBetween(queue, 2, 5)

  // Take between 1 and 10 items (but only 3 remain)
  const batch2 = yield* TxQueue.takeBetween(queue, 1, 10)

  // Would wait for at least 1 item to be available
  // const batch3 = yield* TxQueue.takeBetween(queue, 1, 3)
  return [batch1, batch2] as const
})

await Effect.runPromise(program) // => [[1, 2, 3, 4, 5], [6, 7, 8]]