# Having difficulty adapting the RabbitMQ client to coroutines

**URL:** <https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163>\
**Category:** Support\
**Created:** [February 12, 2022, 1:48am UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163 "2022-02-12T01:48:09Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![mrpug](https://avatars.discourse-cdn.com/v4/letter/m/e5b9ba/32.png) [@mrpug](https://discuss.kotlinlang.org/u/mrpug)\
**Post date:** [February 12, 2022, 1:48am UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/1 "2022-02-12T01:48:09Z")

</div>

Hi all,

I’m attempting to integrate a RabbitMQ consumer into a Kotlin project that uses coroutines heavily, but I’m having some issues adapting the RabbitMQ calls into coroutine calls. Has anyone done anything similar to this?

I’m running into a couple of different roadblocks. If I try to turn the below code into

```auto
val deliverCallback = DeliverCallback { _: String, delivery: Delivery ->
    val message = delivery.body.toString(Charset.forName("UTF-8"))
    // process stuff
    channel.basicAck(delivery.envelope.deliveryTag, false)
}

val cancelCallback = CancelCallback {
    //
}

try {
    channel.basicConsume(queueName, false, deliverCallback, cancelCallback)
} catch (e: IOException) {
    //
}

```

into a suspending function using `produce<Foo>()`

```auto
suspend fun test() = withContext(Dispatchers.IO) {
    return@withContext produce<Delivery> {
        val deliverCallback = DeliverCallback { _: String, delivery: Delivery ->
            // send(delivery) // cannot call
        }

        val cancelCallback = CancelCallback {
            println("cancel")
        }

        rabbitChannel.basicConsume(queueName, false, deliverCallback, cancelCallback)
    }
}

```

I get an error from the compiler. I’ve also tried this as a `flow`, but the model didn’t fit well because I this consumer needs to run for the entirety of the application, and it also needs to react to a message as soon as its received.

Secondly, it seems that I may be able to use `Channel.asyncCompletableRpc` by putting it inside of a `suspendCoroutine` and calling resume in the `handleAsync`, but if I used it to send a `AMQP.Basic.Consume`, the default consumer would be called and I’d be back to a listener in which I can’t call `send` from since it’s not suspending.

Additionally, the `asyncCompletableRpc` approach causes Intellij to warn about improper blocking (side-note, would the withContext([Dispatchers.IO](http://Dispatchers.IO)) {} help in practice?).

```auto
@Suppress("BlockingMethodInNonBlockingContext")
suspend fun ack(tag: Long) = withContext(Dispatchers.IO) {
    return@withContext suspendCoroutine<Unit> { cont ->
        rabbitChannel.asyncCompletableRpc(AMQP.Basic.Ack.Builder().deliveryTag(tag).build())
            .handleAsync { _: Command?, err: Throwable? ->
                if (err == null) {
                    cont.resumeWith(Result.success(Unit))
                } else {
                    cont.resumeWithException(err)
                }
        }
    }
}

```

Are there any other ways to wrap the RabbitMQ client into a coroutines framework? Is a `Producer` the right approach here? Or is there a better approach in reactive streams / reactor? Thanks in advance.

---

<div class="post-metadata">

**Author:** ![qwert](https://avatars.discourse-cdn.com/v4/letter/q/7bcc69/32.png) [@qwert](https://discuss.kotlinlang.org/u/qwert)\
**Post date:** [February 12, 2022, 9:50am UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/2 "2022-02-12T09:50:23Z")

</div>

I think there are several problems here:

1. You can not call `send` in the DeliveryCallback, because `send` is a suspend function, but the `handle` method in the definition of the `DeliveryCallback` interface is not. So the compiler is right to complain, because the context from which your callback gets called might be (and in fact really is in this case) just a regular (non-coroutine) thread.
2. I never worked with the plain RabbitMq client. But doing a quick google search it is my understanding that the `basicConsume` call does not block. So pay extra care here, because according to the docs produce will “close the channel when the coroutine completes” [https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html)

Regarding 1.: Take a look at `callbackFlow` ([https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/callback-flow.html](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/callback-flow.html)) maybe this is for you. If you want to stay with produce, it may help to take a look at `callbackFlow` anyways. The particular problem you have there was solved in `callbackFlow` by simply using `runBlocking` here (see method `trySendBlocking` [kotlinx.coroutines/Channels.kt at master · Kotlin/kotlinx.coroutines · GitHub](https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/concurrent/src/channels/Channels.kt)).  
So this is likely a valid use case for `runBlocking`. And in fact since `trySendBlocking` is just an extenstion on `SendChannel` you can just use it right there instead of `send`.  
Regarding 2: `awaitClose` ([https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/await-close.html](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/await-close.html)) might do the trick.

Your second problem looks to me like it has a similar cause.

---

<div class="post-metadata">

**Author:** ![broot](https://avatars.discourse-cdn.com/v4/letter/b/a88e57/32.png) [@broot](https://discuss.kotlinlang.org/u/broot)\
**Post date:** [February 12, 2022, 12:19pm UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/3 "2022-02-12T12:19:41Z")

</div>

`withContext { produce { ... } }` - ohh, this is actually a deadlock 😃 In Kotlin we usually use suspending functions for synchronous execution and if a function needs to execute a task in the background, it is usually a non-suspending function accepting a `CoroutineScope`. Now look at `produce()`:

```kotlin
fun <E> CoroutineScope.produce(...)

```

It is the latter, it only creates a producer and immediately returns. Now, your function is `suspend` which means it doesn’t launch anything in the background, it waits for its children to finish. That means `test()` won’t return until the producer finishes, but nobody consumes elements from the producer before `test()` could finish.

Instead, make your function an extension on `CoroutineScope`, don’t use `withContext()` and additionally don’t use `Dispatchers.IO`, because you don’t need it here:

```kotlin
fun CoroutineScope.test() = produce<Delivery> { ... }

```

> [@mrpug](#):
>
> `// send(delivery) // cannot call`

As @qwert explained, this is because `send()` is a suspending function. And it is suspending only because the channel/buffer may be full. So you need to decide what should happen in such a case. If you want to block, waiting for the consumer of the channel - use `trySendBlock()`. If you want to reject the item - use `trySend()`. If you want to throw an exception - also `trySend()`.

In any case you may like to provide `capacity` argument to `produce()`. By default it doesn’t create any buffer, so it accepts new items only if there is a consumer waiting for items.

> [@mrpug](#):
>
> I’ve also tried this as a `flow` , but the model didn’t fit well because I this consumer needs to run for the entirety of the application, and it also needs to react to a message as soon as its received.

I’m not sure what do you mean here. No matter if you use Channel or Flow, there has to be a consumer (or consumers) running all the time. Also, in both cases, events are not processed immediately, but wait for the consumer to be available. If you need to guarantee that events are processed immediately then I think neither channels nor flows will work here. You would need to accept a callback to `test()`.

~~But if you don’t need it to be immediate, then I also encourage you to use `callbackFlow()`. Flows are usually easier to use than channels.~~

OK, after thinking about it a little, I’m not sure regarding channel vs flow. Flows are more about observing for changes or some data. Channels are for distributing of events or units of work between multiple producers and consumers. So actually, channel could be a better fit here.

---

<div class="post-metadata">

**Author:** ![broot](https://avatars.discourse-cdn.com/v4/letter/b/a88e57/32.png) [@broot](https://discuss.kotlinlang.org/u/broot)\
**Post date:** [February 12, 2022, 12:52pm UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/4 "2022-02-12T12:52:57Z")

</div>

Regarding the second part.

> [@mrpug](#):
>
> it seems that I may be able to use `Channel.asyncCompletableRpc` by putting it inside of a `suspendCoroutine` and calling resume in the `handleAsync`

Kotlin already provides adapter for suspending on `CompletableFuture`, so you can just do:

```kotlin
suspend fun ack(...) = rabbitChannel.asyncCompletableRpc(...).await()

```

> [@mrpug](#):
>
> Additionally, the `asyncCompletableRpc` approach causes Intellij to warn about improper blocking

I guess this is because `asyncCompletableRpc()` may throw `IOException`, so IntelliJ assumed it is blocking. Its name and docs clearly say it is asynchronous, so `Dispatchers.IO`, should not be needed.

But… I must say docs of RabbitMQ are confusing to me here. It says: “Asynchronously send a method over this channel.” and that it throws `IOException` if there is “Problem transmitting method.”. This is ridiculous, it can’t be both. I would assume it performs only some non-blocking check initially (which may result in `IOException`) and then it performs sending asynchronously. In this case `Dispatchers.IO` is not needed, but it is really unclear to me ¯\_(ツ)\_/¯

edit: Or it sends the data synchronously, by blocking and then only receives ack asynchronously. Then it would probably require `Dispatchers.IO`. But this is contrary to what docs say.

> [@mrpug](#):
>
> but if I used it to send a `AMQP.Basic.Consume` , the default consumer would be called and I’d be back to a listener in which I can’t call `send` from since it’s not suspending.

I’m not sure what do you mean here. What is the problem?

---

<div class="post-metadata">

**Author:** ![viartemev](https://avatars.discourse-cdn.com/v4/letter/v/e19adc/32.png) [@viartemev](https://discuss.kotlinlang.org/u/viartemev)\
**Post date:** [November 15, 2023, 8:14am UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/5 "2023-11-15T08:14:37Z")

</div>

You can use library for it: [GitHub - viartemev/rabbitmq-kotlin: RabbitMQ library for Kotlin based on coroutines](https://github.com/viartemev/rabbitmq-kotlin)

---

<div class="post-metadata">

**Author:** ![Denis-Tudor\_Damir](https://avatars.discourse-cdn.com/v4/letter/d/b5ac83/32.png) [@Denis-Tudor\_Damir](https://discuss.kotlinlang.org/u/Denis-Tudor_Damir)\
**Post date:** [January 18, 2025, 1:35pm UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/6 "2025-01-18T13:35:36Z")

</div>

Or you can use:

> **[GitHub - DamirDenis-Tudor/ktor-server-rabbitmq: Official RabbitMQ plugin for Ktor framework.](https://github.com/DamirDenis-Tudor/ktor-server-rabbitmq)**
>
> Official RabbitMQ plugin for Ktor framework.

---

<div class="post-metadata">

**Author:** ![NathanFallet](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/nathanfallet/32/11105_2.png) [@NathanFallet](https://discuss.kotlinlang.org/u/NathanFallet)\
**Post date:** [December 2, 2025, 3:25pm UTC](https://discuss.kotlinlang.org/t/having-difficulty-adapting-the-rabbitmq-client-to-coroutines/24163/7 "2025-12-02T15:25:53Z")

</div>

Wrote an article about this: [Stop wrapping your RabbitMQ code in runBlocking - DEV Community](https://dev.to/nathanfallet/stop-wrapping-your-rabbitmq-code-in-runblocking-18c2)

Using Kourier or ktor-server-rabbitmq are both great options to have seamless coroutine integration with RabbitMQ.