/**
* Passes values asynchronously between fibers.
*
* A `Queue` accepts values, hands each value to one consumer in offer
* order, and can complete, fail, interrupt, or shut down. Queues can be bounded
* or unbounded, and bounded queues can suspend, drop, or slide values when
* producers are faster than consumers.
*
* @stability stable
* @since 3.8.0
*/
import * as Arr from "./Array.ts";
import type { Cause, Done } from "./Cause.ts";
import type { Effect } from "./Effect.ts";
import type { Exit, Failure } from "./Exit.ts";
import type { Inspectable } from "./Inspectable.ts";
import * as MutableList from "./MutableList.ts";
import * as Option from "./Option.ts";
import * as Pull from "./Pull.ts";
import type { SchedulerDispatcher } from "./Scheduler.ts";
import type * as Types from "./Types.ts";
declare const TypeId = "~effect/Queue";
declare const EnqueueTypeId = "~effect/Queue/Enqueue";
declare const DequeueTypeId = "~effect/Queue/Dequeue";
/**
* Type guard to check if a value is a Queue.
*
* **When to use**
*
* Use to narrow an unknown value to a full `Queue` before passing it to APIs
* that need both offering and taking capabilities.
*
* @see {@link isEnqueue} for checking values that only need write access
* @see {@link isDequeue} for checking values that only need read access
*
* @stability stable
* @category guards
* @since 2.0.0
*/
export declare const isQueue: (u: unknown) => u is Queue;
/**
* Type guard to check if a value is an Enqueue.
*
* **When to use**
*
* Use to narrow an unknown value before calling queue operations that require
* write-side access.
*
* **Gotchas**
*
* A full `Queue` also satisfies this guard because every queue includes the
* enqueue side.
*
* @see {@link isQueue} for checking for a full read-write queue handle
* @see {@link isDequeue} for checking for the read side of a queue
* @see {@link asEnqueue} for narrowing an existing `Queue` to its write-only interface
*
* @stability stable
* @category guards
* @since 2.0.0
*/
export declare const isEnqueue: (u: unknown) => u is Enqueue;
/**
* Type guard to check if a value is a Dequeue.
*
* **When to use**
*
* Use to narrow an unknown value before passing it to read-side queue
* operations.
*
* @see {@link Dequeue} for the read-side queue handle checked by this guard
* @see {@link isQueue} for checking for a full read-write queue handle
* @see {@link isEnqueue} for checking for the write side of a queue
* @see {@link asDequeue} for narrowing an existing `Queue` to its read-only interface
*
* @stability stable
* @category guards
* @since 2.0.0
*/
export declare const isDequeue: (u: unknown) => u is Dequeue;
/**
* Converts a `Queue` to its write-only `Enqueue` interface.
*
* **When to use**
*
* Use to expose only the producer side of a `Queue` to code that should offer
* values or signal queue lifecycle.
*
* **Gotchas**
*
* This is a type-level capability restriction. It returns the same queue
* object, so it does not hide read operations at runtime.
*
* @see {@link asDequeue} for exposing only the read side of a `Queue`
* @see {@link Enqueue} for the write-only queue handle returned by this conversion
*
* @stability stable
* @category converting
* @since 4.0.0
*/
export declare const asEnqueue: (self: Queue) => Enqueue;
/**
* Narrows a `Queue` to a `Dequeue`, exposing the consumer side of the queue.
*
* **When to use**
*
* Use to pass a queue to code that should consume values while keeping
* producer-side operations out of that code's TypeScript type.
*
* **Gotchas**
*
* This is a type-level narrowing operation. It returns the same queue object
* and does not create a runtime wrapper.
*
* @see {@link asEnqueue} for narrowing a queue to its producer side
* @see {@link Dequeue} for the consumer-side queue handle returned by this function
*
* @stability stable
* @category converting
* @since 4.0.0
*/
export declare const asDequeue: (self: Queue) => Dequeue;
/**
* An `Enqueue` is a queue that can be offered to.
*
* **Details**
*
* This interface represents the write-only part of a Queue, allowing you to offer
* elements to the queue but not take elements from it.
*
* **Example** (Offering through enqueue handles)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* // Function that only needs write access to a queue
* const producer = (enqueue: Queue.Enqueue) =>
* Effect.gen(function*() {
* yield* Queue.offer(enqueue, "hello")
* yield* Queue.offerAll(enqueue, ["world", "!"])
* })
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
* yield* producer(queue)
* return yield* Queue.takeAll(queue)
* })
*
* await Effect.runPromise(program) // => ["hello", "world", "!"]
* ```
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface Enqueue extends Inspectable {
readonly [EnqueueTypeId]: Enqueue.Variance;
readonly strategy: "suspend" | "dropping" | "sliding";
readonly dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList;
state: Queue.State;
scheduleRunning: boolean;
}
/**
* Companion namespace containing type-level metadata for the `Enqueue`
* write-only queue interface.
*
* @stability stable
* @since 2.0.0
*/
export declare namespace Enqueue {
/**
* Type-level variance marker for `Enqueue`.
*
* **Details**
*
* `Enqueue` is contravariant in both its offered value type `A` and failure
* type `E`, because values and failures flow into the queue through this
* handle.
*
* @category models
* @since 4.0.0
*/
interface Variance {
_A: Types.Contravariant;
_E: Types.Contravariant;
}
}
/**
* A `Dequeue` is a queue that can be taken from.
*
* **Details**
*
* This interface represents the read-only part of a Queue, allowing you to take
* elements from the queue but not offer elements to it.
*
* **Example** (Taking through dequeue handles)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // A Dequeue can only take elements
* const dequeue: Queue.Dequeue = queue
*
* // Pre-populate the queue
* yield* Queue.offerAll(queue, ["a", "b", "c"])
*
* // Take elements using dequeue interface
* const item = yield* Queue.take(dequeue)
* return item
* })
*
* await Effect.runPromise(program) // => "a"
* ```
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface Dequeue extends Inspectable {
readonly [DequeueTypeId]: Dequeue.Variance;
readonly strategy: "suspend" | "dropping" | "sliding";
readonly dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList;
state: Queue.State;
scheduleRunning: boolean;
}
/**
* Companion namespace containing type-level metadata for the `Dequeue`
* read-only queue interface.
*
* @stability stable
* @since 2.0.0
*/
export declare namespace Dequeue {
/**
* Type-level variance marker for `Dequeue`.
*
* **Details**
*
* `Dequeue` is covariant in both the taken value type `A` and failure type
* `E`, because values and failures are observed through this handle.
*
* @category models
* @since 4.0.0
*/
interface Variance {
_A: Types.Covariant;
_E: Types.Covariant;
}
}
/**
* A `Queue` is an asynchronous queue that can be offered to and taken from.
*
* **Details**
*
* It also supports signaling that it is done or failed.
*
* **Example** (Offering and taking queue values)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* // Create a bounded queue
* const queue = yield* Queue.bounded(10)
*
* // Producer: offer items to the queue
* yield* Queue.offer(queue, "hello")
* yield* Queue.offerAll(queue, ["world", "!"])
*
* // Consumer: take items from the queue
* const item1 = yield* Queue.take(queue)
* const item2 = yield* Queue.take(queue)
* const item3 = yield* Queue.take(queue)
*
* return [item1, item2, item3]
* })
*
* await Effect.runPromise(program) // => ["hello", "world", "!"]
* ```
*
* @stability stable
* @category models
* @since 2.0.0
*/
export interface Queue extends Enqueue, Dequeue {
readonly [TypeId]: Queue.Variance;
}
/**
* Companion namespace containing type-level metadata and low-level state types
* for `Queue`.
*
* @stability stable
* @since 2.0.0
*/
export declare namespace Queue {
/**
* Type-level variance marker for `Queue`.
*
* **Details**
*
* A full `Queue` is invariant in both `A` and `E` because the same handle can
* both produce and consume values and failures.
*
* @category models
* @since 4.0.0
*/
interface Variance {
_A: Types.Invariant;
_E: Types.Invariant;
}
/**
* Tagged state of a `Queue`.
*
* **Details**
*
* `Open` queues can accept offers and takers, `Closing` queues are
* completing with a stored failure exit, and `Done` queues have finished.
* This is low-level metadata exposed by the queue model; most users should
* inspect queues through the public operations.
*
* @category models
* @since 4.0.0
*/
type State = {
readonly _tag: "Open";
readonly takers: Set>;
readonly offers: Set>;
readonly awaiters: Set<(_: Effect) => void>;
} | {
readonly _tag: "Closing";
readonly takers: Set>;
readonly offers: Set>;
readonly awaiters: Set<(_: Effect) => void>;
readonly exit: Failure;
} | {
readonly _tag: "Done";
readonly exit: Failure;
};
/**
* Represents a suspended offer waiting to be admitted to a bounded queue.
*
* **Details**
*
* An entry is either a single message or a batch with an offset into its
* remaining messages, plus a resume callback that completes the suspended
* offer when the queue can accept more input.
*
* @category models
* @since 4.0.0
*/
type OfferEntry = {
readonly _tag: "Array";
readonly remaining: Array;
offset: number;
readonly resume: (_: Effect>) => void;
} | {
readonly _tag: "Single";
readonly message: A;
readonly resume: (_: Effect) => void;
};
/**
* Represents a suspended take waiting for the queue to become readable.
*
* **Details**
*
* `ready` reports whether the take can now complete, so a batch take is
* only resumed once its minimum is available. `resume` completes the
* suspended take, with `void` to retry or with a failure exit when the queue
* is done.
*
* @category models
* @since 4.0.0
*/
interface Taker {
readonly ready: () => boolean;
readonly resume: (_: Effect) => void;
}
}
/**
* Creates a `Queue` with optional capacity and overflow strategy.
*
* **Details**
*
* By default the queue is unbounded and uses the `"suspend"` strategy. Provide
* `capacity` for a bounded queue and choose `"suspend"`, `"dropping"`, or
* `"sliding"` to control what happens when the queue is full. The returned
* queue can be offered to, taken from, failed, ended, interrupted, or shut down.
*
* **Example** (Creating queues)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.make()
*
* // add messages to the queue
* yield* Queue.offer(queue, 1)
* yield* Queue.offer(queue, 2)
* yield* Queue.offerAll(queue, [3, 4, 5])
*
* // take messages from the queue
* const messages = yield* Queue.takeAll(queue)
*
* // signal that the queue is done
* yield* Queue.end(queue)
* const done = yield* Effect.flip(Queue.take(queue))
*
* // signal that another queue has failed
* const failedQueue = yield* Queue.make()
* const failed = yield* Queue.fail(failedQueue, "boom")
* return { messages, done, failed }
* })
*
* await Effect.runPromise(program) // => { messages: [1, 2, 3, 4, 5], done: Cause.Done(), failed: true }
* ```
*
* @stability stable
* @category constructors
* @since 4.0.0
*/
export declare const make: (options?: {
readonly capacity?: number | undefined;
readonly strategy?: "suspend" | "dropping" | "sliding" | undefined;
} | undefined) => Effect>;
/**
* Creates a bounded queue with the specified capacity that uses backpressure strategy.
*
* **Details**
*
* When the queue reaches capacity, producers will be suspended until space becomes available.
* This ensures all messages are processed but may slow down producers.
*
* **Example** (Creating bounded queues)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(5)
*
* // This will succeed as queue has capacity
* yield* Queue.offer(queue, "first")
* yield* Queue.offer(queue, "second")
*
* const size = yield* Queue.size(queue)
* return size
* })
*
* await Effect.runPromise(program) // => 2
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export declare const bounded: (capacity: number) => Effect>;
/**
* Creates a bounded queue with sliding strategy. When the queue reaches capacity,
* new elements are added and the oldest elements are dropped.
*
* **When to use**
*
* Use when you need producer offers not to block and can accept dropping the
* oldest messages, such as when maintaining a rolling window of recent values.
*
* **Example** (Creating sliding queues)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.sliding(3)
*
* // Fill the queue to capacity
* yield* Queue.offer(queue, 1)
* yield* Queue.offer(queue, 2)
* yield* Queue.offer(queue, 3)
*
* // This will succeed, dropping the oldest element (1)
* yield* Queue.offer(queue, 4)
*
* const all = yield* Queue.takeAll(queue)
* return all
* })
*
* await Effect.runPromise(program) // => [2, 3, 4]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export declare const sliding: (capacity: number) => Effect>;
/**
* Creates a bounded queue with dropping strategy. When the queue reaches capacity,
* new elements are dropped and the offer operation returns false.
*
* **When to use**
*
* Use when you need producer offers not to block while preserving existing
* queued messages, even if new messages may be dropped when the queue is full.
*
* **Example** (Creating dropping queues)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.dropping(2)
*
* // Fill the queue to capacity
* const success1 = yield* Queue.offer(queue, 1)
* const success2 = yield* Queue.offer(queue, 2)
*
* // This will be dropped
* const success3 = yield* Queue.offer(queue, 3)
*
* const all = yield* Queue.takeAll(queue)
* return [success1, success2, success3, all]
* })
*
* await Effect.runPromise(program) // => [true, true, false, [1, 2]]
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export declare const dropping: (capacity: number) => Effect>;
/**
* Creates an unbounded queue that can grow to any size without blocking producers.
*
* **When to use**
*
* Use when you need producers to add messages without backpressure and accept
* unbounded memory growth.
*
* **Example** (Creating unbounded queues)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.unbounded()
*
* // Producers can always add messages without blocking
* yield* Queue.offer(queue, "message1")
* yield* Queue.offer(queue, "message2")
* yield* Queue.offerAll(queue, ["message3", "message4", "message5"])
*
* // Check current size
* const size = yield* Queue.size(queue)
*
* // Take all messages
* const messages = yield* Queue.takeAll(queue)
* return { size, messages }
* })
*
* await Effect.runPromise(program) // => { size: 5, messages: ["message1", "message2", "message3", "message4", "message5"] }
* ```
*
* @stability stable
* @category constructors
* @since 2.0.0
*/
export declare const unbounded: () => Effect>;
/**
* Adds a message to the queue. Returns `false` if the queue is done.
*
* **Details**
*
* For bounded queues, this operation may suspend if the queue is at capacity,
* depending on the backpressure strategy. For dropping/sliding queues, it may
* return false or succeed immediately by dropping/sliding existing messages.
*
* **Example** (Offering a value)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* // Successfully add messages to queue
* const success1 = yield* Queue.offer(queue, 1)
* const success2 = yield* Queue.offer(queue, 2)
*
* // Queue state
* const size = yield* Queue.size(queue)
* return { offered: [success1, success2], size }
* })
*
* await Effect.runPromise(program) // => { offered: [true, true], size: 2 }
* ```
*
* @stability stable
* @category offering
* @since 2.0.0
*/
export declare const offer: {
/**
* Adds a message to the queue. Returns `false` if the queue is done.
*
* **Details**
*
* For bounded queues, this operation may suspend if the queue is at capacity,
* depending on the backpressure strategy. For dropping/sliding queues, it may
* return false or succeed immediately by dropping/sliding existing messages.
*
* **Example** (Offering a value)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* // Successfully add messages to queue
* const success1 = yield* Queue.offer(queue, 1)
* const success2 = yield* Queue.offer(queue, 2)
*
* // Queue state
* const size = yield* Queue.size(queue)
* return { offered: [success1, success2], size }
* })
*
* await Effect.runPromise(program) // => { offered: [true, true], size: 2 }
* ```
*
* @stability stable
* @category offering
* @since 2.0.0
*/
(message: A): (self: Enqueue) => Effect;
/**
* Adds a message to the queue. Returns `false` if the queue is done.
*
* **Details**
*
* For bounded queues, this operation may suspend if the queue is at capacity,
* depending on the backpressure strategy. For dropping/sliding queues, it may
* return false or succeed immediately by dropping/sliding existing messages.
*
* **Example** (Offering a value)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* // Successfully add messages to queue
* const success1 = yield* Queue.offer(queue, 1)
* const success2 = yield* Queue.offer(queue, 2)
*
* // Queue state
* const size = yield* Queue.size(queue)
* return { offered: [success1, success2], size }
* })
*
* await Effect.runPromise(program) // => { offered: [true, true], size: 2 }
* ```
*
* @stability stable
* @category offering
* @since 2.0.0
*/
(self: Enqueue, message: Types.NoInfer): Effect;
};
/**
* Adds a message to the queue synchronously. Returns `false` without adding the
* message if the queue is closing or done, or if it is full with a non-sliding
* strategy (`"suspend"` or `"dropping"`). A queue with non-positive capacity can
* still accept a message when a taker is waiting.
*
* **When to use**
*
* Use when you are already in synchronous queue internals or a performance
* boundary where wrapping the mutation in `Effect` is intentionally avoided.
*
* **Gotchas**
*
* This is an unsafe operation that directly modifies the queue without Effect wrapping.
* Use this only when you're certain about the synchronous nature of the operation.
*
* **Example** (Offering a value synchronously)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* // Create a queue effect and extract the queue for unsafe operations
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* // Add messages synchronously using unsafe API
* const success1 = Queue.offerUnsafe(queue, 1)
* const success2 = Queue.offerUnsafe(queue, 2)
*
* // Check current size
* const size = Queue.sizeUnsafe(queue)
* return { offered: [success1, success2], size }
* })
*
* await Effect.runPromise(program) // => { offered: [true, true], size: 2 }
* ```
*
* @stability stable
* @category offering
* @since 4.0.0
*/
export declare const offerUnsafe: (self: Enqueue, message: Types.NoInfer) => boolean;
/**
* Adds multiple messages to the queue. Returns the remaining messages that
* were not added.
*
* **When to use**
*
* Use when producers can submit a batch at once and need to know which messages
* did not fit under the queue's capacity strategy.
*
* **Details**
*
* For bounded queues, this operation may suspend if the queue doesn't have
* enough capacity. The operation returns an array of messages that couldn't
* be added (empty array means all messages were successfully added).
*
* **Example** (Offering multiple values)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.dropping(3)
*
* // Try to add more messages than capacity without suspending
* const remaining1 = yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
* return remaining1
* })
*
* await Effect.runPromise(program) // => [4, 5]
* ```
*
* @stability stable
* @category offering
* @since 2.0.0
*/
export declare const offerAll: {
/**
* Adds multiple messages to the queue. Returns the remaining messages that
* were not added.
*
* **When to use**
*
* Use when producers can submit a batch at once and need to know which messages
* did not fit under the queue's capacity strategy.
*
* **Details**
*
* For bounded queues, this operation may suspend if the queue doesn't have
* enough capacity. The operation returns an array of messages that couldn't
* be added (empty array means all messages were successfully added).
*
* **Example** (Offering multiple values)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.dropping(3)
*
* // Try to add more messages than capacity without suspending
* const remaining1 = yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
* return remaining1
* })
*
* await Effect.runPromise(program) // => [4, 5]
* ```
*
* @stability stable
* @category offering
* @since 2.0.0
*/
(messages: Iterable): (self: Enqueue) => Effect>;
/**
* Adds multiple messages to the queue. Returns the remaining messages that
* were not added.
*
* **When to use**
*
* Use when producers can submit a batch at once and need to know which messages
* did not fit under the queue's capacity strategy.
*
* **Details**
*
* For bounded queues, this operation may suspend if the queue doesn't have
* enough capacity. The operation returns an array of messages that couldn't
* be added (empty array means all messages were successfully added).
*
* **Example** (Offering multiple values)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.dropping(3)
*
* // Try to add more messages than capacity without suspending
* const remaining1 = yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
* return remaining1
* })
*
* await Effect.runPromise(program) // => [4, 5]
* ```
*
* @stability stable
* @category offering
* @since 2.0.0
*/
(self: Enqueue, messages: Iterable): Effect>;
};
/**
* Adds multiple messages to the queue synchronously. Returns the remaining messages that
* were not added.
*
* **When to use**
*
* Use when queue internals or a performance boundary need a synchronous batch
* offer and can handle any messages that do not fit.
*
* **Gotchas**
*
* This is an unsafe operation that directly modifies the queue without Effect wrapping.
*
* **Example** (Offering multiple values synchronously)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* // Create a bounded queue and use unsafe API
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* // Try to add 5 messages to capacity-3 queue using unsafe API
* const remaining = Queue.offerAllUnsafe(queue, [1, 2, 3, 4, 5])
*
* // Check what's in the queue
* const size = Queue.sizeUnsafe(queue)
* return { remaining, size }
* })
*
* await Effect.runPromise(program) // => { remaining: [4, 5], size: 3 }
* ```
*
* @stability stable
* @category offering
* @since 4.0.0
*/
export declare const offerAllUnsafe: (self: Enqueue, messages: Iterable) => Array;
/**
* Fails the queue with an error. If the queue is already done, `false` is
* returned.
*
* **Example** (Failing queues with an error)
*
* ```ts import.meta.vitest
* import { Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Fail the queue with an error
* const failed = yield* Queue.fail(queue, "Something went wrong")
*
* // Taking from the failed queue fails with the error
* const exit = yield* Effect.exit(Queue.take(queue))
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.fail("Something went wrong")]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const fail: {
/**
* Fails the queue with an error. If the queue is already done, `false` is
* returned.
*
* **Example** (Failing queues with an error)
*
* ```ts import.meta.vitest
* import { Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Fail the queue with an error
* const failed = yield* Queue.fail(queue, "Something went wrong")
*
* // Taking from the failed queue fails with the error
* const exit = yield* Effect.exit(Queue.take(queue))
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.fail("Something went wrong")]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
(error: E): (self: Enqueue) => Effect;
/**
* Fails the queue with an error. If the queue is already done, `false` is
* returned.
*
* **Example** (Failing queues with an error)
*
* ```ts import.meta.vitest
* import { Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Fail the queue with an error
* const failed = yield* Queue.fail(queue, "Something went wrong")
*
* // Taking from the failed queue fails with the error
* const exit = yield* Effect.exit(Queue.take(queue))
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.fail("Something went wrong")]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
(self: Enqueue, error: E): Effect;
};
/**
* Fails the queue with a cause. If the queue is already done, `false` is
* returned.
*
* **Example** (Failing queues with a cause)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create a cause and fail the queue
* const cause = Cause.fail("Queue processing failed")
* const failed = yield* Queue.failCause(queue, cause)
*
* // The queue is now done with the specified failure cause
* const exit = yield* Effect.exit(Queue.take(queue))
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.failCause(Cause.fail("Queue processing failed"))]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const failCause: {
/**
* Fails the queue with a cause. If the queue is already done, `false` is
* returned.
*
* **Example** (Failing queues with a cause)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create a cause and fail the queue
* const cause = Cause.fail("Queue processing failed")
* const failed = yield* Queue.failCause(queue, cause)
*
* // The queue is now done with the specified failure cause
* const exit = yield* Effect.exit(Queue.take(queue))
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.failCause(Cause.fail("Queue processing failed"))]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
(cause: Cause): (self: Enqueue) => Effect;
/**
* Fails the queue with a cause. If the queue is already done, `false` is
* returned.
*
* **Example** (Failing queues with a cause)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create a cause and fail the queue
* const cause = Cause.fail("Queue processing failed")
* const failed = yield* Queue.failCause(queue, cause)
*
* // The queue is now done with the specified failure cause
* const exit = yield* Effect.exit(Queue.take(queue))
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.failCause(Cause.fail("Queue processing failed"))]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
(self: Enqueue, cause: Cause): Effect;
};
/**
* Fails the queue with a cause synchronously. If the queue is already done, `false` is
* returned.
*
* **When to use**
*
* Use when queue completion must be driven from synchronous internals while
* preserving the full failure `Cause`.
*
* **Gotchas**
*
* This is an unsafe operation that directly modifies the queue without Effect wrapping.
*
* **Example** (Failing queues with a cause synchronously)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create a cause and fail the queue synchronously
* const cause = Cause.fail("Processing error")
* const failed = Queue.failCauseUnsafe(queue, cause)
*
* // The queue is now done with the specified failure cause
* const exit = Queue.takeUnsafe(queue)
* return [failed, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.failCause(Cause.fail("Processing error"))]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const failCauseUnsafe: (self: Enqueue, cause: Cause) => boolean;
/**
* Signals queue completion.
*
* **When to use**
*
* Use to stop accepting new offers while allowing already queued messages to be
* consumed.
*
* **Details**
*
* Returns `false` if the queue is already done.
*
* **Example** (Ending queues)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add some messages
* yield* Queue.offer(queue, 1)
* yield* Queue.offer(queue, 2)
*
* // Signal completion - no more messages will be accepted
* const ended = yield* Queue.end(queue)
*
* // Trying to offer more messages will return false
* const offerResult = yield* Queue.offer(queue, 3)
*
* // But we can still take existing messages
* const message = yield* Queue.take(queue)
* return [ended, offerResult, message]
* })
*
* await Effect.runPromise(program) // => [true, false, 1]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const end: (self: Enqueue) => Effect;
/**
* Signals queue completion synchronously.
*
* **When to use**
*
* Use when implementing low-level queue integrations that must complete a queue
* without wrapping the operation in `Effect`.
*
* **Details**
*
* Returns `false` if the queue is already done.
*
* **Gotchas**
*
* This is an unsafe operation that directly modifies the queue without Effect wrapping.
*
* **Example** (Ending queues synchronously)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* // Create a queue and use unsafe operations
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add some messages
* Queue.offerUnsafe(queue, 1)
* Queue.offerUnsafe(queue, 2)
*
* // End the queue synchronously
* const ended = Queue.endUnsafe(queue)
*
* // Existing messages can still be consumed while the queue is closing
* const states = [queue.state._tag]
*
* Queue.takeUnsafe(queue)
* Queue.takeUnsafe(queue)
*
* // After buffered messages are consumed, the queue is done
* states.push(queue.state._tag)
* return { ended, states }
* })
*
* await Effect.runPromise(program) // => { ended: true, states: ["Closing", "Done"] }
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const endUnsafe: (self: Enqueue) => boolean;
/**
* Interrupts the queue gracefully, transitioning it to a closing state.
*
* **Details**
*
* This operation stops accepting new offers but allows existing messages to be consumed.
* Once all messages are drained, the queue transitions to the Done state with an interrupt cause.
*
* **Example** (Interrupting queues gracefully)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add some messages
* yield* Queue.offer(queue, 1)
* yield* Queue.offer(queue, 2)
*
* // Interrupt gracefully - no more offers accepted, but messages can be consumed
* const interrupted = yield* Queue.interrupt(queue)
*
* // Trying to offer more messages will return false
* const offerResult = yield* Queue.offer(queue, 3)
*
* // But we can still take existing messages
* const message1 = yield* Queue.take(queue)
*
* const message2 = yield* Queue.take(queue)
*
* // After all messages are consumed, queue is done
* const isDone = queue.state._tag === "Done"
* return { interrupted, offerResult, messages: [message1, message2], isDone }
* })
*
* await Effect.runPromise(program) // => { interrupted: true, offerResult: false, messages: [1, 2], isDone: true }
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const interrupt: (self: Enqueue) => Effect;
/**
* Shuts down the queue immediately, discarding buffered messages and resuming
* pending operations.
*
* **Details**
*
* Returns `true` when the queue is shut down by this call, or `false` when it
* has already been shut down or completed.
*
* **Example** (Shutting down queues)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(2)
*
* // Add messages
* yield* Queue.offer(queue, 1)
* yield* Queue.offer(queue, 2)
*
* // Shutdown clears buffered messages and prevents further offers
* const wasShutdown = yield* Queue.shutdown(queue)
*
* // Queue is now done and cleared
* const size = yield* Queue.size(queue)
* return { wasShutdown, size }
* })
*
* await Effect.runPromise(program) // => { wasShutdown: true, size: 0 }
* ```
*
* @see {@link shutdownUnsafe} for synchronous shutdown
* @stability stable
* @category completion
* @since 2.0.0
*/
export declare const shutdown: (self: Enqueue) => Effect;
/**
* Shuts down the queue synchronously, discarding buffered messages and resuming
* pending operations.
*
* **When to use**
*
* Use when a synchronous callback must discard buffered messages and settle
* pending queue operations before returning.
*
* **Details**
*
* An open queue completes with an interruption. A queue already closing retains
* its completion cause. Call `failCauseUnsafe` first to shut down with a specific
* failure. Returns `true` when the queue is shut down by this call, or `false`
* when it has already been shut down or completed.
*
* @see {@link shutdown} for the effectful variant
* @see {@link failCauseUnsafe} to set a failure before discarding buffered messages
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const shutdownUnsafe: (self: Enqueue) => boolean;
/**
* Takes and returns all currently buffered messages without waiting for more.
*
* **Details**
*
* Returns an empty array when the queue is empty or has completed normally. If
* the queue has failed, the effect fails with the queue's error.
*
* **Example** (Clearing queued values)
*
* ```ts import.meta.vitest
* import { Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
*
* // Clear all messages from the queue
* const messages = yield* Queue.clear(queue)
*
* // Queue is now empty
* const size = yield* Queue.size(queue)
*
* // Clearing empty queue returns empty array
* const empty = yield* Queue.clear(queue)
* return { messages, size, empty }
* })
*
* await Effect.runPromise(program) // => { messages: [1, 2, 3, 4, 5], size: 0, empty: [] }
* ```
*
* @stability stable
* @category taking
* @since 4.0.0
*/
export declare const clear: (self: Dequeue) => Effect, Pull.ExcludeDone>;
/**
* Takes all currently available messages, waiting until at least one message
* is available when the queue is empty.
*
* **When to use**
*
* Use when consumers should process the next non-empty batch of buffered
* messages instead of repeatedly taking one message at a time.
*
* **Details**
*
* Returns a non-empty array. If the queue completes or fails before a message
* can be taken, the effect fails with the queue's terminal error.
*
* **Example** (Taking all available values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(5)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
*
* // Take all available messages
* const messages1 = yield* Queue.takeAll(queue)
* return messages1
* })
*
* await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
export declare const takeAll: (self: Dequeue) => Effect, E>;
/**
* Takes all messages from the queue, until the queue has errored or is done.
*
* **Example** (Collecting values until completion)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(5)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
* yield* Queue.end(queue)
*
* // Collect all available messages
* return yield* Queue.collect(queue)
* })
*
* await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
* ```
*
* @stability stable
* @category taking
* @since 4.0.0
*/
export declare const collect: (self: Dequeue) => Effect, Pull.ExcludeDone>;
/**
* Takes up to `n` messages from the queue.
*
* **Details**
*
* The operation may wait until enough messages are available to satisfy the
* queue's batching rules. Finite fractional values of `n` are rounded down.
* If `n` is `NaN` or non-positive, it succeeds with an empty array. If the
* queue is closing, drains the currently available messages even when fewer
* than `n` are available. Once the queue is done, the effect fails with the
* queue's terminal error.
*
* **Example** (Taking a fixed number of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7])
*
* // Take exactly 3 messages
* const first3 = yield* Queue.takeN(queue, 3)
*
* // Take exactly 2 more messages
* const next2 = yield* Queue.takeN(queue, 2)
*
* // Take remaining messages
* const remaining = yield* Queue.takeN(queue, 2)
* return [first3, next2, remaining]
* })
*
* await Effect.runPromise(program) // => [[1, 2, 3], [4, 5], [6, 7]]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
export declare const takeN: {
/**
* Takes up to `n` messages from the queue.
*
* **Details**
*
* The operation may wait until enough messages are available to satisfy the
* queue's batching rules. Finite fractional values of `n` are rounded down.
* If `n` is `NaN` or non-positive, it succeeds with an empty array. If the
* queue is closing, drains the currently available messages even when fewer
* than `n` are available. Once the queue is done, the effect fails with the
* queue's terminal error.
*
* **Example** (Taking a fixed number of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7])
*
* // Take exactly 3 messages
* const first3 = yield* Queue.takeN(queue, 3)
*
* // Take exactly 2 more messages
* const next2 = yield* Queue.takeN(queue, 2)
*
* // Take remaining messages
* const remaining = yield* Queue.takeN(queue, 2)
* return [first3, next2, remaining]
* })
*
* await Effect.runPromise(program) // => [[1, 2, 3], [4, 5], [6, 7]]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
(n: number): (self: Dequeue) => Effect, E>;
/**
* Takes up to `n` messages from the queue.
*
* **Details**
*
* The operation may wait until enough messages are available to satisfy the
* queue's batching rules. Finite fractional values of `n` are rounded down.
* If `n` is `NaN` or non-positive, it succeeds with an empty array. If the
* queue is closing, drains the currently available messages even when fewer
* than `n` are available. Once the queue is done, the effect fails with the
* queue's terminal error.
*
* **Example** (Taking a fixed number of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7])
*
* // Take exactly 3 messages
* const first3 = yield* Queue.takeN(queue, 3)
*
* // Take exactly 2 more messages
* const next2 = yield* Queue.takeN(queue, 2)
*
* // Take remaining messages
* const remaining = yield* Queue.takeN(queue, 2)
* return [first3, next2, remaining]
* })
*
* await Effect.runPromise(program) // => [[1, 2, 3], [4, 5], [6, 7]]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
(self: Dequeue, n: number): Effect, E>;
};
/**
* Takes between `min` and `max` messages from the queue.
*
* **Details**
*
* The operation waits when fewer than the required minimum messages are
* available. It returns at most `max` messages. 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 messages even when fewer
* than `min` are available. Once the queue is done, the effect fails with the
* queue's terminal error.
*
* **Example** (Taking a bounded batch of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8])
*
* // Take between 2 and 5 messages
* const batch1 = yield* Queue.takeBetween(queue, 2, 5)
*
* // Take between 1 and 10 messages (but only 3 remain)
* const batch2 = yield* Queue.takeBetween(queue, 1, 10)
*
* // No more messages available, will wait or return done
* // const batch3 = yield* Queue.takeBetween(queue, 1, 3)
* return [batch1, batch2]
* })
*
* await Effect.runPromise(program) // => [[1, 2, 3, 4, 5], [6, 7, 8]]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
export declare const takeBetween: {
/**
* Takes between `min` and `max` messages from the queue.
*
* **Details**
*
* The operation waits when fewer than the required minimum messages are
* available. It returns at most `max` messages. 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 messages even when fewer
* than `min` are available. Once the queue is done, the effect fails with the
* queue's terminal error.
*
* **Example** (Taking a bounded batch of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8])
*
* // Take between 2 and 5 messages
* const batch1 = yield* Queue.takeBetween(queue, 2, 5)
*
* // Take between 1 and 10 messages (but only 3 remain)
* const batch2 = yield* Queue.takeBetween(queue, 1, 10)
*
* // No more messages available, will wait or return done
* // const batch3 = yield* Queue.takeBetween(queue, 1, 3)
* return [batch1, batch2]
* })
*
* await Effect.runPromise(program) // => [[1, 2, 3, 4, 5], [6, 7, 8]]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
(min: number, max: number): (self: Dequeue) => Effect, E>;
/**
* Takes between `min` and `max` messages from the queue.
*
* **Details**
*
* The operation waits when fewer than the required minimum messages are
* available. It returns at most `max` messages. 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 messages even when fewer
* than `min` are available. Once the queue is done, the effect fails with the
* queue's terminal error.
*
* **Example** (Taking a bounded batch of values)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add several messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6, 7, 8])
*
* // Take between 2 and 5 messages
* const batch1 = yield* Queue.takeBetween(queue, 2, 5)
*
* // Take between 1 and 10 messages (but only 3 remain)
* const batch2 = yield* Queue.takeBetween(queue, 1, 10)
*
* // No more messages available, will wait or return done
* // const batch3 = yield* Queue.takeBetween(queue, 1, 3)
* return [batch1, batch2]
* })
*
* await Effect.runPromise(program) // => [[1, 2, 3, 4, 5], [6, 7, 8]]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
(self: Dequeue, min: number, max: number): Effect, E>;
};
/**
* Takes a single message from the queue, or wait for a message to be
* available.
*
* **Details**
*
* If the queue is done, it will fail with `Done`. If the
* queue fails, the Effect will fail with the error.
*
* **Example** (Taking one value)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* // Add some messages
* yield* Queue.offer(queue, "first")
* yield* Queue.offer(queue, "second")
*
* // Take messages one by one
* const msg1 = yield* Queue.take(queue)
* const msg2 = yield* Queue.take(queue)
*
* // End the queue
* yield* Queue.end(queue)
*
* // Taking from an ended queue fails with Done
* const result = yield* Effect.exit(Queue.take(queue))
* return [[msg1, msg2], result]
* })
*
* await Effect.runPromise(program) // => [["first", "second"], Exit.fail(Cause.Done())]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
export declare const take: (self: Dequeue) => Effect;
/**
* Attempts to take one item from the queue without waiting.
*
* **Details**
*
* Returns `Option.some` when an item is immediately available. Returns
* `Option.none` when no item is available, when the queue is done, or when the
* immediate take observes a queue failure.
*
* **Example** (Polling without blocking)
*
* ```ts import.meta.vitest
* import { Effect, Option, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Poll returns Option.none if empty
* const maybe1 = yield* Queue.poll(queue)
*
* // Add an item
* yield* Queue.offer(queue, 42)
*
* // Poll returns Option.some with the item
* const maybe2 = yield* Queue.poll(queue)
* return [maybe1, maybe2]
* })
*
* await Effect.runPromise(program) // => [Option.none(), Option.some(42)]
* ```
*
* @stability stable
* @category taking
* @since 2.0.0
*/
export declare const poll: (self: Dequeue) => Effect>;
/**
* Peeks at the next item without removing it.
*
* **Details**
*
* Blocks until an item is available. If the queue is done or fails, the error is propagated.
*
* **Example** (Peeking at the next value)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
* yield* Queue.offer(queue, 42)
*
* // Peek at the next item without removing it
* const item = yield* Queue.peek(queue)
* return item
* })
*
* await Effect.runPromise(program) // => 42
* ```
*
* @stability stable
* @category taking
* @since 4.0.0
*/
export declare const peek: (self: Dequeue) => Effect;
/**
* Attempts to take one message from the queue synchronously.
*
* **When to use**
*
* Use when polling queue internals must not suspend or register a waiting taker,
* and `undefined` is an acceptable result for an empty queue.
*
* **Details**
*
* Returns an `Exit` for an immediately available message or for the queue's
* terminal state. Returns `undefined` when no message is immediately available.
* This operation does not wait or register a taker.
*
* **Example** (Taking one value synchronously)
*
* ```ts import.meta.vitest
* import { Effect, Exit, Queue } from "effect"
*
* // Create a queue and use unsafe operations
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Add some messages
* Queue.offerUnsafe(queue, 1)
* Queue.offerUnsafe(queue, 2)
*
* // Take a message synchronously
* const result1 = Queue.takeUnsafe(queue)
*
* const result2 = Queue.takeUnsafe(queue)
*
* // No more messages - returns undefined
* const result3 = Queue.takeUnsafe(queue)
* return [result1, result2, result3]
* })
*
* await Effect.runPromise(program) // => [Exit.succeed(1), Exit.succeed(2), undefined]
* ```
*
* @stability stable
* @category taking
* @since 4.0.0
*/
export declare const takeUnsafe: (self: Dequeue) => Exit | undefined;
/**
* Manually releases current queue takers synchronously.
*
* **When to use**
*
* Use when synchronous offers should release waiting consumers immediately
* instead of waiting for the scheduled release task.
*
* **Details**
*
* This immediately runs the queue's taker-release pass instead of waiting for
* its scheduled task. It does not complete the queue or resume fibers waiting
* on `Queue.await`.
*
* **Example** (Releasing a waiting taker synchronously)
*
* ```ts import.meta.vitest
* import { Effect, Fiber, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.unbounded()
* const taker = yield* Queue.take(queue).pipe(Effect.forkChild)
* yield* Effect.yieldNow
*
* Queue.offerUnsafe(queue, 1)
* Queue.flushUnsafe(queue)
*
* return yield* Fiber.join(taker)
* })
*
* await Effect.runPromise(program) // => 1
* ```
*
* @stability stable
* @category offering
* @since 4.0.0
*/
export declare const flushUnsafe: (self: Enqueue) => void;
/**
* Manually releases current queue takers.
*
* **When to use**
*
* Use when synchronous offers should release waiting consumers through an
* `Effect` instead of waiting for the scheduled release task.
*
* **Details**
*
* This immediately runs the queue's taker-release pass instead of waiting for
* its scheduled task. It does not complete the queue or resume fibers waiting
* on `Queue.await`.
*
* **Example** (Releasing a waiting taker)
*
* ```ts import.meta.vitest
* import { Effect, Fiber, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.unbounded()
* const taker = yield* Queue.take(queue).pipe(Effect.forkChild)
* yield* Effect.yieldNow
*
* Queue.offerUnsafe(queue, 1)
* yield* Queue.flush(queue)
*
* return yield* Fiber.join(taker)
* })
*
* await Effect.runPromise(program) // => 1
* ```
*
* @stability stable
* @category offering
* @since 4.0.0
*/
export declare const flush: (self: Enqueue) => Effect;
declare const await_: (self: Dequeue) => Effect>;
export {
/**
* Waits until a queue reaches the `Done` state.
*
* **When to use**
*
* Use to suspend a fiber until no further values can be taken from the queue
* and its terminal outcome is known.
*
* **Details**
*
* The effect succeeds with `void` for normal `Done` completion. Other
* terminal causes are preserved, so failures and interruptions complete this
* effect with the same terminal outcome.
*
* **Gotchas**
*
* A queue can be closing before it is done. `await` resumes at `Done`, not at
* the first completion signal, so buffered messages may need to be drained
* first.
*
* @see {@link end} for signaling normal completion while preserving buffered messages for consumers
* @see {@link fail} for signaling an error while preserving buffered messages for consumers
* @see {@link interrupt} for graceful interruption after buffered messages are drained
* @see {@link shutdown} for immediately discarding buffered messages and resuming pending operations
*
* @stability stable
* @category completion
* @since 4.0.0
*/
await_ as await };
/**
* Returns the current number of buffered messages in the queue.
*
* **Details**
*
* After `end`, a queue remains `Closing` while buffered messages are drained,
* and its size continues to include those messages. A `Done` queue reports a
* size of `0`.
*
* **Example** (Checking queue size)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Check size of empty queue
* const size1 = yield* Queue.size(queue)
*
* // Add some messages
* yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
*
* // Check size after adding messages
* const size2 = yield* Queue.size(queue)
*
* // End the queue
* yield* Queue.end(queue)
*
* // Ending retains the buffered size while the queue is Closing
* const size3 = yield* Queue.size(queue)
* return [size1, size2, size3]
* })
*
* await Effect.runPromise(program) // => [0, 5, 5]
* ```
*
* @stability stable
* @category sizes
* @since 2.0.0
*/
export declare const size: (self: Dequeue) => Effect;
/**
* Checks whether the queue is full.
*
* **Example** (Checking if queues are full)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* const before = yield* Queue.isFull(queue)
*
* // Add some messages
* yield* Queue.offerAll(queue, [1, 2, 3])
*
* const after = yield* Queue.isFull(queue)
* return [before, after]
* })
*
* await Effect.runPromise(program) // => [false, true]
* ```
*
* @stability stable
* @category predicates
* @since 2.0.0
*/
export declare const isFull: (self: Dequeue) => Effect;
/**
* Returns the current number of buffered messages in the queue synchronously.
*
* **When to use**
*
* Use when you need an immediate `Queue` size snapshot for diagnostics or
* internals and do not need the read wrapped in `Effect`.
*
* **Details**
*
* After `endUnsafe`, a queue remains `Closing` while buffered messages are
* drained, and its size continues to include those messages. A `Done` queue
* reports a size of `0`. This unsafe operation reads the queue state directly
* without Effect wrapping.
*
* **Example** (Checking queue size synchronously)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Check size of empty queue
* const size1 = Queue.sizeUnsafe(queue)
*
* // Add some messages
* Queue.offerUnsafe(queue, 1)
* Queue.offerUnsafe(queue, 2)
* Queue.offerUnsafe(queue, 3)
*
* // Check size after adding messages
* const size2 = Queue.sizeUnsafe(queue)
*
* // End the queue
* Queue.endUnsafe(queue)
*
* // Ending retains the buffered size while the queue is Closing
* const size3 = Queue.sizeUnsafe(queue)
* return [size1, size2, size3]
* })
*
* await Effect.runPromise(program) // => [0, 3, 3]
* ```
*
* @stability stable
* @category sizes
* @since 4.0.0
*/
export declare const sizeUnsafe: (self: Dequeue) => number;
/**
* Checks whether the queue is full synchronously.
*
* **When to use**
*
* Use when an immediate `Queue` capacity snapshot is needed outside effectful
* code and racing queue changes are acceptable.
*
* **Example** (Checking fullness synchronously)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(3)
*
* const before = Queue.isFullUnsafe(queue)
*
* // Add some messages
* yield* Queue.offerAll(queue, [1, 2, 3])
*
* const after = Queue.isFullUnsafe(queue)
* return [before, after]
* })
*
* await Effect.runPromise(program) // => [false, true]
* ```
*
* @stability stable
* @category predicates
* @since 4.0.0
*/
export declare const isFullUnsafe: (self: Dequeue) => boolean;
/**
* Runs an `Effect` into a `Queue`, where success ends the queue and failure
* fails the queue.
*
* **Example** (Running effects into queues)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create an effect that succeeds
* const dataProcessing = Effect.gen(function*() {
* yield* Effect.yieldNow
* return "Processing completed successfully"
* })
*
* // Pipe the effect into the queue
* // If dataProcessing succeeds, queue ends successfully
* // If dataProcessing fails, queue fails with the error
* const effectIntoQueue = Queue.into(queue)(dataProcessing)
*
* const wasCompleted = yield* effectIntoQueue
* const exit = yield* Effect.exit(Queue.take(queue))
* return [wasCompleted, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.fail(Cause.Done())]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
export declare const into: {
/**
* Runs an `Effect` into a `Queue`, where success ends the queue and failure
* fails the queue.
*
* **Example** (Running effects into queues)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create an effect that succeeds
* const dataProcessing = Effect.gen(function*() {
* yield* Effect.yieldNow
* return "Processing completed successfully"
* })
*
* // Pipe the effect into the queue
* // If dataProcessing succeeds, queue ends successfully
* // If dataProcessing fails, queue fails with the error
* const effectIntoQueue = Queue.into(queue)(dataProcessing)
*
* const wasCompleted = yield* effectIntoQueue
* const exit = yield* Effect.exit(Queue.take(queue))
* return [wasCompleted, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.fail(Cause.Done())]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
(self: Enqueue): (effect: Effect) => Effect;
/**
* Runs an `Effect` into a `Queue`, where success ends the queue and failure
* fails the queue.
*
* **Example** (Running effects into queues)
*
* ```ts import.meta.vitest
* import { Cause, Effect, Exit, Queue } from "effect"
*
* const program = Effect.gen(function*() {
* const queue = yield* Queue.bounded(10)
*
* // Create an effect that succeeds
* const dataProcessing = Effect.gen(function*() {
* yield* Effect.yieldNow
* return "Processing completed successfully"
* })
*
* // Pipe the effect into the queue
* // If dataProcessing succeeds, queue ends successfully
* // If dataProcessing fails, queue fails with the error
* const effectIntoQueue = Queue.into(queue)(dataProcessing)
*
* const wasCompleted = yield* effectIntoQueue
* const exit = yield* Effect.exit(Queue.take(queue))
* return [wasCompleted, exit]
* })
*
* await Effect.runPromise(program) // => [true, Exit.fail(Cause.Done())]
* ```
*
* @stability stable
* @category completion
* @since 4.0.0
*/
(effect: Effect, self: Enqueue): Effect;
};
//# sourceMappingURL=Queue.d.ts.map