How do I attach a piece of code to run when a

Flow

is cancelled/disposed. Something like

ObservableEmitter.setCancellable { ... }

in the rx world. I want to use

Flow

to model an add/remove listener so that a listener is attached when the flow is “subscribed” to and detached when the subscription is “disposed”. Forgive the rx lingo, pretty new to

Flow

so I will also appreciate reading material that can tie or at least differenciate between my RxJava knowlege and

Flow

perhaps

onErrorCollect

or

onErrorReturn

?

Yeah I’m definitely missing a correct mental model.

flowViaChannel

is what you need : https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow-via-channel.html You can't directly cancel a

Flow

, but you can cancel the coroutine it is running in. This will cause the underlying channel to be closed, triggering its

invokeOnClose

block and thus removing your listener.

Works like a charm. Thank you @tseisel 🙏

If you don’t have callbacks from outside APIs, then you don’t need a channel and you can use a simple `try-finally`:

flow {
    try { 
        while (isActive) { 
             /* emit elements */ 
        }
    } finally { 
        /* called when it is cancelled */ 
    }
}