Class RabbitMqQueue

RabbitMQ queue adapter implementation.

Uses AMQP 0-9-1 via amqplib. Implements the claim-based reserve pattern to prevent double-dispatch. Uses polling via basicGet (NOT push consume) and per-message TTL + DLX for delayed re-delivery.

Key topology per job name <n>:

he.queue.<n>.ready - Ready queue (polling via basicGet)
he.queue.<n>.delay - Delay queue (TTL + DLX → ready)
he.queue.<n>.dead - Dead queue (final resting place)

All queues are reached via the default exchange (routing key = queue name).

Since

0.1.0

Implements

QueueAdapter

Index

Constructors

constructor

new RabbitMqQueue(
runtime: IRuntimeServices,
options?: RabbitMqQueueOptions,
): RabbitMqQueue
Creates a new RabbitMQ queue adapter.
Parameters
- runtime: IRuntimeServices
  Runtime services for clock conversion (absolute → relative TTL)
- Optionaloptions: RabbitMqQueueOptions
  RabbitMQ connection and configuration options
Returns RabbitMqQueue
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:337

Properties

`Optional`isHealthy

isHealthy?: () => Promise<boolean>

M70c: present only when the connection exposes on?; its absence is unknown reachability, not false. A fault returns false immediately; otherwise a throwaway channel round trip measures reachability. QueueService bounds and caches this probe.

Since

0.1.0

Methods

ack

ack(name: string, id: string, claimToken?: string): Promise<void>
Acknowledges a job as successfully processed.
Parameters
- name: string
  Job name
- id: string
  Job ID (stable public identity)
- OptionalclaimToken: string
  Opaque claim token returned from reserve (prevents stale settlement)
Returns Promise<void>
Resolves when acknowledged

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.ack
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:781

advanceRecurring

advanceRecurring(id: string, nextRunAtMs: number): Promise<void>
Advances a recurring job's next run time.
Parameters
- id: string
  Recurring job ID
- nextRunAtMs: number
  Next run timestamp
Returns Promise<void>
Resolves when advanced

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.advanceRecurring
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:895

connect

connect(): Promise<void>
Connects the adapter to its backend.

Returns Promise<void>
Resolves when connected

Since
0.1.0
Implementation of QueueAdapter.connect
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:434

deadLetter

deadLetter(
    name: string,
    id: string,
    nowMs: number,
    claimToken?: string,
): Promise<void>
Moves a job to the dead letter queue.
Parameters
- name: string
  Job name
- id: string
  Job ID (stable public identity)
- nowMs: number
  Current timestamp in ms (for dead-letter timestamp)
- OptionalclaimToken: string
  Opaque claim token returned from reserve (prevents stale settlement)
Returns Promise<void>
Resolves when dead-lettered

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.deadLetter
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:842

disconnect

disconnect(): Promise<void>
Disconnects the adapter.

Returns Promise<void>
Resolves when disconnected

Since
0.1.0
Implementation of QueueAdapter.disconnect
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:647

enqueue

enqueue<T>(job: StoredJob<T>): Promise<void>
Enqueues a job.
Type Parameters
- T
Parameters
- job: StoredJob<T>
  The job to enqueue
Returns Promise<void>
Resolves when enqueued

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.enqueue
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:686

fetchRecurringDue

fetchRecurringDue(nowMs: number): Promise<readonly StoredRecurring[]>
Fetches recurring jobs that are due.
Parameters
- nowMs: number
  Current timestamp in ms
Returns Promise<readonly StoredRecurring[]>
The due recurring jobs

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.fetchRecurringDue
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:880

isReady

isReady(): boolean
Checks if the adapter is ready/connected.

Returns boolean
true if connected

Since
0.1.0
Implementation of QueueAdapter.isReady
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:682

jobNameProblem

jobNameProblem(name: string): string | null
Internal
Why this adapter cannot use a job name, or null when it can. The derived <prefix>.<name>.ready / .delay queue names (the longest suffixes) must each fit an AMQP short string. The text never quotes the name, which may be caller data. QueueService calls this at process() and addRecurring() so an unusable name fails at registration rather than on every poll.
Parameters
- name: string
  The job name
Returns string | null
The refusal text, or null
Implementation of QueueAdapter.jobNameProblem
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:391

requeue

requeue<T>(
    name: string,
    id: string,
    availableAtMs: number,
    attempts: number,
    claimToken?: string,
): Promise<void>
Requeues a job with a new available timestamp.
Type Parameters
- T
Parameters
- name: string
  Job name
- id: string
  Job ID (stable public identity)
- availableAtMs: number
  When the job becomes available again
- attempts: number
  Updated attempt count
- OptionalclaimToken: string
  Opaque claim token returned from reserve (prevents stale settlement)
Returns Promise<void>
Resolves when requeued

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.requeue
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:797

reserve

reserve<T>(
    name: string,
    limit: number,
    _nowMs: number,
): Promise<readonly StoredJob<T>[]>
Reserves up to limit due jobs. Redis scripts use server time; adapters without a server clock compare availableAtMs to nowMs.

CLAIMS jobs: moves them from ready set to processing set. A reserved job is not returned by subsequent reserve calls.
Type Parameters
- T
Parameters
- name: string
  Job name
- limit: number
  Maximum jobs to reserve
- _nowMs: number
Returns Promise<readonly StoredJob<T>[]>
The reserved jobs

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.reserve
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:712

storeRecurring

storeRecurring(rec: StoredRecurring): Promise<void>
Stores a recurring job.
Parameters
- rec: StoredRecurring
  The recurring job to store
Returns Promise<void>
Resolves when stored

Throws
If the adapter is not connected

Since
0.1.0
Implementation of QueueAdapter.storeRecurring
- Defined in packages/queue-plugin/src/adapters/rabbitmq-queue.ts:871

Class RabbitMqQueue

Since

Implements

Index

Constructors

Properties

Methods

Constructors

constructor

Parameters

Returns RabbitMqQueue

Properties

OptionalisHealthy

Since

Methods

ack

Parameters

Returns Promise<void>

Throws

Since

advanceRecurring

Parameters

Returns Promise<void>

Throws

Since

connect

Returns Promise<void>

Since

deadLetter

Parameters

Returns Promise<void>

Throws

Since

disconnect

Returns Promise<void>

Since

enqueue

Type Parameters

Parameters

Returns Promise<void>

Throws

Since

fetchRecurringDue

Parameters

Returns Promise<readonly StoredRecurring[]>

Throws

Since

isReady

Returns boolean

Since

jobNameProblem

Parameters

Returns string | null

requeue

Type Parameters

Parameters

Returns Promise<void>

Throws

Since

reserve

Type Parameters

Parameters

Returns Promise<readonly StoredJob<T>[]>

Throws

Since

storeRecurring

Parameters

Returns Promise<void>

Throws

Since

Settings

On This Page

`Optional`isHealthy