Hi everyone, trying out Channels for the first tim...

# coroutines

[Tash]

Hi everyone, trying out Channels for the first time. My use case is a “notification handler” type class where I can invoke notify(...) on an object of that class and it internally, with back-pressure awareness, processes and runs some computations on each notification.

This is on Android, and for this example creating an appScope by using GlobalScope + CoroutineExceptionHandler because I want to treat the processing of the notifications as “fire-and-forget”. Also trying to use consumeAsFlow() + collect() on the Channel:

class ChannelExample {
    
    private val appScope = GlobalScope + CoroutineExceptionHandler { coroutineContext, throwable -> 
        // Log message + throwable
    }
    private val numbersChannel = Channel<Int>()

    init {
        appScope.launch(Dispatchers.Default) {
            numbersChannel
                .consumeAsFlow()
                .collect { number -> handleNumber(number)}
        }
    }
    
    fun notifyNumber(number: Int) {
        appScope.launch {
            numbersChannel.send(number)
        }
    }

    private suspend fun handleNumber(number: Int) {
        try {
            // Do some computations
        } catch (ce: CancellationException) {
            throw ce
        } catch (ex: Exception) {
            // Log message + exception
        }
    }
}

1) do I need to call close() or cancel() on numbersChannel?
- if yes, where would be the best place to do that?
- if no, why not?
2) are there any other things in this example that stray from best practices?
3) are there better ways to achieve my original use case?

Thanks!

[Dominaezzz]

notifyNumber should just use offer.

[Zach Klippenstein (he/him) [MOD]]

Channels already support a "forEach" style API - don't use consumeAsFlow().collect, just use consumeEach directly.

[Zach Klippenstein (he/him) [MOD]]

No need to pass Dispatchers.Default to launch, you're basing your scope on GlobalScope which will already use that dispatcher.

[Zach Klippenstein (he/him) [MOD]]

How are you trying to be back-pressure aware? Since your notify method is fire-and-forget, you're losing backpressure there.

[Zach Klippenstein (he/him) [MOD]]

do I need to call close() or cancel() on numbersChannel?

Not necessarily, but you should at least ensure the coroutine you launch in the init block is cleaned up and doesn't leak. Usually the best approach here is to pass the base scope in through the constructor so this class doesn't have to concern itself with lifetime. In other words, make use of structured concurrency. Then whatever is instantiating this class is responsible for ensuring that a scope with the right lifetime is used. E.g. if this is instantiated by an Application, your Application class can probably just pass GlobalScope since it is a process-scoped Singleton. And in your unit tests, you can pass in a scope that gets cancelled after every test.

If you do that, then because you're using the consume* APIs the channel will automatically get cancelled when the scope is.

[Tash]

thanks @Zach Klippenstein (he/him) [MOD] re: back-pressure: for this example, just wanted to experiment with the default RendezvousChannel. If i understand correctly, every time notifyNumber() is invoked the coroutine that sends on the channel will suspend until the receiver is ready to receive more items…Hence I was also unsure about the difference in this case of using `send`/`offer`

re: `GlobalScope`: Yes that’s correct; in the actual project, the scope will be passed into the class’s constructor through a DI setup that provides GlobalScope.

re: "If you do that, then because you're using the consume* APIs the channel will automatically get cancelled when the scope is." : Ahh that’s what I was not sure about, thanks for clarifying! Was confused with the relationship of scope + channel. Will just use the consumeEach directly.

[Tash]

thanks @Dominaezzz
i see; do you happen to know the differences in behavior for using send vs offer specifically for a RendezvousChannel?

[Dominaezzz]

offer drops the item if there's no coroutine calling receive. send waits for receive to be called.

[Dominaezzz]

Would be better to use offer with unlimited channel, instead of launching new coroutine for every send.

[Tash]

ahh i see, makes sense esp if many sends are anticipated, thanks!