All Symbols - queue documentation

default

Examples

Basic send with Task semantics

import { Queue, createInMemory } from "@anabranch/queue";

const connector = createInMemory();
const queue = await Queue.connect(connector).run();

const messageId = await queue
  .send("notifications", { userId: 123, type: "welcome" })
  .run();

Stream messages with concurrent processing and error collection

const connector = createInMemory();
const queue = await Queue.connect(connector).run();

const { successes, errors } = await queue
  .stream("notifications")
  .withConcurrency(5)
  .map(async (msg) => await sendEmail(msg.data))
  .tapErr((err) => logError(err))
  .partition();

Delayed messages with visibility timeout

import { Queue, createInMemory } from "@anabranch/queue";

const connector = createInMemory({ visibilityTimeout: 60_000 });
const queue = await Queue.connect(connector).run();

await queue.send("notifications", reminder, { delayMs: 30_000 }).run();

Dead letter queue with max attempts

import { Queue, createInMemory } from "@anabranch/queue";

const connector = createInMemory({
  queues: {
    orders: {
      maxAttempts: 3,
      deadLetterQueue: "orders-dlq",
    },
  },
});
const queue = await Queue.connect(connector).run();

await queue.nack("orders", msg.id, { deadLetter: true }).run();

Classes

Queue(adapter: QueueAdapter)

Queue wrapper with Task/Stream semantics for error-tolerant message processing.

ack(
queue: string,
...ids: string[]
): Task<void, QueueAckFailed>
Acknowledge one or more messages as successfully processed.
close(): Task<void, QueueCloseFailed>
Release the connection back to its source (e.g., pool).
connect(connector: QueueConnector): Task<Queue, QueueConnectionFailed>
Connect to a queue via a connector.
continuousStream<T>(
queue: string,
options?: { count?: number; signal?: AbortSignal; prefetch?: number; backoff?: { initialDelay?: number; multiplier?: number; maxDelay?: number; }; }
): Source<QueueMessage<T>, QueueReceiveFailed>
Continuous stream that polls for messages until stopped.
nack(
queue: string,
id: string,
options?: { requeue?: boolean; delay?: number; deadLetter?: boolean; }
): Task<void, QueueAckFailed | QueueMaxAttemptsExceeded>
Negative acknowledgment - indicates processing failure.
send<T>(
queue: string,
data: T,
options?: SendOptions
): Task<string, QueueSendFailed>
Send a message to a queue.
sendBatch<T>(
queue: string,
data: T[],
options?: SendOptions & { parallel?: boolean; }
): Task<string[], QueueSendFailed>
Send multiple messages to a queue in batch.
stream<T>(
queue: string,
options?: { count?: number; }
): Source<QueueMessage<T>, QueueReceiveFailed>
Stream messages from a queue for memory-efficient concurrent processing.

QueueAborted(

message?: string,

queue?: string

)

Operation was aborted via an AbortSignal.

QueueAckFailed(

message: string,

queue: string,

messageId: string,

originalError?: unknown

)

Acknowledgment operation failed.

QueueBufferFull(queue: string)

Operation failed because the queue's internal buffer is full.

QueueCloseFailed(

message: string,

originalError?: unknown

)

Connection close operation failed.

QueueConfigError(

message: string,

queue?: string

)

Configuration or capability error.

QueueConnectionFailed(

message: string,

originalError?: unknown

)

Connection establishment failed.

QueueError(

Base error for all queue-related failures.

QueueMaxAttemptsExceeded(

Message exceeded maximum delivery attempts.

QueueNackFailed(

message: string,

queue: string,

messageId: string,

originalError?: unknown

)

Negative acknowledgment operation failed.

QueueReceiveFailed(

message: string,

queue: string,

originalError?: unknown

)

Message receive operation failed.

QueueSendFailed(

message: string,

queue: string,

originalError?: unknown

)

Message send operation failed.

Source<T, E>(

resultSource: () => AsyncGenerator<Result<T, E>>,

concurrency?: number,

bufferSize?: number

)

The entry point for creating a Stream. Wraps an async generator so that yielded values become success results and any thrown error becomes a single error result.

from<T, E>(source: AsyncIterable<T>): Source<T, E>
Creates a Source from an existing AsyncIterable or async generator function. Each value emitted becomes a success result; any thrown error becomes an error result.
fromArray<T>(items: T[]): Source<T, never>
Creates a Source from an array of values.
fromRange(
start: number,
end: number
): Source<number, never>
Creates a Source that emits a range of numbers from start (inclusive) to end (exclusive). Each number is emitted as a success result.
fromResults<T, E>(source: () => AsyncGenerator<Result<T, E>>): Source<T, E>
Creates a Source from an async generator that yields Result values directly. This is useful when you want to yield both successes and errors from the source without terminating it on the first error.
fromSchedule(
cron: string,
options?: { signal?: AbortSignal; }
): Source<Tick, never>
Creates a Source that yields a Tick on each cron schedule match. Supports 5-field (minute) and 6-field (second) cron expressions, day/month names (MON-FRI, JAN), and aliases (@daily, @hourly).
fromTask<T, E>(task: Task<T, E>): Source<T, E>
Creates a Source from a Task.
withBufferSize(n: number): Source<T, E>
Sets the maximum number of buffered results before backpressure is applied to the stream. If the buffer is full, the stream will pause until there is space in the buffer for new results.
withConcurrency(n: number): Source<T, E>
Sets the maximum number of concurrent operations for the stream.

Task<T, E>(task: (signal?: AbortSignal) => Promise<T>)

A single async task with error-aware utilities like retries and timeouts.

acquireRelease<R, T, E>(unnamed 0: { acquire: (signal?: AbortSignal) => Promise<R>; release: (resource: R) => Promise<void>; use: (resource: R) => Task<T, E>; }): Task<T, E>
Acquires a resource, runs a task that uses it, and releases it regardless of success or failure. Useful for resource lifecycle management when the use computation is a composed Task chain.
all<T, E>(tasks: Task<T, E>[]): Task<T[], E>
Runs multiple tasks and collects results. Rejects on the first failure.
allSettled<T, E>(tasks: Task<T, E>[]): Task<Result<T, E>[], never>
Runs multiple tasks and collects all results without throwing.
chain(arr: []): Task<void, never>
Runs tasks sequentially, passing the successful value from one to the next. If any task fails, the chain is short-circuited and the error is returned.
delayWithSignal(
ms: number,
signal?: AbortSignal
): Promise<void>
flatMap<U, E2 = E>(fn: (value: T) => Task<U, E2>): Task<U, E | E2>
Chains another task based on the successful value.
flatMapErr<F>(fn: (error: E) => Task<T, F>): Task<T, F>
Chains another task based on the error value.
map<U, E2 = E>(fn: (value: T) => Promisable): Task<U, E | E2>
Maps the successful value. Errors are passed through unchanged.
mapErr<F>(fn: (error: E) => Promisable<F>): Task<T, F>
Maps the error value. Successful values are passed through unchanged.
mergeExternalSignals(
outer: AbortSignal,
inner: AbortSignal
): { signal: AbortSignal; cleanup: () => void; }
mergeSignals(
outer: AbortSignal,
inner?: AbortSignal
): { signal: AbortSignal; cleanup: () => void; }
of<R, E>(task: (signal?: AbortSignal) => Promisable<R>): Task<R, E>
Creates a Task from a sync or async function. The function receives an optional AbortSignal that is active when the task is run with a signal via withSignal or run.
race<T, E>(tasks: Task<T, E>[]): Task<Result<T, E[]>, never>
Runs tasks concurrently and resolves with the first settled result.
recover(fn: (error: E) => Promisable): Task<T | U, never>
Recovers from errors by mapping them to a successful value.
recoverWhen<E2 extends E, U>(
guard: (error: E) => error is E2,
fn: (error: E2) => Promisable
): Task<T | U, Exclude<E, E2>>
Recovers from specific error types by mapping them to a successful value.
result(): Promise<Result<T, E>>
Executes the task and returns a tagged result instead of throwing.
retry(options: { attempts: number; delay?: number | ((
attempt: number,
error: E
) => number); when?: (error: E) => boolean; }): Task<T, E>
Retries the task when the predicate returns true.
run(): Promise<T>
Executes the task and resolves the value or throws the error.
runTask: (signal?: AbortSignal) => Promise<T>
runWithSignal(signal?: AbortSignal): Promise<T>
tap(fn: (value: T) => Promisable<void>): Task<T, E>
Runs a side effect on the successful value.
tapErr(fn: (error: E) => Promisable<void>): Task<T, E>
Runs a side effect on the error value.
timeout(
ms: number,
error?: E
): Task<T, E>
Fails if the task does not complete within the specified time.
withSignal(signal: AbortSignal): Task<T, E>
Wraps the task with an external abort signal.

Functions

createInMemory(options?: InMemoryOptions): InMemoryConnector

Creates an in-memory queue connector using a simple message store.

Interfaces

InMemoryConnector

In-memory queue connector.

NackOptions

Options for negative acknowledgment.

deadLetter: boolean
Explicit dead letter queue target
delay: number
Delay before the message is requeued
requeue: boolean
Requeue the message instead of dead-letter routing

QueueAdapter

Queue adapter interface for low-level queue operations.

ack(
queue: string,
...ids: string[]
): Promise<void>
Acknowledge that one or more messages have been successfully processed. The broker should remove these messages from the queue.
close(): Promise<void>
Release any resources held by this adapter instance. Does not necessarily terminate the underlying connection (which is managed by the Connector).
nack(
queue: string,
id: string,
options?: NackOptions
): Promise<void>
Indicate that a message processing failed.
receive<T>(
queue: string,
count?: number
): Promise<QueueMessage<T>[]>
Retrieve one or more messages from the specified queue.
send<T>(
queue: string,
data: T,
options?: SendOptions
): Promise<string>
Send a single message to the specified queue.
sendBatch<T>(
queue: string,
data: T[],
options?: SendOptions
): Promise<string[]>
Send a batch of messages to the specified queue. Implementation should optimize this operation if possible (e.g. using pipelines or batch APIs).

QueueConnector

Connector that produces connected QueueAdapter instances.

connect(signal?: AbortSignal): Promise<QueueAdapter>
Acquire a connected adapter.
end(): Promise<void>
Close all connections and clean up resources. After calling end(), the connector cannot be used to create new adapters.

QueueMessage

Queue adapter interface for queue-agnostic operations.

attempt: number
Number of times this message has been delivered
data: T
Message payload.
id: string
Unique message identifier
metadata: { [key: string]: unknown; headers?: Record<string, string>; }
Optional metadata from the broker
timestamp: number
Timestamp when the message was first enqueued

QueueOptions

Queue configuration options.

deadLetterOptions: { maxAttempts?: number; delay?: number; }
Dead letter queue specific options
deadLetterQueue: string
Default dead letter queue for failed messages
maxAttempts: number
Maximum delivery attempts before routing to dead letter queue
visibilityTimeout: number
Message visibility timeout (milliseconds) - time between delivery and ACK/NACK

SendOptions

Options for sending a message with delay or scheduling.

deadLetterQueue: string
Override the default dead letter queue for this message
delayMs: number
Delay in milliseconds before the message becomes available
headers: Record<string, string>
Arbitrary key-value pairs to attach to the message
priority: number
Message priority (higher = more important, if supported)
scheduledAt: Date
Explicit scheduled delivery time

Stream

A TypeScript library that provides a powerful and flexible way to handle errors in asynchronous streams. It allows you to collect and manage errors alongside successful values in a stream, enabling you to process data while gracefully handling any issues that may arise.

_getSource(): () => AsyncGenerator<Result<T, E>>
Internal method used to enable things like zip There's probably no reason to call this
chunks(size: number): Stream<T[], E>
Collects consecutive successful values into fixed-size arrays. Errors pass through without breaking the current chunk.
collect(): Promise<T[]>
Collects all successful values emitted by the stream into an array. If any errors were collected during the stream processing, they will be thrown as an AggregateError containing all collected errors.
errors(): AsyncIterable<E>
Returns an async iterable of all errors collected during the stream processing. If any successful values were emitted during the stream processing, they will be ignored in this iterable.
filter(fn: (
value: T,
arrivalIndex: number
) => value is U): Stream<U, E>
Similar to Array.prototype.filter, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.
filterErr<F extends E>(fn: (
error: E,
errorIndex: number
) => error is F): Stream<T, F>
Similar to Array.prototype.filter, but works on the stream of errors. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.
flatMap<U, E2 = E>(fn: (
value: T,
arrivalIndex: number
) => Promisable<AsyncIterable | Iterable>): Stream<U, E | E2>
Similar to Array.prototype.flatMap, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.
flatten(this: Stream<Iterable | AsyncIterable, E>): Stream<U, E>
Flattens a stream of iterables into a stream of individual values. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream. Handles concurrency via the flatMap method, so if the inner iterables are produced out of order due to concurrency, the flattened results will reflect that order.
fold(
fn: (
acc: U,
value: T,
arrivalIndex: number
) => Promisable,
initialValue: U
): Promise
Similar to Array.prototype.reduce, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.
foldErr<F>(
fn: (
acc: F,
error: E,
errorIndex: number
) => Promisable<F>,
initialValue: F
): Promise<F>
Similar to Array.prototype.reduce, but works on the stream of errors. If the provided function throws an error or returns a rejected promise, the new error will be collected and emitted as an error result in the stream.
map<U, E2 = E>(fn: (
value: T,
arrivalIndex: number
) => Promisable): Stream<U, E | E2>
Similar to Array.prototype.map, but works on the stream of results. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in the stream.
mapErr<F>(fn: (
error: E,
errorIndex: number
) => Promisable<F>): Stream<T, F>
Similar to Array.prototype.map, but works on the stream of errors. If the provided function throws an error or returns a rejected promise, the new error will be collected and emitted as an error result in the stream.
merge(other: Stream<T, E>): Stream<T, E>
Merges two streams by interleaving their results. Both streams must have compatible error types. The merged stream yields values from either stream as they become available.
partition(): Promise<{ successes: T[]; errors: E[]; }>
Collects all results into separate successes and errors arrays. Unlike collect(), this never throws.
recover(fn: (
error: E,
errorIndex: number
) => Promisable): Stream<T | U, never>
Recovers from all errors by applying the provided function to transform them into successful values. This allows you to handle all errors gracefully while still collecting successful values in the stream.
recoverWhen<E2 extends E, U>(
guard: (
error: E,
errorIndex: number
) => error is E2,
fn: (
error: E2,
recoveryIndex: number
) => Promisable
): Stream<T | U, Exclude<E, E2>>
Recovers from specific error types by applying the provided function to transform them into successful values. This allows you to handle specific errors gracefully while still collecting other errors in the stream.
scan(
fn: (
acc: U,
value: T,
arrivalIndex: number
) => Promisable,
initialValue: U
): Stream<U, E>
Like fold but emits the running accumulator after each successful value, allowing downstream operations to react to intermediate states.
scanErr<F>(
fn: (
acc: F,
error: E,
errorIndex: number
) => Promisable<F>,
initialValue: F
): Stream<T, F>
Like scan but works on the stream of errors. Emits the running accumulator after each error, allowing downstream operations to react to intermediate error states. If the provided function throws an error or returns a rejected promise, the new error will be collected and emitted as an error result in the stream.
splitBy<K extends string | number | symbol>(
keys: readonly K[],
cb: (
value: T,
arrivalIndex: number
) => Promisable<K>,
bufferSize: number,
options?: SplitOptions
): Record<
K,
Stream<T, E | MissingKeyError | PumpError | NoKeysError>
>
Splits the stream into separate streams based on computed keys. Each result is sent to the stream corresponding to its computed key. If a result's key does not match any of the provided keys, an error result is emitted for that value. The split streams share the same source, so they are not independent; if one split stream is slower than the others, it will cause backpressure on the source stream and all other splits until it catches up. Use with caution to avoid unintended bottlenecks.
splitN(
n: 0 | 1,
bufferSize: number,
options?: SplitOptions
): [Stream<T, E | PumpError>]
Splits the stream into n separate streams that each receive the same results.
successes(): AsyncIterable<T>
Returns an async iterable of all successful values emitted by the stream. If any errors were collected during the stream processing, they will be ignored in this iterable.
take(n: number): Stream<T, E>
Limits the stream to at most n successful values. Errors pass through without counting against the limit. After n successes are yielded, the stream stops immediately (any pending errors from earlier in the pipeline may still be yielded before stopping).
takeWhile(fn: (
value: T,
arrivalIndex: number
) => Promisable<boolean>): Stream<T, E>
Yields successful values while the predicate returns true. Once the predicate returns false, iteration stops. Errors pass through until the stream is stopped. If the predicate throws, an error result is emitted and iteration stops.
tap(fn: (
value: T,
arrivalIndex: number
) => Promisable<void>): Stream<T, E>
Runs a side-effect function on each successful value without transforming it. If the provided function throws an error or returns a rejected promise, the error will be collected and emitted as an error result in place of the original value.
tapErr(fn: (
error: E,
arrivalIndex: number
) => Promisable<void>): Stream<T, E>
Runs a side-effect function on each error without transforming it. If the provided function throws an error or returns a rejected promise, the new error replaces the original.
throwOn<E2 extends E>(guard: (error: E) => error is E2): Stream<T, Exclude<E, E2>>
Throws the specified error types if they are encountered in the stream. This allows you to handle specific errors immediately while continuing to process other errors.
toArray(): Promise<Result<T, E>[]>
Collects all results emitted by the stream into an array of Result objects, which can represent either successful values or errors. This method allows you to see the full outcome of the stream processing, including both successes and errors, without throwing an aggregate error.
zip<U, F>(other: Stream<U, F>): Stream<[T, U], E | F>
Combines this stream with another into tuples, yielding one result per pair of values. The shorter stream determines when zipping completes.

StreamAdapter

Extended adapter interface for broker-native streaming. Implement this if your broker has push-based message consumption (e.g., RabbitMQ channels, Kafka consumer groups, SQS long polling).

subscribe<T>(
queue: string,
options?: { signal?: AbortSignal; prefetch?: number; }
): AsyncIterable<QueueMessage<T>>