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

> Learn to implement the publish-subscribe pattern with exchanges in Kourier. Broadcast messages using a fan-out exchange and exclusive queues for efficient subscriber communication.

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

---

**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`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt), this suspend function wraps the AMQP `Exchange.Declare` method.

```kotlin
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.

```kotlin
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`](https://github.com/guimauvedigital/kourier/blob/main/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.

```kotlin
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`](https://github.com/guimauvedigital/kourier/blob/main/Extensions.kt). Target the exchange, not a specific queue.

```kotlin
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.

```kotlin
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`](https://github.com/guimauvedigital/kourier/blob/main/PublishSubscribeTest.kt) in the repository.

```kotlin
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/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`](https://github.com/guimauvedigital/kourier/blob/main/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.