# How to Implement Publisher Confirms in Kourier for Reliable Message Publishing

> Implement Kourier publisher confirms for reliable message publishing. Learn how to use channel.confirmSelect() and consume publishConfirmResponses for message acknowledgment.

- Repository: [Guimauve Digital/kourier](https://github.com/guimauvedigital/kourier)
- Tags: how-to-guide
- Published: 2026-03-02

---

**Call `channel.confirmSelect()` to enable RabbitMQ publisher confirm mode, then consume the cold `Flow` from `publishConfirmResponses` to receive `Ack` or `Nack` objects that indicate whether the broker successfully persisted each message.**

Kourier (guimauvedigital/kourier) is a Kotlin multiplatform AMQP client that implements the RabbitMQ Publisher-Confirm pattern for reliable message delivery. When you implement publisher confirms in Kourier, the broker asynchronously notifies your application whether each message reached the queue, enabling exactly-once semantics or high-throughput batch processing. This mechanism is essential for building fault-tolerant systems where message loss is unacceptable.

## Enabling Confirm Mode on the Channel

Before publishing messages with delivery guarantees, you must place the channel into confirm mode using the `confirmSelect()` method.

### The confirmSelect() Method

In [`dev/kourier/amqp/channel/DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/dev/kourier/amqp/channel/DefaultAMQPChannel.kt) (lines 46-57), the `confirmSelect()` implementation sends the `Confirm.Select` RPC command to the broker and atomically flips the `isConfirmMode` boolean to `true`:

```kotlin
suspend fun confirmSelect() {
    // Sends Confirm.Select frame to broker
    rpc(AMQPMethod.Channel.Confirm.Select(nowait = false))
    isConfirmMode = true
}

```

Once enabled, every subsequent `basicPublish` call returns a `Published` response containing a monotonically increasing `deliveryTag`. This tag is generated atomically using a `Mutex` to ensure thread safety across concurrent coroutines (lines 74-80).

### The publishConfirmResponses Flow

The `AMQPChannel` interface declares `publishConfirmResponses` as a cold `Flow<AMQPResponse.Channel.Basic.PublishConfirm>` (lines 36-41 in [`AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/AMQPChannel.kt)). The default implementation in [`DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/DefaultAMQPChannel.kt) (lines 49-51) filters the internal `MutableSharedFlow<AMQPResponse>` to emit only confirm frames:

```kotlin
override val publishConfirmResponses: Flow<AMQPResponse.Channel.Basic.PublishConfirm>
    get() = channelResponses.filterIsInstance()

```

Because this is a **cold flow**, you control when to start collecting confirms. The flow emits either `Ack` (broker persisted the message) or `Nack` (broker rejected or failed to process) objects, each carrying the `deliveryTag` and a `multiple` flag.

## How Publisher Confirms Work Internally

When `isConfirmMode` is active, the publishing workflow follows these steps:

1. **Tag Generation** – `basicPublish` increments an internal `deliveryTag` counter protected by a `Mutex` and returns it in the `Published` response.
2. **Broker Acknowledgment** – The RabbitMQ broker later sends a `Basic.Ack` or `Basic.Nack` frame containing the same delivery tag.
3. **Frame Decoding** – Kourier's connection decoder transforms these frames into `AMQPResponse.Channel.Basic.PublishConfirm.Ack` or `.Nack` instances and pushes them into the shared `channelResponses` flow.
4. **Client Consumption** – Your collector receives these confirms via `publishConfirmResponses`, allowing you to correlate them with the original publish operation using the `deliveryTag`.

The `multiple` flag indicates that the broker is acknowledging all messages up to and including the specified delivery tag in a single frame, which is useful for optimizing high-throughput scenarios.

## Implementation Strategies for Kourier Publisher Confirms

Kourier supports three primary patterns for handling confirms, each balancing latency against throughput.

### Strategy 1: Individual Publish-and-Wait (Synchronous)

For strict exactly-once guarantees where you must confirm persistence before proceeding, publish a single message and block until the corresponding confirm arrives. This pattern is demonstrated in [`PublisherConfirmsTest.kt`](https://github.com/guimauvedigital/kourier/blob/main/PublisherConfirmsTest.kt) (lines 20-58):

```kotlin
channel.confirmSelect()

for (i in 1..messageCount) {
    channel.basicPublish(
        "Message $i".toByteArray(),
        exchange = "",
        routingKey = queueName,
        properties = Properties()
    )

    // Block until the broker acknowledges this single message
    when (val confirm = channel.publishConfirmResponses.first()) {
        is AMQPResponse.Channel.Basic.PublishConfirm.Ack  -> println("✅ $i acked")
        is AMQPResponse.Channel.Basic.PublishConfirm.Nack -> println("❌ $i nacked")
    }
}

```

**When to use:** Low-volume, critical financial transactions or command messages where durability is more important than throughput.

### Strategy 2: Batch Publishing for High Throughput

To reduce network round-trips, publish a batch of messages and wait for the entire batch to be acknowledged. This approach is implemented in [`PublisherConfirmsTest.kt`](https://github.com/guimauvedigital/kourier/blob/main/PublisherConfirmsTest.kt) (lines 90-120):

```kotlin
channel.confirmSelect()
val batchSize = 10

while (published < totalMessages) {
    repeat(batchSize) {
        channel.basicPublish(
            "Batch ${published + it + 1}".toByteArray(),
            exchange = "",
            routingKey = queueName,
            properties = Properties()
        )
    }

    // Collect exactly `batchSize` confirm frames
    val confirms = channel.publishConfirmResponses
        .take(batchSize)
        .toList()

    val acks = confirms.count { it is AMQPResponse.Channel.Basic.PublishConfirm.Ack }
    val nacks = confirms.size - acks
    println("Batch $published: $acks acks, $nacks nacks")
    published += batchSize
}

```

**When to use:** High-volume logging, telemetry, or event streaming where you can tolerate small windows of unconfirmed messages in memory.

### Strategy 3: Asynchronous Confirm Handling

For fire-and-forget publishing with background verification, launch a coroutine that collects confirms independently of your publishing loop. This pattern (lines 124-173 in [`PublisherConfirmsTest.kt`](https://github.com/guimauvedigital/kourier/blob/main/PublisherConfirmsTest.kt)) allows maximum publisher throughput:

```kotlin
channel.confirmSelect()
val messageCount = 20
val confirmed = mutableListOf<ULong>()
val nacked = mutableListOf<ULong>()

// Coroutine that consumes confirms as they arrive
val confirmJob = launch {
    channel.publishConfirmResponses
        .take(messageCount)
        .collect { confirm ->
            when (confirm) {
                is AMQPResponse.Channel.Basic.PublishConfirm.Ack -> {
                    confirmed.add(confirm.deliveryTag)
                    println("✅ ack ${confirm.deliveryTag}")
                }
                is AMQPResponse.Channel.Basic.PublishConfirm.Nack -> {
                    nacked.add(confirm.deliveryTag)
                    println("❌ nack ${confirm.deliveryTag}")
                }
            }
        }
}

// Fire-and-forget publishing
repeat(messageCount) { i ->
    channel.basicPublish(
        "Async $i".toByteArray(),
        exchange = "",
        routingKey = queueName,
        properties = Properties()
    )
}

confirmJob.join()
println("Done – ${confirmed.size} acks, ${nacked.size} nacks")

```

**When to use:** Real-time systems where you cannot block the publisher but need to log or retry failed messages asynchronously.

## Handling Bulk Confirms with the Multiple Flag

RabbitMQ can acknowledge ranges of messages using the `multiple` flag to reduce network traffic. When `multiple` is `true`, the delivery tag represents the highest tag in a contiguous range of confirmed messages. The test suite (lines 184-206) demonstrates how to detect and handle these bulk confirms:

```kotlin
val confirms = channel.publishConfirmResponses.take(messageCount).toList()
val multi = confirms.filter { it.multiple }
if (multi.isNotEmpty()) {
    println("Received ${multi.size} bulk confirms (multiple = true)")
}
assertTrue(confirms.all { it is AMQPResponse.Channel.Basic.PublishConfirm.Ack })

```

When processing confirms with `multiple = true`, you should consider all delivery tags less than or equal to the specified tag as acknowledged.

## Summary

- **Enable confirms** by calling `confirmSelect()` on the channel, which sets `isConfirmMode = true` in [`DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/DefaultAMQPChannel.kt).
- **Consume confirms** via the cold `Flow` at `publishConfirmResponses`, which filters the internal `channelResponses` stream for `Ack` and `Nack` frames.
- **Correlate messages** using the `deliveryTag` returned by `basicPublish`, which is atomically incremented and protected by a `Mutex`.
- **Choose your strategy** based on reliability requirements: individual blocking confirms for safety, batch confirms for throughput, or asynchronous collection for maximum performance.
- **Handle bulk confirms** by checking the `multiple` flag, which indicates that all messages up to the delivery tag are acknowledged.

## Frequently Asked Questions

### How does Kourier generate delivery tags for publisher confirms?

According to the source code in [`DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/DefaultAMQPChannel.kt) (lines 74-80), Kourier maintains an internal `deliveryTag` counter as a `ULong`. Each call to `basicPublish` while `isConfirmMode` is true atomically increments this counter using a `Mutex` to prevent race conditions in concurrent coroutines. The incremented value is returned in the `Published` response, allowing you to match it against the `deliveryTag` in the subsequent `Ack` or `Nack`.

### What is the difference between Ack and Nack in Kourier's publishConfirmResponses?

`AMQPResponse.Channel.Basic.PublishConfirm.Ack` indicates that the broker successfully received and persisted the message to disk or a queue, while `AMQPResponse.Channel.Basic.PublishConfirm.Nack` signals that the broker was unable to process the message due to resource constraints, queue limits, or internal errors. Your application should handle `Nack` by logging the failure, retrying the publish, or routing the message to a dead-letter queue based on the `deliveryTag`.

### What happens if I publish without calling confirmSelect()?

If you call `basicPublish` without first enabling confirm mode via `confirmSelect()`, the channel operates in standard fire-and-forget mode. The `basicPublish` method will not return a `deliveryTag`, and `publishConfirmResponses` will remain empty because the broker does not send `Basic.Ack` or `Basic.Nack` frames for non-confirm channels. You lose the ability to verify message durability at the application level.

### How do I handle the multiple flag when receiving confirms?

When `PublishConfirm.multiple` is `true`, the broker is acknowledging all messages up to and including the specified `deliveryTag` in a single frame. In Kourier, you process this by checking the flag on each confirm object and, if true, removing all pending tags less than or equal to that delivery tag from your internal tracking set. This optimization reduces the number of frames the broker must send during high-throughput scenarios.