# How to Consume Messages Using Flow-Based API in Kourier

> Learn to consume AMQP messages as a Kotlin Flow in Kourier. Use basicConsume and consumeAsFlow extension to get a cold flow of messages.

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

---

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

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

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

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

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