How to Declare Exchanges and Queues Using Kourier's API: Builder DSL Guide

To declare exchanges and queues using Kourier's API, use the declaredExchange { } and declaredQueue { } DSL builders to create immutable state objects, then pass them to AMQPChannel.exchangeDeclare() or AMQPChannel.queueDeclare() suspend functions, or use inline builder blocks for concise single-line declarations.

Kourier is a Kotlin multiplatform AMQP client that abstracts RabbitMQ protocol details through a type-safe builder-plus-extension pattern. The library separates declaration state from channel operations, allowing you to define exchanges and queues as immutable data classes in DeclaredExchange.kt and DeclaredQueue.kt before transmitting them to the broker via extension functions in Channel Extensions.kt.

The Three-Layer Declaration Architecture

1. Immutable State Objects

The foundation consists of data classes DeclaredExchange and DeclaredQueue that hold AMQP entity properties. These objects are immutable and serializable, representing the desired state without side effects until passed to a channel.

2. Builder DSL Functions

Mutable builders (DeclaredExchangeBuilder and DeclaredQueueBuilder) provide a Kotlin DSL for configuring properties. The entry points declaredExchange { } and declaredQueue { } in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/Extensions.kt instantiate these builders and return the immutable state objects.

3. Channel Extension Suspend Functions

The AMQPChannel interface provides suspend functions exchangeDeclare, queueDeclare, and their passive variants in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt. These accept either state objects or builder blocks directly, converting the declarations into AMQP protocol frames via DefaultAMQPChannel.kt.

Declaring Exchanges Using the DSL

Standard Exchange Declaration

Use the inline builder syntax on any AMQPChannel instance to declare a durable direct exchange:

// `channel` is an AMQPChannel obtained from a connection
val exchangeResponse = channel.exchangeDeclare {
    name = "my.direct"
    type = "direct"
    durable = true          // survive broker restarts
    autoDelete = false
    internal = false
    arguments = emptyMap()
}
println("Exchange declared: ${exchangeResponse}")

Behind the scenes, the lambda builds a DeclaredExchange via DeclaredExchangeBuilder and the channel extension forwards the fields to the RabbitMQ exchange.declare method.

Passive Exchange Verification

To verify an exchange exists without modifying it, use exchangeDeclarePassive:

val passiveResponse = channel.exchangeDeclarePassive {
    name = "my.direct"
}
println("Passive exchange check: $passiveResponse")

Declaring Queues with Custom Arguments

Declare a durable queue with specific broker arguments using the same pattern:

val queueResponse = channel.queueDeclare {
    name = "task_queue"
    durable = true          // survive broker restarts
    exclusive = false
    autoDelete = false
    arguments = mapOf("x-max-length" to 1000L) // limit queue length
}
println("Queue declared: $queueResponse")

Explicit Builder Pattern vs. Inline DSL

While inline blocks are concise, explicitly building state objects enables configuration reuse across multiple channels:

// Build the state objects first (useful when re-using the same config)
val myExchange = declaredExchange {
    name = "logs"
    type = "fanout"
    durable = true
}
val myQueue = declaredQueue {
    name = "logs_queue"
    durable = true
    exclusive = false
    autoDelete = false
}

// Declare via the channel
channel.exchangeDeclare(myExchange)
channel.queueDeclare(myQueue)

// Bind the queue to the exchange
channel.queueBind {
    queue = myQueue.name
    exchange = myExchange.name
    routingKey = ""          // fanout ignores routingKey
}

This approach stores immutable configurations in myExchange and myQueue, allowing you to declare the same topology on multiple channels or connections without rebuilding the configuration.

Automatic Re-declaration with RobustAMQPChannel

For production environments requiring automatic recovery, RobustAMQPChannel (implemented in amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPChannel.kt) caches all declarations and replays them after connection failures:

val robustChannel = RobustAMQPChannel(wrappedChannel)

// Declare once; the robust channel stores the definitions
robustChannel.exchangeDeclare {
    name = "my.direct"
    type = "direct"
    durable = true
}
robustChannel.queueDeclare {
    name = "my_queue"
    durable = true
}

// If the connection drops, `RobustAMQPChannel` will replay the
// stored declarations when it reconnects.

The robust wrapper ensures idempotent startup by maintaining an internal registry of DeclaredExchange and DeclaredQueue objects, automatically re-executing the protocol declarations when the underlying connection recovers.

Summary

• State Objects: Immutable DeclaredExchange and DeclaredQueue classes in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/ encapsulate AMQP entity configurations.
• Builder DSL: The declaredExchange { } and declaredQueue { } functions in Extensions.kt provide type-safe configuration blocks.
• Channel Operations: AMQPChannel.exchangeDeclare() and AMQPChannel.queueDeclare() in Channel Extensions.kt transmit declarations to the broker as AMQP frames.
• Resilience: RobustAMQPChannel caches declarations from RobustAMQPChannel.kt to automatically restore topology after network partitions.

Frequently Asked Questions

What is the difference between exchangeDeclare and exchangeDeclarePassive in Kourier?

exchangeDeclare creates or updates an exchange with the specified properties, while exchangeDeclarePassive only verifies that an exchange with the given name already exists without modifying it. The passive variant throws an exception if the exchange does not exist, making it useful for validating topology before publishing messages.

Can I reuse a DeclaredQueue configuration across multiple channels?

Yes. Because DeclaredQueue and DeclaredExchange are immutable data classes, you can instantiate them once using declaredQueue { } or declaredExchange { } and pass the resulting objects to multiple AMQPChannel instances. This ensures consistent topology declarations across connection pools or clustered consumers.

How does RobustAMQPChannel handle re-declaration after a connection loss?

RobustAMQPChannel maintains an internal cache of all previously declared exchanges and queues. When the underlying connection drops and reconnects, the channel automatically replays the cached DeclaredExchange and DeclaredQueue objects through the standard exchangeDeclare and queueDeclare methods, ensuring your application topology survives broker restarts or network interruptions without manual intervention.

Where are the builder classes defined in the Kourier repository?

The mutable builders are defined in DeclaredExchangeBuilder.kt and DeclaredQueueBuilder.kt within amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/. These builders implement the DSL pattern used by the declaredExchange and declaredQueue convenience functions in Extensions.kt.

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