# How to Set QoS (Prefetch Count) for Consumer Flow Control in Kourier

> Learn how to set QoS prefetch count in Kourier for effective consumer flow control. Use basicQos() before basicConsume() to manage message delivery and optimize performance.

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

---

**Use the `basicQos()` suspend function on an `AMQPChannel` to configure the prefetch count before calling `basicConsume()`, passing the limit as a `UShort` and a boolean flag to control whether the limit applies globally or per-consumer.**

Kourier, the Kotlin multiplatform AMQP client from guimauvedigital/kourier, implements RabbitMQ’s **basic.qos** command through a type-safe suspending API. Configuring the **QoS prefetch count** determines how many unacknowledged messages the broker may deliver to a consumer before pausing delivery, providing essential flow control and fair dispatch across distributed workers.

## The `basicQos()` Method Signature

According to the source code in [[`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt)](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt), the contract is defined as:

```kotlin
suspend fun basicQos(count: UShort, global: Boolean = false): AMQPResponse.Channel.Basic.QosOk

```

**Key parameters:**

- **`count`** (`UShort`): The maximum number of unacknowledged messages the broker will deliver. When this limit is reached, delivery pauses until a message is acknowledged, rejected, or nacked.
- **`global`** (`Boolean`): Determines the scope of the limit. When `false` (default), the prefetch count applies only to the current consumer. When `true`, the limit applies collectively to **all** consumers on the channel.

## Three Ways to Configure Prefetch Count

Kourier offers multiple idiomatic Kotlin patterns for invoking QoS settings, defined 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)](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt).

### 1. Direct Parameter Invocation

The most straightforward approach calls `basicQos` with explicit arguments immediately after opening a channel:

```kotlin
val connection = createAMQPConnection(this, amqpConfig { server { host = "localhost" } })
val channel = connection.openChannel()

// Allow only 1 unacknowledged message per consumer (fair dispatch)
channel.basicQos(count = 1u, global = false)

val consumer = channel.basicConsume(queueName, noAck = false)
for (delivery in consumer) {
    processMessage(delivery)
    channel.basicAck(delivery.message, multiple = false)
}

```

This pattern ensures the broker sends the next message only after the current one is processed and acknowledged.

### 2. Using a `DeclaredQos` Data Object

For configurations constructed elsewhere in your application, pass a [`DeclaredQos`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQos.kt) instance:

```kotlin
import dev.kourier.amqp.states.DeclaredQos

val qosSettings = DeclaredQos(count = 5u, global = false)
channel.basicQos(qosSettings)

```

The [`DeclaredQos`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQos.kt) data class in [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQos.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQos.kt) encapsulates the `count` and `global` properties for type-safe transport between configuration layers.

### 3. DSL Builder Syntax

Kourier provides a [`DeclaredQosBuilder`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQosBuilder.kt) for fluent configuration:

```kotlin
import dev.kourier.amqp.channel.declaredQos

channel.basicQos(declaredQos {
    count = 10u
    global = true  // Apply to all consumers on this channel
})

```

The builder, defined in [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQosBuilder.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/states/DeclaredQosBuilder.kt), generates a `DeclaredQos` object that the extension function in [`Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/Extensions.kt) accepts.

## Global vs. Per-Consumer Scope

Choosing the correct scope prevents bottlenecks in multi-consumer architectures:

- **Per-consumer (`global = false`)**: The default and most common pattern. Each consumer maintains its own prefetch buffer. Use this when individual workers have varying processing speeds or when you open multiple consumers on the same channel and want independent flow control for each.

- **Global (`global = true`)**: Creates a shared prefetch limit across every consumer on the channel. If you set `count = 10` with `global = true` and open three consumers, the three consumers together may hold only 10 unacknowledged messages total. This is useful for simple applications using a single channel with multiple consumers that share a collective throughput ceiling.

## Complete Fair Dispatch Example

The tutorial tests in [[`amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/WorkQueuesTest.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/WorkQueuesTest.kt)](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonTest/kotlin/dev/kourier/tutorials/WorkQueuesTest.kt) demonstrate the canonical pattern for round-robin work distribution:

```kotlin
suspend fun workerCoroutine(channel: AMQPChannel, queueName: String) {
    // Critical: set QoS before consuming to ensure flow control is active
    channel.basicQos(count = 1u, global = false)
    
    val consumer = channel.basicConsume(queueName, noAck = false)
    
    for (delivery in consumer) {
        performWork(delivery.body)
        channel.basicAck(delivery.message, multiple = false)
    }
}

```

**Critical ordering rule:** Always invoke `basicQos()` **before** `basicConsume()`. The prefetch setting takes effect immediately and persists for the lifetime of the channel (or until you issue another `basicQos` call to reconfigure it).

## Summary

- **Primary API**: The `basicQos(count: UShort, global: Boolean)` suspend function on `AMQPChannel` (defined in [`AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/AMQPChannel.kt)).
- **Alternative APIs**: Extension functions accepting `DeclaredQos` objects or DSL builder blocks (defined in [`Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/Extensions.kt)).
- **Timing**: Call `basicQos()` once per channel before starting consumption.
- **Scope**: Use `global = false` for per-consumer limits (recommended) or `global = true` for channel-wide collective limits.
- **Effect**: The broker pauses delivery when unacknowledged messages reach the prefetch count, resuming only after acknowledgments flow back.

## Frequently Asked Questions

### What happens if I don't set a prefetch count in Kourier?

Without calling `basicQos()`, RabbitMQ pushes messages to the consumer as fast as network capacity allows, regardless of the consumer's processing rate. This can lead to memory exhaustion on the client and unfair distribution where one busy worker hoards messages while idle workers starve.

### Can I change the prefetch count after starting a consumer?

Yes. You may call `basicQos()` at any time while the channel is open; the new limit takes effect immediately for subsequent deliveries. However, messages already in transit or buffered by the client are not affected by the change until they are processed and acknowledged.

### What is the difference between `basicQos` and `noAck` mode?

The `noAck` parameter in `basicConsume()` disables acknowledgments entirely, meaning the broker removes messages from the queue immediately upon delivery. In contrast, `basicQos` operates on channels with acknowledgments enabled (`noAck = false`), creating a sliding window of delivered-but-not-yet-acknowledged messages. You cannot use prefetch limits effectively with `noAck = true` because there is no mechanism to signal message processing completion back to the broker.

### Why does Kourier use `UShort` for the count parameter?

RabbitMQ’s AMQP 0-9-1 protocol defines the prefetch count field as a 16-bit unsigned integer. Kourier’s use of Kotlin’s `UShort` type in [`AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/AMQPChannel.kt) directly maps to this wire protocol constraint, ensuring type safety and preventing overflow errors at the compilation stage.