feat: auto-ack and idempotent ack in AMQPQueue.subscribe()

baelter · baelter · commit eb6e76b12e47 · 2026-03-05T17:26:34.000+01:00
Callback form: wraps the user callback to ack on return and nack+requeue on
throw. Pass `{ noAck: true }` to opt out; `requeueOnNack: false` to discard
on error instead of requeuing.

Iterator form: acks the previous message when the loop advances to the next.
The last message (after `break`) is left unacked. Call `msg.ack()` /
`msg.nack()` before advancing to override.

AMQPMessage.ack/nack/reject are now idempotent — calling them more than once
is safe and sends only one wire frame. This lets callback and library both
ack without coordinating.

AMQPConsumer.cancel() is also idempotent via a memoised promise, preventing
a double-basicCancel race when generator cleanup and sub.cancel() overlap.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -19,12 +19,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `session.closed` — `true` when the underlying connection is closed
   - `session.stop()` — cancels reconnection, clears all subscriptions, and closes the connection
 - `AMQPQueue` — reconnect-safe queue handle returned by `session.queue()`, with `publish()`, `subscribe()`, `get()`, `bind()`, `unbind()`, `purge()`, `delete()` ([#186](https://github.com/cloudamqp/amqp-client.js/pull/186))
-  - `subscribe(params?, callback?)` accepts `QueueSubscribeParams` — `ConsumeParams` plus an optional `prefetch` that sets channel QoS before each consume, including after reconnect
+  - `subscribe(callback)` / `subscribe(params, callback)` — auto-acks after the callback returns; nacks and requeues on throw; call `msg.ack()` / `msg.nack()` inside the callback to override; pass `{ noAck: true }` to skip acking entirely; `requeueOnNack` controls requeue behaviour on error ([#189](https://github.com/cloudamqp/amqp-client.js/pull/189))
+  - `subscribe()` / `subscribe(params)` — async-iterator form; auto-acks the previous message when the loop advances; the last message (after `break`) is left unacked; call `msg.ack()` / `msg.nack()` before advancing to override; pass `{ noAck: true }` to skip acking ([#189](https://github.com/cloudamqp/amqp-client.js/pull/189))
   - Subscriptions survive reconnection automatically; the async-iterator form continues yielding without any caller changes
 - `AMQPExchange` — reconnect-safe exchange handle returned by `session.exchange()`, with `publish()`, `bind()`, `unbind()`, `delete()` ([#186](https://github.com/cloudamqp/amqp-client.js/pull/186))
 - `AMQPSubscription` — stable consumer handle across reconnections: exposes `channel`, `consumerTag`, and `cancel()`
 - `AMQPGeneratorSubscription` — extends `AMQPSubscription` with `AsyncIterable<AMQPMessage>` support
-- `QueueSubscribeParams` — exported type combining `ConsumeParams` with `prefetch?`
+- `QueueSubscribeParams` — exported type combining `ConsumeParams` with `prefetch?` and `requeueOnNack?` (default `true`) ([#189](https://github.com/cloudamqp/amqp-client.js/pull/189))
 - `QueuePublishOptions` / `ExchangePublishOptions` — exported types for publish options; both extend `AMQPProperties` with a `confirm?` flag; `ExchangePublishOptions` adds `routingKey?`
 - `ondisconnect` hook on `AMQPBaseClient` (TCP and WebSocket) — fires when the connection drops
 
diff --git a/README.md b/README.md
@@ -40,17 +40,18 @@ const session = await AMQPSession.connect("amqp://localhost")
 const q = await session.queue("my-queue")
 await q.publish("Hello World", { deliveryMode: 2 })
 
-// Subscribe with a callback — consumer recovers automatically on reconnect
-const sub = await q.subscribe({ noAck: false }, async (msg) => {
+// Subscribe with a callback — consumer recovers automatically on reconnect.
+// Messages are acked after the callback returns. If it throws, the message is
+// nacked and requeued. Call msg.ack() / msg.nack() yourself to override.
+const sub = await q.subscribe(async (msg) => {
   console.log(msg.bodyString())
-  await msg.ack()
 })
 
-// Or subscribe with an async iterator
-const iterSub = await q.subscribe({ noAck: false })
+// Or subscribe with an async iterator — messages are acked when the loop advances.
+// Call msg.ack() / msg.nack() before the next iteration to override.
+const iterSub = await q.subscribe()
 for await (const msg of iterSub) {
   console.log(msg.bodyString())
-  await msg.ack()
 }
 
 // Exchanges work the same way
@@ -81,8 +82,8 @@ Subscriptions created via `queue.subscribe()` are automatically re-established a
 
 ```javascript
 const q = await session.queue("my-queue", { durable: true })
-const sub = await q.subscribe({ noAck: false, prefetch: 10 }, async (msg) => {
-  await msg.ack()
+const sub = await q.subscribe({ prefetch: 10 }, async (msg) => {
+  // process msg — acked on return, nacked and requeued on throw
 })
 
 // sub.consumerTag and sub.channel reflect the current consumer (updated on reconnect)
diff --git a/package-lock.json b/package-lock.json
diff --git a/src/amqp-consumer.ts b/src/amqp-consumer.ts
@@ -47,12 +47,18 @@ export class AMQPConsumer {
   /**
    * Cancel/abort/stop the consumer. No more messages will be deliviered to the consumer.
    * Note that any unacked messages are still unacked as they belong to the channel and not the consumer.
+   * Safe to call multiple times — concurrent calls share the same underlying wire operation.
    */
   cancel() {
     if (this.channel.closed) return Promise.resolve(this.channel)
-    return this.channel.basicCancel(this.tag)
+    if (!this._cancelPromise) {
+      this._cancelPromise = this.channel.basicCancel(this.tag)
+    }
+    return this._cancelPromise
   }
 
+  private _cancelPromise?: Promise<AMQPChannel>
+
   /**
    * @ignore
    * @param [err] - why the consumer was closed
diff --git a/src/amqp-message.ts b/src/amqp-message.ts
@@ -30,6 +30,12 @@ export class AMQPMessage {
   messageCount?: number
   replyCode?: number
   replyText?: string
+  #acked = false
+
+  /** True if the message has already been acked, nacked, or rejected. */
+  get isAcked(): boolean {
+    return this.#acked
+  }
 
   /**
    * @param channel - Channel this message was delivered on
@@ -56,16 +62,22 @@ export class AMQPMessage {
 
   /** Acknowledge the message */
   ack(multiple = false) {
+    if (this.#acked) return Promise.resolve()
+    this.#acked = true
     return this.channel.basicAck(this.deliveryTag, multiple)
   }
 
   /** Negative acknowledgment (same as reject) */
   nack(requeue = false, multiple = false) {
+    if (this.#acked) return Promise.resolve()
+    this.#acked = true
     return this.channel.basicNack(this.deliveryTag, requeue, multiple)
   }
 
-  /** Rejected the message */
+  /** Reject the message */
   reject(requeue = false) {
+    if (this.#acked) return Promise.resolve()
+    this.#acked = true
     return this.channel.basicReject(this.deliveryTag, requeue)
   }
 
diff --git a/src/amqp-queue.ts b/src/amqp-queue.ts
@@ -14,6 +14,11 @@ import { publishConfirmed, publishNoConfirm, type Body } from "./amqp-publisher.
 export type QueueSubscribeParams = ConsumeParams & {
   /** Per-consumer prefetch limit (sets QoS on the channel before consuming). */
   prefetch?: number
+  /**
+   * Whether to requeue messages that are nacked due to a callback error.
+   * Defaults to `true`.
+   */
+  requeueOnNack?: boolean
 }
 
 /** Options for {@link AMQPQueue#publish}. */
@@ -56,27 +61,48 @@ export class AMQPQueue {
     return this
   }
 
-  /**
-   * Subscribe to this queue with automatic consumer recovery on reconnection.
-   * @param params - consume and prefetch parameters
-   * @param callback - called for each delivered message
-   */
+  /** Subscribe with a callback. Messages are acked after the callback returns, nacked on error. */
+  subscribe(callback: (msg: AMQPMessage) => void | Promise<void>): Promise<AMQPSubscription>
+  /** Subscribe with a callback and custom params. */
   subscribe(
     params: QueueSubscribeParams,
     callback: (msg: AMQPMessage) => void | Promise<void>,
   ): Promise<AMQPSubscription>
   /**
-   * Subscribe to this queue with automatic consumer recovery on reconnection.
-   * Messages are delivered through an async-iterable subscription that continues
-   * across reconnections.
-   * @param [params] - consume and prefetch parameters
+   * Subscribe via an async iterator. Messages continue yielding across reconnections.
+   * @example
+   * ```ts
+   * for await (const msg of await q.subscribe()) {
+   *   console.log(msg.bodyString())
+   *   await msg.ack()
+   * }
+   * ```
    */
   subscribe(params?: QueueSubscribeParams): Promise<AMQPGeneratorSubscription>
   async subscribe(
-    params?: QueueSubscribeParams,
+    params?: QueueSubscribeParams | ((msg: AMQPMessage) => void | Promise<void>),
     callback?: (msg: AMQPMessage) => void | Promise<void>,
   ): Promise<AMQPSubscription | AMQPGeneratorSubscription> {
-    const { prefetch, ...consumeParams } = params ?? {}
+    if (typeof params === "function") [callback, params] = [params, undefined]
+    const { prefetch, requeueOnNack = true, ...consumeParams } = params ?? {}
+    // When auto-acking (callback or generator), force noAck: false so the server
+    // tracks delivery tags. basicConsume defaults noAck to true, so we must be explicit.
+    if (callback !== undefined && !consumeParams.noAck) {
+      consumeParams.noAck = false
+      const userCallback = callback
+      callback = async (msg: AMQPMessage) => {
+        try {
+          await userCallback(msg)
+          await msg.ack()
+        } catch {
+          await msg.nack(requeueOnNack)
+        }
+      }
+    }
+    // Generator form: also force noAck: false for auto-ack unless caller opted out.
+    if (callback === undefined && consumeParams.noAck !== true) {
+      consumeParams.noAck = false
+    }
     const def: ConsumerDefinition = {
       queueName: this.name,
       consumeParams,
diff --git a/src/amqp-subscription.ts b/src/amqp-subscription.ts
@@ -93,6 +93,8 @@ export class AMQPGeneratorSubscription extends AMQPSubscription implements Async
   }
 
   async *[Symbol.asyncIterator](): AsyncGenerator<AMQPMessage, void, undefined> {
+    const autoAck = !this.def.consumeParams.noAck
+    let prev: AMQPMessage | undefined
     while (!this.stopped) {
       const consumer = this.consumer
       if (!(consumer instanceof AMQPGeneratorConsumer)) {
@@ -101,11 +103,15 @@ export class AMQPGeneratorSubscription extends AMQPSubscription implements Async
       try {
         for await (const msg of consumer.messages) {
           if (this.stopped) return
+          if (autoAck) await prev?.ack()
+          prev = msg
           yield msg
         }
       } catch {
         // Consumer's channel was closed — wait for reconnect to provide a new consumer
       }
+      // Reset on disconnect; unacked messages are requeued by the server when the channel closes
+      prev = undefined
       if (!this.stopped) {
         await new Promise<void>((resolve) => {
           this.consumerReady = resolve
diff --git a/test/amqp-session.ts b/test/amqp-session.ts
