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.

37 exports Added in v2.0.0 Source

Combinators

awaitCompletion

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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. If n is less than or equal to zero, 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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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/transactions/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

Added in v4.0.0 Source

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/transactions/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

Added in v4.0.0 Source

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/transactions/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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

Namespace containing type definitions for TxDequeue variance annotations.

TxEnqueue

Added in v4.0.0 Source

Namespace containing type definitions for TxEnqueue variance annotations.

TxQueue

Added in v4.0.0 Source

Namespace containing type definitions for TxQueue variance annotations.

Predicates

isClosing

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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]

isOpen

Added in v4.0.0 Source

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

Added in v2.0.0 Source

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

Added in v2.0.0 Source

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

Details

If the queue is closing, drains the currently available items even when fewer than min are available and transitions to Done. Invalid 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]]