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

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), the contract is defined as:

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).

1. Direct Parameter Invocation

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

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 instance:

import dev.kourier.amqp.states.DeclaredQos

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

The DeclaredQos data class in 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 for fluent configuration:

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, generates a DeclaredQos object that the extension function in 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) demonstrate the canonical pattern for round-robin work distribution:

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).
• Alternative APIs: Extension functions accepting DeclaredQos objects or DSL builder blocks (defined in 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 directly maps to this wire protocol constraint, ensuring type safety and preventing overflow errors at the compilation stage.

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