Is it possible to get hot flows or are all flows c...

# coroutines

[janvladimirmostert]

Is it possible to get hot flows or are all flows cold and I should rather use a channel?

Use-case: logging

When I have a 10k concurrent things happening and they are all logging I want to throw the log line on a "channel" and not block the call-site until the log is completed
Then on a consumer side, I want to process the incoming logs in batches, grab a 100 at a time or until the "channel" is empty and do a single println for example or a single POST if it's going to an external logging server

[Casey Brooks]

SharedFlow and StateFlow are both hot flows, or you can emit the events to a Channel and read the events from the channel with receiveAsFlow() to process them with all the normal Flow operators.

The difference between the two approaches is mainly how many collectors do you want to allow. With Shared/State Flows, a single message will be sent to all collectors. If you want to ensure each message is only ever processed exactly once regardless of how many collectors are subscribed, then you'll want to use the Channel instead.

[janvladimirmostert]

that sounds awesome!

so if I have a console collector and an api collector, I can effectively "fanout" the logs to two different places

[Casey Brooks]

Yup, I've got a Desktop application where I'm basically doing the same thing (printing to logs in the console, displaying in UI widgets of the application itself, and writing to file).

[janvladimirmostert]

The StateFlow looks brilliant for managing application state, can't quite see how it's used for handling log streams since it doesn't react if the value stays the same

val state = MutableStateFlow(0)
launch {
   state.collect {
      println("State changed to $it")
   }
}

Channels on the other hand with the receiveAsFlow as you mentioned, works great!

val channel = Channel<Int>()
launch {
   channel.receiveAsFlow().collect {
      println("Received $it on channel1")
   }
}
launch {
   channel.receiveAsFlow().collect() {
      println("Received $it on channel2")
   }
}

launch {
   var i = 0
   while (true) {
      state.value = i
      state.value = i
      channel.send(i)
      channel.send(i)
      i++
      delay(1000)
   }
}

State changed to 110
Received 110 on channel1
Received 110 on channel2
State changed to 111
Received 111 on channel1
Received 111 on channel2
State changed to 112
Received 112 on channel1
Received 112 on channel2

Thanks for the help!

[gildor]

since it doesn’t react if the value stays the same

It’s a case for SharedFlow

[janvladimirmostert]

brilliant, I see SharedFlow effectively works like what I think of as an EventBus in which case it sounds perfect for logging