To Flow or not to Flow? Message subscription in Kotlin.
Tags: kotlin, coroutines, api
Published: 2022-08-12 10:31:49

In this short article I want to discuss to different patterns of doing reaction subscriptions in Kotlin via callbacks and via Flow.

Let's assume that you have some kind of object or structure and you want to observe changes done to this structure. 

The classic way to this is the following:

```kotlin
fun MyStructure.onChange(block: MyStructure.(Key) -> Unit)
```
and then use it like this:
```kotlin
myStructure.onChange{ key ->
  println(get(key))
}
```

I've intentionally done it using Kotlin receiver feature because it makes the classic layout even more neat. You pass not only the change place, but also a reference to a changed object so you can use pre-defined block function.

Now let's see how it looks using Kotlin coroutines [Flow API](https://kotlinlang.org/docs/flow.html):

```kotlin
val MyStructure.changes: Flow<Key>

myStructure.changes.onEach{ key ->
  println(myStructure.get(key)
}.launchIn(scope)
```

It seems like Flow is much harder to use. Also it requires a `CoroutineScope` to be used. And since you should not use `GlobalScope` for this, you need to think about the scope lifecycle. 

I do not want to talk about performance issues here. If you have so many subscribers, it is a performance problem, you need to solve this problem elsewhere, but `Flow` overhead is probably larger as well.

On the other hand, you can easily do transformations with events from `Flow`. Like this:

```kotlin
myStructure.changes
  .filter{ key -> key.startsWith("my")}
  .map{ key -> myStructure.get(key)}
  .onEach{ value -> println(value)}
  .launchIn(scope)
```

---

## But there is a problem people frequently forget about

If you make a subscription, you probably want to unsubscribe as well. Otherwise it leads to memory leaks (in VM sense) - some objects avoiding garbage collections because somebody holds their subscription handle.

So you write a code like this:

```kotlin
fun MyStructure.onChange(block: MyStructure.(Key) -> Unit)
fun MyStructure.removeChangeListener(block: MyStructure.(Key) -> Unit)
```
Actually, please do **not** write a code like this. Because in this case you rely on function equality to remove the correct listener and function equality is not reliable. Moreover, you will not be able to create a function via lambda like I've done above. You need to create it separately and store somewhere.

I usually do it this way:
```kotlin
fun MyStructure.onChange(owner: Any, block: MyStructure.(Key) -> Unit)
fun MyStructure.removeChangeListener(owner: Any)
```

In this case, I use the `owner` object to check equality of a callback holder. If a single owner holds several callbacks, all of them will be removed and it has its uses. The only problem now is not to forget to remove the listener when it is no longer used (and it means that you need to think about lifetimes like in case of `Flow`). 

The Flow does not require specific unsubscription methods. Usually if you stop consuming the Flow, it is not used anymore. Also you can manually unsubscribe from it by canceling the subscription job:

```kotlin
val job = myStructure.changes.onEach{ key ->
  println(myStructure.get(key)
}.launchIn(scope)

job.cancel()
```
Due to [structured concurrency](https://kotlinlang.org/docs/coroutines-basics.html#structured-concurrency), you usually do not need to think about canceling the job manually. When the external scope is closed, all child jobs are canceled automatically. So the lifetime management is automatic.

Still, you need to manage the lifetime via coroutine scope (which is quite convenient).

---

Both ways are valid and I can't say that one of them should be always used instead of the other. I think I would prefer the `Flow` way for complex systems because there is not risk of forgetting to unsubscribe. But in some cases you do not want to bother with scopes.



----

Cover image photo by <a href="https://unsplash.com/@chrisstenger?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText">Chris Stenger</a> on <a href="https://unsplash.com/s/photos/flow?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText">Unsplash</a>