Benefits of Using Kourier's Robust Client Over the Standard AMQP Client

Kourier's robust client provides automatic reconnection, full topology restoration, and stale delivery-tag handling, eliminating the manual recovery logic required when using the standard AMQP client.

The guimauvedigital/kourier library provides Kotlin multiplatform AMQP connectivity with two distinct client implementations. While the standard client offers basic publish/subscribe capabilities, the robust client adds production-grade resilience features that handle network failures transparently. Understanding the benefits of using Kourier's robust client over the standard AMQP client enables developers to build self-healing message processors without boilerplate reconnection code.

Automatic Reconnection and Connection Resilience

The standard client (DefaultAMQPConnection) provides no built-in reconnection logic. When the underlying TCP socket drops, the connection becomes permanently unusable, forcing the application to detect the failure and manually recreate the entire client stack.

In contrast, the robust client implements a dedicated background reconnect loop. Inside RobustAMQPConnection.kt, the connect() method launches a coroutine named reconnectSubscription that continuously retries the connection factory until it succeeds (lines 44-80). This loop catches all exceptions, logs them via logger.error("Connection factory failed, will retry", e), and retries after a short delay, ensuring the client survives transient network glitches without application intervention.

Channel Restoration and Topology Persistence

When the standard client's channel closes—whether by network error or broker decision—it is removed from the internal registry (shouldRemoveOnBrokerClose() → true). The application must then manually reopen the channel and redeclare all exchanges, queues, bindings, and consumers.

The robust client eliminates this burden through automatic state caching and restoration. In RobustAMQPChannel.kt, the channel overrides shouldRemoveOnBrokerClose() to return false (lines 37-41), preventing removal from the registry upon broker-initiated closure.

State Caching Mechanism

The robust channel maintains internal maps that cache every topology declaration:

• declaredQos – stores QoS settings
• declaredExchanges – stores exchange declarations
• declaredQueues – stores queue declarations
• boundExchanges and boundQueues – store binding information
• consumedQueues – stores consumer registrations

These caches are populated during normal operation (lines 21-27 in RobustAMQPChannel.kt) and serve as the source of truth during recovery.

Automatic Restoration Workflow

When the broker closes the channel, RobustAMQPChannel.cancelAll() triggers the restore() method (lines 52-85). This method executes a deterministic recovery sequence:

1. Re-opens the channel with the broker
2. Re-applies cached QoS settings via basicQos
3. Redeclares all exchanges and queues
4. Re-establishes all bindings
5. Restores all consumers with their original callbacks

This ensures that after a network partition heals, the application resumes processing messages without manual intervention or message loss due to missing consumers.

Stale Delivery-Tag Protection

After a reconnection, the AMQP broker resets its delivery tag sequence. If the application attempts to acknowledge a message using a delivery tag from the previous session, the broker returns a PRECONDITION_FAILED error and closes the channel.

The standard client offers no protection against this scenario, requiring developers to track connection state manually and discard stale tags.

The robust client solves this automatically. In RobustAMQPChannel.kt, the channel tracks the highest delivery tag seen (maxSeenDeliveryTag) and computes an offset (deliveryTagOffsetBeforeRestore) during the restoration process (lines 60-78). When basicAck, basicNack, or basicReject are called, the channel checks isStaleDeliveryTag() and silently ignores any tags that belong to the pre-restore period, preventing broker errors and channel closures.

Transparent API Compatibility

Despite the added resilience features, the robust client maintains full API compatibility with the standard implementation. Both clients implement the same AMQPConnection and AMQPChannel interfaces, allowing seamless substitution without refactoring application logic.

To instantiate the robust client, developers simply use the factory method RobustAMQPConnection.create() (lines 30-39 in RobustAMQPConnection.kt) instead of the standard connection constructor. The returned connection object can be used identically to the standard client, with all reconnection and restoration logic handled transparently in the background.

Practical Implementation Examples

The following examples demonstrate how to leverage the robust client in real-world scenarios.

Creating a Robust Connection

import dev.kourier.amqp.AMQPConfig
import dev.kourier.amqp.robust.RobustAMQPConnection
import kotlinx.coroutines.runBlocking

val config = AMQPConfig(
    server = AMQPConfig.Server(
        host = "localhost",
        port = 5672,
        user = "guest",
        password = "guest",
        vhost = "/",
        connectionName = "my-app"
    ),
    connection = AMQPConfig.Connection.Plain,
    server = AMQPConfig.ServerSettings(timeout = 10.seconds)
)

runBlocking {
    // The robust client automatically reconnects on failure
    val connection = RobustAMQPConnection.create(this, config)

    // Use the connection exactly like the standard client
    val channel = connection.openChannel()
    // …
}

Key source: RobustAMQPConnection.create() implements the reconnect-ready factory (lines 30-39 in RobustAMQPConnection.kt).

Declaring Topology with Automatic Restoration

val channel = connection.openChannel()
channel.exchangeDeclare(
    name = "orders",
    type = "topic",
    durable = true,
    autoDelete = false,
    internal = false,
    arguments = emptyMap()
)
channel.queueDeclare(name = "order-queue", durable = true, exclusive = false, autoDelete = false, arguments = emptyMap())
channel.queueBind(queue = "order-queue", exchange = "orders", routingKey = "order.*", arguments = emptyMap())

Even if the network drops, the robust channel’s internal caches (declaredExchanges, declaredQueues, boundQueues) will automatically replay these declarations during restoration (lines 21-27 in RobustAMQPChannel.kt).

Setting Up a Consumer with Auto-Restore

val consumerTag = channel.basicConsume(
    queue = "order-queue",
    consumerTag = "",
    noAck = false,
    exclusive = false,
    arguments = emptyMap(),
    onDelivery = { delivery ->
        // Process message
        println("Got ${delivery.message.bodyString()}")
        channel.basicAck(delivery.message.deliveryTag, multiple = false)
    },
    onCanceled = { reason ->
        println("Consumer canceled: $reason")
    }
)

If the broker closes the channel, RobustAMQPChannel.cancelAll() triggers restore() which re-issues the basicConsume request, preserving the consumer tag and callback logic (lines 93-102 in RobustAMQPChannel.kt).

Safe Acknowledgment Handling

// After a reconnection, any ack for a delivery tag that belonged to the previous session
// is ignored automatically – no extra code needed.
channel.basicAck(deliveryTag = oldTag, multiple = false) // safe, will be ignored if stale

The robust channel checks isStaleDeliveryTag() before forwarding the ACK/NACK/REJECT to the broker, preventing PRECONDITION_FAILED errors (lines 60-78 in RobustAMQPChannel.kt).

Summary

• Automatic reconnection: A background coroutine in RobustAMQPConnection continuously retries failed connections without application intervention.
• Full topology restoration: The robust client caches all declarations and automatically replays them after reconnection, including exchanges, queues, bindings, and consumers.
• Protection against stale delivery tags: Automatic filtering of acknowledgment attempts for messages from pre-restore sessions prevents broker errors.
• Drop-in replacement: The robust client implements the same AMQPConnection and AMQPChannel interfaces as the standard client, requiring only a factory method change to RobustAMQPConnection.create().

Frequently Asked Questions

How does automatic reconnection work in Kourier's robust client?

The robust client runs a dedicated reconnectSubscription coroutine inside RobustAMQPConnection.connect() that continuously attempts to re-establish the TCP connection when failures occur (lines 44-80 in RobustAMQPConnection.kt). This loop catches all exceptions, logs the failure, and retries after a delay, ensuring the connection self-heals without throwing exceptions to the application layer.

What topology elements are restored after a reconnection?

The robust client automatically restores all cached state via RobustAMQPChannel.restore(): QoS settings, exchange declarations, queue declarations, bindings, and consumer registrations (lines 52-85 in RobustAMQPChannel.kt). This is possible because the channel maintains internal maps like declaredExchanges, declaredQueues, and consumedQueues that record every declaration made during normal operation (lines 21-27).

How does the robust client prevent errors from stale delivery tags?

When a connection drops, the broker resets delivery tag sequences. The robust client tracks the maximum delivery tag seen (maxSeenDeliveryTag) and computes an offset (deliveryTagOffsetBeforeRestore) during the restoration process (lines 60-78 in RobustAMQPChannel.kt). When basicAck, basicNack, or basicReject are called, the channel checks isStaleDeliveryTag() and silently ignores any tags that belong to the pre-restore period, preventing broker errors and channel closures.

Is the API different between the standard and robust clients?

No. The robust client implements the exact same AMQPConnection and AMQPChannel interfaces as the standard client. You instantiate it using RobustAMQPConnection.create() (lines 30-39 in RobustAMQPConnection.kt) instead of the standard connection constructor, but all subsequent operations—opening channels, declaring topology, and consuming messages—use identical method signatures and semantics.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}