example.kt

fun receiveMessages() = flow {
            while(isActive) {
                val messages = getMesagesFromSqs() // getMessages here takes 1 second
                logger.info("Nr of received messages $nrOfReturnedMessages")
                emit(messages)
            }
        }
}

suspend fun processMessage(flow: Flow<List<Message>>) {
        flow.buffer(1). collect {
            it.map {
                GlobalScope.async {
                    logger.info("Processing a message $it")
                    nrOfProcessedMessages++
                    delay(10000)
                }
            }.awaitAll()
            logger.info("Processed messages: $nrOfProcessedMessages")
        }
    }
}

private suspend fun getMesagesFromSqs(): List<Message> {
            // TODO: Real impl
            logger.info("SqsCall to receive messages")
            delay(1000)
            nrOfReturnedMessages +=10
            return (1..10).map {
            Message().addAttributesEntry("$it", "$it")
            }
}