How to Implement the Publish-Subscribe Pattern with Exchanges in Kourier

Use a fan-out exchange to broadcast messages to multiple temporary queues, where each subscriber creates an exclusive, auto-delete queue bound to the exchange.

Kourier implements the standard RabbitMQ messaging model, meaning producers never send messages directly to queues. Instead, you implement the publish-subscribe pattern with exchanges in Kourier by publishing to an exchange that routes messages to all bound queues. This approach enables scalable, decoupled communication where multiple consumers receive identical message copies without competing for them.

Understanding the Publish-Subscribe Model in Kourier

Kourier follows the AMQP 0-9-1 protocol semantics where exchanges act as message routing agents. For true broadcast behavior, you use a fan-out exchange, which ignores routing keys entirely and forwards every received message to all queues bound to it.

Each subscriber requires a unique, temporary queue that exists only for the duration of the connection. By declaring a queue with an empty name (""), Kourier lets the broker generate a unique identifier (e.g., amq.gen-...). Setting exclusive = true and autoDelete = true ensures the queue disappears automatically when the subscriber disconnects, preventing resource leaks.

Step-by-Step Implementation

Declaring a Fan-Out Exchange

First, declare the exchange using AMQPChannel.exchangeDeclare. In amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt, this suspend function wraps the AMQP Exchange.Declare method.

channel.exchangeDeclare(
    "logs",
    BuiltinExchangeType.FANOUT,
    durable = false,
    autoDelete = false,
    internal = false,
    arguments = emptyMap()
)

This idempotent operation creates the logs exchange if it does not exist. The durable = false flag means the exchange will not survive broker restarts, which is appropriate for transient log broadcasting.

Creating Temporary Subscriber Queues

Each subscriber must create its own isolated queue. Use queueDeclare with an empty name to generate a server-named temporary queue.

val queue = channel.queueDeclare(
    name = "",                // Server-generated unique name
    durable = false,
    exclusive = true,         // Only this connection can access
    autoDelete = true,        // Deleted when consumer disconnects
    arguments = emptyMap()
).queueName

As implemented in Extensions.kt, this returns a DeclareOk object containing the generated queue name, which you must capture for the subsequent binding step.

Binding Queues to the Exchange

Bind the temporary queue to the fan-out exchange using queueBind. For fan-out exchanges, the routing key is ignored, so you pass an empty string.

channel.queueBind(
    queue = queue,
    exchange = "logs",
    routingKey = ""
)

This creates the relationship that allows the exchange to route messages into this specific queue. The binding persists until the queue is deleted.

Publishing Messages to the Exchange

Publish messages using basicPublish from Extensions.kt. Target the exchange, not a specific queue.

channel.basicPublish(
    body = "Application started".toByteArray(),
    exchange = "logs",
    routingKey = "",          // Ignored by FANOUT exchange
    properties = Properties()
)

Every active subscriber bound to the logs exchange receives this message simultaneously.

Consuming Messages from Temporary Queues

Consume messages using basicConsume, which returns a Flow<Delivery> as defined in the Kourier channel API.

val consumer = channel.basicConsume(queue, noAck = true)

consumer.collect { delivery ->
    val message = delivery.message.body.decodeToString()
    println("Received: $message")
}

Setting noAck = true enables automatic acknowledgment, suitable for transient logs where message loss during crashes is acceptable. The consumer runs until the channel closes, at which point the auto-delete queue cleans itself up.

Complete Working Example

The following example demonstrates a full implementation with one publisher and three concurrent subscribers, adapted from PublishSubscribeTest.kt in the repository.

import dev.kourier.amqp.*
import dev.kourier.amqp.channel.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

suspend fun emitLog(scope: CoroutineScope, message: String) {
    val config = amqpConfig { server { host = "localhost" } }
    val connection = createAMQPConnection(scope, config)
    val channel = connection.openChannel()

    channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT)
    
    channel.basicPublish(
        body = message.toByteArray(),
        exchange = "logs",
        routingKey = "",
        properties = Properties()
    )
    println(" [x] Sent '$message'")

    channel.close()
    connection.close()
}

suspend fun receiveLogs(
    scope: CoroutineScope,
    subscriberName: String
) {
    val config = amqpConfig { server { host = "localhost" } }
    val connection = createAMQPConnection(scope, config)
    val channel = connection.openChannel()

    channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT)
    
    val queue = channel.queueDeclare(
        name = "",
        durable = false,
        exclusive = true,
        autoDelete = true,
        arguments = emptyMap()
    ).queueName

    channel.queueBind(queue = queue, exchange = "logs", routingKey = "")

    println(" [$subscriberName] Waiting for logs. To exit press CTRL+C")

    val consumer = channel.basicConsume(queue, noAck = true)
    consumer.collect { delivery ->
        val msg = delivery.message.body.decodeToString()
        println(" [$subscriberName] $msg")
    }
}

fun main() = runBlocking {
    // Launch three concurrent subscribers
    launch { receiveLogs(this, "Subscriber-1") }
    launch { receiveLogs(this, "Subscriber-2") }
    launch { receiveLogs(this, "Subscriber-3") }

    delay(500) // Allow time for consumers to bind
    
    emitLog(this, "info: Application started")
    emitLog(this, "warning: High memory usage")
    emitLog(this, "error: Database connection failed")

    delay(30000) // Keep alive to observe output
}

Source: [PublishSubscribeTest.kt](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/PublishSubscribeTest.kt)

Summary

• Use a fan-out exchange to broadcast messages to all bound queues without evaluating routing keys.
• Create temporary queues with name = "", exclusive = true, and autoDelete = true so each subscriber gets a unique, self-cleaning endpoint.
• Bind queues to the exchange using queueBind with an empty routing key.
• Publish to the exchange, not to specific queues, using basicPublish with the exchange name.
• Consume using basicConsume, which returns a Kotlin Flow of deliveries for coroutine-based processing.

Frequently Asked Questions

What exchange type should I use for publish-subscribe in Kourier?

Use BuiltinExchangeType.FANOUT. According to the Kourier source code in Extensions.kt, fan-out exchanges ignore routing keys and route messages to all bound queues, making them the correct choice for broadcast scenarios where every subscriber must receive every message.

Why use temporary queues instead of named queues?

Temporary queues—declared with empty names ("") and marked exclusive = true and autoDelete = true—ensure that each subscriber receives its own copy of messages without competing with other consumers. When a subscriber disconnects, the broker automatically deletes the queue, preventing stale resources and name collisions in the guimauvedigital/kourier implementation.

How does Kourier handle message routing with fan-out exchanges?

Kourier delegates routing to the underlying AMQP broker. When you call basicPublish targeting a fan-out exchange, the broker ignores the routing key parameter and instead iterates over all bindings associated with that exchange, delivering a copy of the message to each bound queue. This behavior is defined by the AMQP protocol and exposed through the AMQPChannel extensions in Kourier.

Can I use routing keys with fan-out exchanges in Kourier?

No. While the basicPublish method requires a routing key parameter, fan-out exchanges in Kourier (and standard AMQP) discard this value. If you need selective routing based on routing keys, use a BuiltinExchangeType.DIRECT or BuiltinExchangeType.TOPIC exchange instead, as implemented in the channel extension methods.

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