How to wrap single callback with multiple flows

user_946409 0
Viewed 520

I have 3rd party a library, where you can subscribe to some topic to receive updates on that topic (with callback). But the problem is, that when you call the subscribe method with the same topic again, the previous subscription is silently replaced with the new one.

Now, I'd like to replace the callback with Flow. So I wrote the following code:

val flow: Flow<Message> = callbackFlow {
    lib.subscribe(topic) { message -> offer(message) }
    awaitClose { lib.unsubscribe(topic) }
}

But when I call collect multiple times, only the last one is getting the messages.

So how can I achieve this:

  • Have multiple subscriptions to the same topic
  • When the last subscriber closes/cancels, unsubscribe from the given topic.

I've already looked at BroadcastChannel, but I couldn't find anything to solve the second requirement. And I've just started exploring Flows, so maybe there is some simple wrapper which I simply missed.

kotlin-coroutines
1 Answers

If you convert your Flow to a SharedFlow (of course, you could still expose it as the more abstract Flow type), then subsequent subscribers will be added without re-executing the block that's passed to callbackFlow(), meaning that the original subscriber to the library won't be replaced.

The Kotlin Flow library provides a Flow<T>.shareIn(): SharedFlow<T> extension, which makes it really easy to convert to a SharedFlow.

There's one other important part to this answer. When you call shareIn(), you'll need to pass SharingStarted.WhileSubscribed as one of the args. This is crucial in your use case because without that, awaitClose() may not be called, which means that you wouldn't be unsubscribed from the library. With WhileSubscribed, completely unsubscribing from the SharedFlow will also result in unsubscribing from the original Flow, which will trigger a call to awaitClose().

Your code might look like this:

val flow: Flow<Message> = callbackFlow {
    lib.subscribe(topic) { message -> offer(message) }
    awaitClose { lib.unsubscribe(topic) }
}.shareIn(coroutineScope, SharingStarted.WhileSubscribed(), replay = 0)

From the doc on shareIn():

WhileSubscribed() — starts the upstream flow when the first subscriber appears, immediately stops when the last subscriber disappears...

user_22684270
Related