Open ageorgousakis opened 1 year ago
I found a way to make it work. It seems runBlocking
blocks the event loop of Ktor engine. The receiver and Ktor route handlers works fine if it is started in the following way:
on(MonitoringEvent(ApplicationStarted)) { application ->
// .. code to initialise the receiver
CoroutineScope(Dispatchers.IO).launch {
receiver
.receive("my-topic")
.collect {
application.log.info("${it.key()} -> ${it.value()}")
it.offset.acknowledge()
}
}
}
I don't know which is the best way to start a receiver in Ktor.
@nomisRev Can you recommend or give us an example how to use it with Ktor?
Regards
Hey @ageorgousakis,
Thanks for your interest in the library, and opening a ticket! 🙏
As you've show there already, the returning Flow
is coupled to it's surrounding CoroutineScope
.
The simplest way to running a Flow for the same length as the server is to use launchIn
with Application
.
So to provide a small snippet:
val flow: Flow<Unit> = receiver
.receive("my-topic")
.map { // <-- changed collect to map, so the result is still a Flow
application.log.info("${it.key()} -> ${it.value()}")
it.offset.acknowledge()
}
val application: Application = TODO("This is the Application from Ktor")
flow.launchIn(application)
The Ktor Application
implements CoroutineScope
, and it cancels the CoroutineScope
when the server is cancelled.
So it will also cancel your Flow
.
NOTE: I've run into some issues where Ktor doesn't properly cancel on SIGINT
from K8S for example, and I've therefor build suspendapp.
SuspendApp With Ktor and K8S
SuspendApp with Kafka.
Alternatively, you could also use ApplicationEngine#addShutdownHook and use it to manually control a CoroutineScope
.
The take-away is the the Flow
cancels with the CoroutineScope
it's called from.
Hope that helps @ageorgousakis ! Let's leave this issue open, so it can serve as reminder to include this information it in the documentation 👍
Hi,
I'm trying to user KafkaReceiver inside Ktor 2. I'm not quite sure what is the best way to start KafkaReciever with Ktor. What I did was to create an application plugin like the following and install it in Ktor engine.
I seems the receiver starts and collects the messages. The question is how can we stop the receiver when Ktor shutdowns. When I stop the Ktor application it seems stuck and I have to kill the process.