Default deliveryMode to 2 (persistent) in publish

baelter · baelter · commit 492c6c0a486b · 2026-05-07T15:32:17.000+02:00
AMQPQueue.publish and AMQPExchange.publish now default `deliveryMode: 2`
when not set, matching the Ruby client. The high-level API already defaults
`confirm: true` — persistent-by-default is the consistent stance for an API
that optimizes for "don't lose data." Explicit `deliveryMode: 1` is honored.

Low-level AMQPChannel.basicPublish stays as-is. RPC client/server publish
paths are not touched: requests/replies via amq.rabbitmq.reply-to are
inherently transient.
diff --git a/src/amqp-exchange.ts b/src/amqp-exchange.ts
@@ -43,6 +43,9 @@ export class AMQPExchange<
    * by the matching parser's `serialize` method. Without parsers, `body` must
    * be a string, Buffer, Uint8Array, or null.
    *
+   * Defaults: `confirm: true`, `deliveryMode: 2` (persistent). Pass
+   * `deliveryMode: 1` to send a transient message.
+   *
    * @param options - routing key, publish properties; set `confirm: false` to skip broker confirmation
    * @returns `this` for chaining
    */
@@ -61,6 +64,7 @@ export class AMQPExchange<
       properties,
       defaults,
     )
+    if (encoded.properties.deliveryMode === undefined) encoded.properties.deliveryMode = 2
     const ch = confirm ? await this.session.getConfirmChannel() : await this.session.getOpsChannel()
     await ch.basicPublish(this.name, routingKey, encoded.body, encoded.properties)
     return this
diff --git a/src/amqp-queue.ts b/src/amqp-queue.ts
@@ -63,6 +63,9 @@ export class AMQPQueue<
    * by the matching parser's `serialize` method. Without parsers, `body` must
    * be a string, Buffer, Uint8Array, or null.
    *
+   * Defaults: `confirm: true`, `deliveryMode: 2` (persistent). Pass
+   * `deliveryMode: 1` to send a transient message.
+   *
    * @param options - publish properties; set `confirm: false` to skip broker confirmation
    * @returns `this` for chaining
    */
@@ -81,6 +84,7 @@ export class AMQPQueue<
       properties,
       defaults,
     )
+    if (encoded.properties.deliveryMode === undefined) encoded.properties.deliveryMode = 2
     const ch = confirm ? await this.session.getConfirmChannel() : await this.session.getOpsChannel()
     await ch.basicPublish("", this.name, encoded.body, encoded.properties)
     return this
diff --git a/test/amqp-session.ts b/test/amqp-session.ts
@@ -251,6 +251,54 @@ test("AMQPQueue (session-backed).publish() and get() round-trip", () =>
     expect(msg?.bodyString()).toBe("round-trip")
   }))
 
+test("AMQPQueue.publish() defaults deliveryMode to 2 (persistent)", () =>
+  withSession(async (session) => {
+    const q = await session.queue("test-sq-persist-" + Math.random(), { durable: false, autoDelete: true })
+    await q.publish("default")
+
+    const msg = await q.get({ noAck: true })
+    expect(msg?.properties.deliveryMode).toBe(2)
+  }))
+
+test("AMQPQueue.publish() honors explicit deliveryMode: 1 (transient)", () =>
+  withSession(async (session) => {
+    const q = await session.queue("test-sq-transient-" + Math.random(), { durable: false, autoDelete: true })
+    await q.publish("transient", { deliveryMode: 1 })
+
+    const msg = await q.get({ noAck: true })
+    expect(msg?.properties.deliveryMode).toBe(1)
+  }))
+
+test("AMQPExchange.publish() defaults deliveryMode to 2 (persistent)", () =>
+  withSession(async (session) => {
+    const xName = "test-x-persist-" + Math.random()
+    const qName = "test-xq-persist-" + Math.random()
+    const x = await session.fanoutExchange(xName, { durable: false, autoDelete: true })
+    const q = await session.queue(qName, { durable: false, autoDelete: true })
+    await q.bind(x)
+
+    await x.publish("default")
+    await new Promise((resolve) => setTimeout(resolve, 50))
+
+    const msg = await q.get({ noAck: true })
+    expect(msg?.properties.deliveryMode).toBe(2)
+  }))
+
+test("AMQPExchange.publish() honors explicit deliveryMode: 1 (transient)", () =>
+  withSession(async (session) => {
+    const xName = "test-x-transient-" + Math.random()
+    const qName = "test-xq-transient-" + Math.random()
+    const x = await session.fanoutExchange(xName, { durable: false, autoDelete: true })
+    const q = await session.queue(qName, { durable: false, autoDelete: true })
+    await q.bind(x)
+
+    await x.publish("transient", { deliveryMode: 1 })
+    await new Promise((resolve) => setTimeout(resolve, 50))
+
+    const msg = await q.get({ noAck: true })
+    expect(msg?.properties.deliveryMode).toBe(1)
+  }))
+
 test("AMQPQueue (session-backed).subscribe() recovers after reconnect", () =>
   withSession(
     async (session) => {
