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

> Implement RPC request-reply with Kourier AMQP client. Publish messages with replyTo and correlationId, then match responses on an exclusive callback queue.

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

---

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

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

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

```kotlin
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`](https://github.com/guimauvedigital/kourier/blob/main/RPCTest.kt), demonstrates how to use Kourier’s DSL from [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/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:

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

```kotlin
// 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:

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

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