class RabbitMqQueue
implements QueueAdapter
Since 0.1.0

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).

Constructors

RabbitMqQueue(
runtime: IRuntimeServices,
)

Creates a new RabbitMQ queue adapter.

Parameters
runtime: IRuntimeServices

Runtime services for clock conversion (absolute → relative TTL)

optional
options: RabbitMqQueueOptions

RabbitMQ connection and configuration options

Properties

Since 0.1.0
optional
isHealthy: () => Promise<boolean>

M70c: present only when the connection exposes on?; its absence is unknown reachability, not false.

Methods

Since 0.1.0
ack(
name: string,
id: string,
_claimToken?: string
): Promise<void>

Acknowledges a job as successfully processed.

Since 0.1.0
advanceRecurring(
id: string,
nextRunAtMs: number
): Promise<void>

Advances a recurring job's next run time.

Since 0.1.0
connect(): Promise<void>

Connects the adapter to its backend.

Since 0.1.0
deadLetter(
name: string,
id: string,
nowMs: number,
_claimToken?: string
): Promise<void>

Moves a job to the dead letter queue.

Since 0.1.0
disconnect(): Promise<void>

Disconnects the adapter.

Since 0.1.0
enqueue<T>(job: StoredJob<T>): Promise<void>

Enqueues a job.

Since 0.1.0
fetchRecurringDue(nowMs: number): Promise<readonly StoredRecurring[]>

Fetches recurring jobs that are due.

Since 0.1.0
isReady(): boolean

Checks if the adapter is ready/connected.

Since 0.1.0
requeue<T>(
name: string,
id: string,
availableAtMs: number,
attempts: number,
_claimToken?: string
): Promise<void>

Requeues a job with a new available timestamp.

Since 0.1.0
reserve<T>(
name: string,
limit: number,
_nowMs: number
): Promise<readonly StoredJob<T>[]>

Reserves up to limit jobs that are due (availableAtMs <= nowMs).

CLAIMS jobs: moves them from ready set to processing set. A reserved job is not returned by subsequent reserve calls.

Since 0.1.0
storeRecurring(rec: StoredRecurring): Promise<void>

Stores a recurring job.

Usage

import { RabbitMqQueue } from "queue-plugin/src/index.ts";