How to Consume Messages Using Flow-Based API in Kourier

To consume AMQP messages as a Kotlin Flow in Kourier, call basicConsume() to obtain an AMQPReceiveChannel, then invoke the consumeAsFlow() extension to transform the channel into a cold flow of AMQPResponse.Channel.Message.Delivery objects.

The guimauvedigital/kourier library provides a coroutine-native AMQP client that integrates seamlessly with Kotlin Flow. By leveraging the Flow-based API, you can process incoming messages using standard flow operators like map, filter, and combine, while maintaining structured concurrency and automatic backpressure handling.

Understanding the Flow-Based Consumption Architecture

Kourier’s channel API exposes the basicConsume function in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt at line 150. This function returns an AMQPReceiveChannel, defined in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPReceiveChannel.kt at line 6, which serves as a thin wrapper around kotlinx.coroutines.channels.ReceiveChannel:

class AMQPReceiveChannel(
    val consumeOk: AMQPResponse.Channel.Basic.ConsumeOk,
    val receiveChannel: ReceiveChannel<AMQPResponse.Channel.Message.Delivery>,
) : ReceiveChannel<AMQPResponse.Channel.Message.Delivery> by receiveChannel

Because AMQPReceiveChannel implements ReceiveChannel by delegation, you can convert the stream of deliveries into a Kotlin Flow using the standard consumeAsFlow() extension from kotlinx.coroutines.channels. This allows you to treat incoming AMQP messages exactly like any other cold or hot flow, applying operators, collecting in coroutines, and handling backpressure naturally.

Critical distinction: Kourier also provides a flowResponses property (located in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt), but that flow reports broker flow-control frames (Channel.Flow/Channel.FlowOk), not message deliveries. The Flow-based consumption of messages is achieved exclusively by wrapping the ReceiveChannel returned by basicConsume.

Consuming Messages with Kotlin Flow

Basic Flow Consumption

To start consuming messages as a Flow, open a connection and channel, then transform the AMQPReceiveChannel into a Flow using consumeAsFlow():

import dev.kourier.amqp.*
import dev.kourier.amqp.channel.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

suspend fun createChannel(scope: CoroutineScope): AMQPChannel {
    val config = amqpConfig {
        server { host = "localhost" }
    }
    val connection = createAMQPConnection(scope, config)
    return connection.openChannel()
}

suspend fun consumeLogs(scope: CoroutineScope, queue: String) {
    val channel = createChannel(scope)

    // basicConsume returns AMQPReceiveChannel (a ReceiveChannel)
    val receiveChannel = channel.basicConsume(queueName = queue, noAck = true)

    // Convert to Kotlin Flow
    val deliveryFlow: Flow<AMQPResponse.Channel.Message.Delivery> = receiveChannel.consumeAsFlow()

    // Process each delivery
    deliveryFlow
        .map { it.message.body.decodeToString() }
        .onEach { println("[log] $it") }
        .catch { e -> println("Error while consuming: $e") }
        .launchIn(scope)
}

The launchIn(scope) operator starts collection in the given coroutine scope, ensuring structured concurrency and proper cancellation propagation.

Applying Flow Operators

The Flow-based API enables sophisticated stream processing using standard operators. This example filters messages by routing key, throttles the throughput, and uses manual acknowledgments:

suspend fun consumeCriticalEvents(scope: CoroutineScope, queue: String) {
    val channel = createChannel(scope)

    val deliveryFlow = channel
        .basicConsume(queueName = queue, noAck = false)   // manual ack
        .consumeAsFlow()
        .map { it.message }                               // extract the raw AMQPMessage
        .filter { it.routingKey.startsWith("error.") }    // only error events
        .throttleLatest(500)                              // at most one every 500 ms
        .onEach { msg ->
            println("⚠️ ${msg.routingKey}: ${msg.body.decodeToString()}")
            channel.basicAck(msg.deliveryTag)
        }
        .catch { e -> println("Failed: $e") }

    deliveryFlow.launchIn(scope)
}

Combining Message Flows with Other Asynchronous Sources

You can combine the message stream with other flows, such as timers or state flows, to create complex processing pipelines:

suspend fun monitorWithMetrics(scope: CoroutineScope, queue: String) {
    val channel = createChannel(scope)

    val messageFlow = channel
        .basicConsume(queue, noAck = true)
        .consumeAsFlow()
        .map { it.message.body.decodeToString() }

    val ticker = tickerFlow(5.seconds)

    combine(messageFlow, ticker) { msg, _ -> msg }
        .onEach { println("[${Instant.now()}] $it") }
        .launchIn(scope)
}

Key Implementation Details

Structured Concurrency: All consumption runs on top of Kotlin coroutines, providing automatic cancellation propagation and scope management through the CoroutineScope.

Backpressure Handling: Since consumeAsFlow() creates a cold flow backed by a ReceiveChannel, standard backpressure mechanisms apply. The channel buffers messages according to its capacity, suspending the producer when the buffer is full to prevent memory overflow.

Manual Acknowledgments: When using noAck = false, you must call channel.basicAck(deliveryTag) within your flow collection logic. The Flow-based API does not provide automatic acknowledgment—you retain full control over message confirmation.

Summary

• Use basicConsume() to obtain an AMQPReceiveChannel from AMQPChannel, as implemented in guimauvedigital/kourier at AMQPChannel.kt.
• Transform to Flow by calling consumeAsFlow() on the AMQPReceiveChannel to access standard Kotlin Flow operators and composability.
• Distinguish from flowResponses—the latter handles broker flow-control frames, not message deliveries.
• Leverage structured concurrency by launching collection within a CoroutineScope, ensuring proper cancellation and resource cleanup.
• Handle backpressure naturally through the underlying ReceiveChannel mechanism, which suspends the producer when buffers are full.

Frequently Asked Questions

What is the difference between AMQPReceiveChannel and a standard Kotlin Flow?

AMQPReceiveChannel is a wrapper class defined in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPReceiveChannel.kt that delegates to a ReceiveChannel<AMQPResponse.Channel.Message.Delivery>. It is not itself a Flow, but because it implements ReceiveChannel, you can convert it to a Flow using consumeAsFlow(). This design preserves access to the consumeOk response metadata while enabling full Flow interoperability.

How does backpressure handling work when consuming messages as a Flow?

Backpressure is handled by the underlying ReceiveChannel buffer. When you convert the channel to a Flow using consumeAsFlow(), the flow respects the channel's buffer capacity. If your collector is slow, the channel will suspend the producer once the buffer fills, preventing memory exhaustion. You can configure the channel's buffer capacity when setting up the consumer if the default is insufficient for your throughput requirements.

Can I use manual acknowledgments with the Flow-based API?

Yes. Pass noAck = false to basicConsume(), then call channel.basicAck(deliveryTag) inside your onEach or collect block after successfully processing the message. The Flow-based API does not automatically acknowledge messages—you retain full control over when to ack, nack, or reject each delivery based on your processing logic.

Is there a performance difference between the callback API and Flow-based consumption?

Both APIs use the same underlying coroutine machinery in Kourier. The Flow-based approach introduces minimal overhead consisting of a single delegation layer through AMQPReceiveChannel and the consumeAsFlow() conversion. The primary architectural difference is composability: Flow provides superior operator chaining and stream merging capabilities, while callback APIs may be preferable for simple, high-throughput scenarios where you want to avoid the allocation overhead of flow operators.

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