Hello, I have this class that reads messages from...

# coroutines

[Paulius Ruminas]

Hello,

I have this class that reads messages from rabbitmq and sends acknowledgement if consumer processed the message successfully or sends a retry if the consumer fails. I managed to achieve this with two flows because I need a try/catch block to send acknowledgement or retry. Is there a more idiomatic way to solve this problems by using only one flow?

private typealias MessageTag = Long

interface ClientMessageConsumer {
    fun consume(): Flow<String>
}

class AmqpClientMessageConsumer(connector: AmqpConnector) : ClientMessageConsumer {

    // Creates amqp channel and bindings
    private val amqpChannel = connector.createChannel {
        queueDeclare(AmqpClient.QueueName(1), true, false, false, mapOf("x-expires" to AmqpClient.QueueTtl))
        queueBind(AmqpClient.QueueName(1), AmqpClient.ExchangeName, "")
    }

    override fun consume(): Flow<String> {
        // Consuming messages from rabbitmq queue
        val callbackFlow = callbackFlow<Pair<MessageTag, String>> {
            val consumerTag = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
                amqpChannel.basicConsume(AmqpClient.QueueName(1), false, object : DefaultConsumer(amqpChannel) {
                    override fun handleDelivery(consumerTag: String, envelope: Envelope, properties: AMQP.BasicProperties, body: ByteArray) {
                        sendBlocking(Pair(envelope.deliveryTag, String(body)))
                    }
                })
            }

            awaitClose {
                amqpChannel.basicCancel(consumerTag)
            }
        }

        return flow {
            callbackFlow.collect { (tag, message) ->
                try {
                    emit(message)
                    amqpChannel.acknowledge(tag)  // Tells rabbitmq that the message was processed and needs to be removed from the queue
                } catch (e: Throwable) {
                    amqpChannel.retry(tag)        // Tells rabbitmq that the message was not processed and needs to be put back into the queue
                    throw e
                }
            }
        }
    }
}