How to Implement the RPC Pattern (Request-Reply) with Kourier

Use Kourier's coroutine-based AMQP client to build a request-reply system by publishing messages with replyTo and correlationId, then matching responses on an exclusive callback queue.

The guimauvedigital/kourier library provides a Kotlin-native, suspendable API for RabbitMQ that simplifies implementing the RPC pattern (request-reply). Unlike blocking AMQP clients, Kourier leverages coroutines to handle concurrent RPC calls without thread starvation, making it ideal for high-throughput microservices.

Understanding the RPC Pattern Components

When you implement the RPC pattern with Kourier, you coordinate four distinct elements:

1. Client – Creates a temporary, exclusive callback queue and publishes a request containing a unique correlationId and the replyTo address.
2. Broker – Routes the request from the default exchange to the rpc_queue declared by the server.
3. Server – Consumes from rpc_queue, executes business logic, and publishes the result to the replyTo queue while preserving the correlationId.
4. Client (continuation) – Listens on the callback queue, filters messages by correlationId, and resumes the suspended coroutine with the result.

Implementing the RPC Server with Kourier

The server implementation resides in amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/RPCTest.kt. It uses the low-level API defined in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/DefaultAMQPChannel.kt to handle acknowledgments and publishing.

Declaring the RPC Queue

Before consuming, the server must declare a durable queue that acts as the entry point for RPC requests. In Kourier, this is done via the AMQPChannel interface:

channel.queueDeclare(
    "rpc_queue",
    durable = false,
    exclusive = false,
    autoDelete = false,
    arguments = emptyMap()
)

Setting Up Fair Dispatch with basicQos

To prevent a slow server instance from hoarding messages, enable fair dispatch by setting the prefetch count to 1. This ensures the server receives only one unacknowledged message at a time:

channel.basicQos(count = 1u, global = false)

Processing Requests and Sending Replies

The server consumes with noAck = false to enable explicit acknowledgment after processing. It extracts the correlationId and replyTo properties, computes the result (e.g., a Fibonacci function), and publishes the response:

val consumer = channel.basicConsume("rpc_queue", noAck = false)
for (delivery in consumer) {
    val props = delivery.message.properties
    val correlationId = props.correlationId
    val replyTo = props.replyTo
    val n = delivery.message.body.decodeToString().toIntOrNull() ?: 0

    val result = fib(n) // Business logic

    val replyProps = properties { this.correlationId = correlationId }

    if (replyTo != null) {
        channel.basicPublish(
            result.toString().toByteArray(),
            exchange = "",
            routingKey = replyTo,
            properties = replyProps
        )
    }
    channel.basicAck(delivery.message, multiple = false)
}

Implementing the RPC Client with Kourier

The client implementation, also found in RPCTest.kt, demonstrates how to use Kourier’s DSL from amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/Extensions.kt to manage connections and channels.

Creating the Exclusive Callback Queue

The client generates a unique, temporary queue that will receive the server’s response. By setting exclusive = true and autoDelete = true, RabbitMQ automatically cleans up the queue when the client disconnects:

val callback = channel.queueDeclare(
    name = "",
    durable = false,
    exclusive = true,
    autoDelete = true,
    arguments = emptyMap()
)
val callbackName = callback.queueName
val correlationId = UUID.randomUUID().toString()

Publishing Requests with correlationId

Before publishing, the client starts consuming from the callback queue to avoid a race condition where the reply arrives before the consumer is ready. It then publishes the message with the mandatory replyTo and correlationId properties:

// Start consuming before publishing
val consumer = channel.basicConsume(callbackName, noAck = true)

val requestProps = properties {
    this.correlationId = correlationId
    this.replyTo = callbackName
}
channel.basicPublish(
    n.toString().toByteArray(),
    exchange = "",
    routingKey = "rpc_queue",
    properties = requestProps
)

Consuming Replies and Matching correlationId

The client iterates over the consumer flow, checking each message’s correlationId against the one it sent. It uses Kotlin’s withTimeout to prevent indefinite suspension if the server fails to respond:

var result = 0
withTimeout(10_000) {
    for (delivery in consumer) {
        if (delivery.message.properties.correlationId == correlationId) {
            result = delivery.message.body.decodeToString().toInt()
            break
        }
    }
}
return result

Complete Working Example

To run the RPC system, initialize both the server and client within a coroutine scope. The following snippet from RPCTest.kt demonstrates the orchestration:

fun main() = runBlocking {
    // Start server in background
    launch { rpcServer(this) }
    delay(1000) // Allow server to initialize

    // Execute RPC call
    val answer = rpcClient(this, 30)
    println("fib(30) = $answer")
}

The test suite in RPCTest.kt validates single-client requests, concurrent multi-client scenarios, and explicit correlation ID matching to ensure production reliability.

Summary

• Kourier enables RPC pattern (request-reply) implementation through coroutine-native AMQP operations.
• The client creates an exclusive callback queue and sends requests with replyTo and correlationId properties.
• The server consumes from a shared rpc_queue, processes messages, and publishes responses to the replyTo address while preserving the correlationId.
• Fair dispatch via basicQos(count = 1u) prevents server overload, while coroutine timeouts prevent client hangs.
• All implementation details are available in amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/RPCTest.kt and the connection DSL in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/Extensions.kt.

Frequently Asked Questions

How does Kourier handle RPC timeouts?

Kourier uses Kotlin’s withTimeout coroutine builder to prevent indefinite suspension. In the client implementation, the response consumer is wrapped in a withTimeout(10_000) block, causing the RPC call to throw a TimeoutCancellationException if the server does not respond within 10 seconds. This integrates naturally with structured concurrency, ensuring resources are cleaned up even on timeout.

Can multiple Kourier RPC servers handle the same queue?

Yes. The server implementation uses channel.basicQos(count = 1u, global = false) to enable fair dispatch. When multiple server instances consume from the same rpc_queue, RabbitMQ round-robins messages between them, and each instance processes only one unacknowledged message at a time. This allows horizontal scaling of RPC workers without code changes.

Why is correlationId required in Kourier RPC?

The correlationId property ensures that a client can match responses to requests when using a shared callback queue or when multiple RPC calls run concurrently. According to the implementation in RPCTest.kt, the client generates a UUID, includes it in the request properties, and filters incoming messages until it finds one with a matching correlationId. This prevents race conditions and ensures request-reply integrity.

What is the difference between noAck=true and noAck=false in Kourier RPC?

In the Kourier RPC pattern, noAck (no acknowledgment) controls message reliability. The server uses noAck = false when consuming from rpc_queue, requiring explicit channel.basicAck() after processing to ensure messages are not lost if the server crashes. The client uses noAck = true when consuming from its temporary callback queue because it does not need to acknowledge replies; the queue is exclusive and auto-deleted when the client disconnects anyway.

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