How to Implement Routing Using Direct Exchanges in Kourier

Direct exchanges in Kourier route messages to queues based on exact routing key matches, using BuiltinExchangeType.DIRECT and the queueBind method to filter messages by specific criteria like log severity.

Kourier is a Kotlin multiplatform AMQP 0-9-1 client that communicates with RabbitMQ-compatible brokers. When you implement routing using direct exchanges in Kourier, you create deterministic message flows where publishers target specific consumers through exact key matching rather than broadcasting to all queues.

What Are Direct Exchanges in Kourier?

A direct exchange routes messages to every queue whose binding key exactly matches the routing key supplied with basicPublish. This is implemented in amqp-core/src/commonMain/kotlin/dev/kourier/amqp/BuiltinExchangeType.kt, where the constant BuiltinExchangeType.DIRECT identifies the exchange type.

This routing pattern is ideal for:

• Filtering logs by severity (error, warning, info)
• Selecting specific consumer groups by logical tags
• Implementing request-reply patterns with unique reply queue keys

Unlike fanout exchanges that broadcast to all bound queues, direct exchanges provide selective delivery based on string equality.

Architecture of Direct Exchange Routing

The message flow follows five distinct steps as demonstrated in amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/RoutingTest.kt:

1. Declare the direct exchange using exchangeDeclare with BuiltinExchangeType.DIRECT
2. Create a temporary queue using queueDeclare with an empty name, exclusive = true, and autoDelete = true
3. Bind the queue using queueBind with a specific routingKey parameter for each desired message type
4. Publish messages using basicPublish with a routingKey that matches the binding keys
5. Consume messages using basicConsume to receive only the filtered deliveries

The exchange acts as a router, comparing the routing key from the publisher against the binding keys of all queues. Only exact matches result in delivery.

Implementing Direct Exchange Routing in Kourier

Declaring the Direct Exchange

All participants must declare the same exchange name and type. In RoutingTest.kt (lines 29-36), the exchange is declared with non-durable settings suitable for tutorials:

channel.exchangeDeclare(
    "direct_logs",
    BuiltinExchangeType.DIRECT,
    durable = false,
    autoDelete = false,
    internal = false,
    arguments = emptyMap()
)

The BuiltinExchangeType.DIRECT parameter ensures the broker treats this as a direct exchange, enforcing exact-match routing semantics.

Creating and Binding Queues

Consumers create temporary queues and bind them to specific routing keys. The following pattern from the test suite shows how to bind multiple severities to a single queue:

// Create temporary exclusive queue
val queue = channel.queueDeclare(
    name = "",
    durable = false,
    exclusive = true,
    autoDelete = true,
    arguments = emptyMap()
).queueName

// Bind to multiple routing keys
for (severity in listOf("info", "warning", "error")) {
    channel.queueBind(queue, "direct_logs", severity)
}

As shown in RoutingTest.kt (lines 60-63 and 92-99), multiple queueBind calls on the same queue enable a single consumer to listen to several routing keys simultaneously.

Publishing with Routing Keys

Publishers specify the target routing key in the basicPublish call. The broker delivers the message only to queues bound with matching keys:

suspend fun emitLogDirect(scope: CoroutineScope, severity: String, message: String) {
    val config = amqpConfig { server { host = "localhost" } }
    val connection = createAMQPConnection(scope, config)
    val channel = connection.openChannel()
    
    channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT)
    
    channel.basicPublish(
        message.toByteArray(),
        exchange = "direct_logs",
        routingKey = severity,  // This determines which queues receive the message
        properties = Properties()
    )
    
    channel.close()
    connection.close()
}

The routingKey parameter in basicPublish (lines 39-44 in RoutingTest.kt) must exactly match the binding key for delivery to occur.

Consuming Filtered Messages

Consumers use basicConsume to stream matching deliveries. The implementation filters automatically at the broker level:

suspend fun receiveLogsDirect(
    scope: CoroutineScope,
    subscriberName: String,
    severities: List<String>
) {
    val config = amqpConfig { server { host = "localhost" } }
    val connection = createAMQPConnection(scope, config)
    val channel = connection.openChannel()
    
    channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT)
    
    val queue = channel.queueDeclare(
        name = "",
        exclusive = true,
        autoDelete = true
    ).queueName
    
    // Create bindings for each severity
    for (s in severities) {
        channel.queueBind(queue, "direct_logs", s)
    }
    
    val consumer = channel.basicConsume(queue, noAck = true)
    for (delivery in consumer) {
        val key = delivery.message.routingKey
        val body = delivery.message.body.decodeToString()
        println(" [$subscriberName] Received '$key':'$body'")
    }
}

The consumer will only receive messages where the publish routing key matches one of the bound keys.

Complete End-to-End Example

The following runnable example from docs/tutorials/routing.md demonstrates selective log delivery:

fun main() = runBlocking {
    // Subscriber receiving only "error" logs
    launch { 
        receiveLogsDirect(
            this, 
            "Error-Logger", 
            listOf("error"), 
            mutableListOf()
        ) 
    }

    // Subscriber receiving all severities
    launch { 
        receiveLogsDirect(
            this, 
            "All-Logger", 
            listOf("info", "warning", "error"), 
            mutableListOf()
        ) 
    }

    delay(1000) // Allow time for queue binding

    // Publish various severity levels
    launch {
        emitLogDirect(this, "info", "Application started")
        delay(500)
        emitLogDirect(this, "warning", "Memory usage high")
        delay(500)
        emitLogDirect(this, "error", "Database connection failed")
    }

    delay(30_000) // Keep alive to observe output
}

When executed, the Error-Logger receives only the database connection failure message, while the All-Logger receives all three messages. This confirms the exact-match routing behavior defined in the AMQP specification.

Testing and Validation

The RoutingTest.kt file contains comprehensive verification of direct exchange behavior:

• Exact key matching: Confirms that error.database does not match bindings for error or error.disk (see testExactRoutingKeyMatch)
• Multiple bindings: Validates that a single queue can receive messages from multiple routing keys
• Non-matching keys: Ensures messages with unmatched routing keys are dropped by the exchange

According to the source code in amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/RoutingTest.kt, these tests verify that the Kourier implementation correctly adheres to AMQP 0-9-1 direct exchange semantics.

Summary

• Direct exchanges use BuiltinExchangeType.DIRECT to enable exact-match routing between publishers and consumers.
• Routing keys in basicPublish must exactly match binding keys in queueBind for message delivery to occur.
• Temporary queues with exclusive = true and autoDelete = true are standard for direct exchange consumers that need dynamic routing.
• Multiple bindings on a single queue allow one consumer to receive messages from several routing keys simultaneously.
• The implementation is validated in RoutingTest.kt and documented in docs/tutorials/routing.md.

Frequently Asked Questions

What is the difference between a direct exchange and a fanout exchange in Kourier?

A direct exchange routes messages based on exact routing key matches, delivering only to queues bound with identical keys. A fanout exchange broadcasts every message to all bound queues regardless of routing keys. Use direct exchanges when you need selective filtering; use fanout when all consumers should receive every message.

Can a single queue bind to multiple routing keys in the same direct exchange?

Yes. As demonstrated in RoutingTest.kt (lines 60-63), you can call queueBind multiple times on the same queue with different routing keys. The queue will then receive any message whose routing key matches any of the bound keys, effectively implementing an "OR" logic for message subscription.

What happens if I publish a message with a routing key that has no matching bindings?

The broker drops the message. In RoutingTest.kt, the testExactRoutingKeyMatch function verifies that keys like error.database do not match bindings for error or error.disk. If no queue is bound to the exact routing key provided in basicPublish, the message is silently discarded unless the exchange has an alternate exchange configured.

When should I use a topic exchange instead of a direct exchange in Kourier?

Use a topic exchange when you need pattern matching with wildcards (* for one word, # for zero or more words) in routing keys. Use a direct exchange when you need exact string equality for routing decisions. Direct exchanges are simpler and slightly more performant for exact-match scenarios like severity levels or specific service identifiers, while topic exchanges provide flexibility for hierarchical routing key structures.

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