1. Publishing and subscribing in SharedFlow
If you’ve just written a neat Event<T> with a snapshot and an unsubscribe token, a logical question sounds like this: “Why do I need some SharedFlow too? I’ve just defeated the chaos.” And that’s a fair question: your own Event<T> really is suitable for synchronous broadcasting within regular code.
But as soon as we enter the world of coroutines, a new lifestyle appears: subscribers become coroutines, it’s more convenient to control a subscription’s “lifecycle” via Job, and event broadcasting naturally fits into the data-stream model (Flow). Kotlin provides several ways to exchange data between coroutines, and SharedFlow is described as a mechanism that “shares each value with all active collectors.”
The point of this lecture: learn to use SharedFlow as an “event bus” in coroutine code, understand how subscription and “unsubscription” work, and why the replay parameter makes learning examples stable and not dependent on “made it in time / didn’t make it in time.”
MutableSharedFlow and SharedFlow: “remote” and “screen”
When you first see MutableSharedFlow<T>, you might want to hand it out to everyone: “let whoever wants emit.” That’s roughly like leaving the “launch missile” button on the kitchen table next to the cookies. Formally convenient, but then you spend half a day figuring out who and why “fired” an extra event.
In the coroutine world, it’s common to separate roles: MutableSharedFlow<T> is the publication point (events get “pushed” there), and SharedFlow<T> is the safe “read-only” view (events are only read from it via collect). This approach aligns well with the general idea of separation of concerns and reduced coupling: externally, we expose the ability to subscribe, but we don’t expose the ability to publish from anywhere.
Let’s build a minimal “event bus” for our console app (let it be a small CLI task tracker with add/remove commands, where events are needed for auditing and notifications).
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
sealed interface AppEvent
data class TaskAdded(val text: String) : AppEvent
class EventBus {
private val _events = MutableSharedFlow<AppEvent>(replay = 1)
val events: SharedFlow<AppEvent> = _events.asSharedFlow()
suspend fun emit(event: AppEvent) = _events.emit(event)
}
Notice two things. First, _events is private and “mutable” (you can change it; you can emit into it). Second, what’s exposed is events: SharedFlow<AppEvent> — subscribe as much as you want, but you can’t “publish” directly.
Subscribing via collect: the “listener” lives in a coroutine
In our homemade event system, a listener was a regular function (T) -> Unit, which we stored in a list. In SharedFlow, the listener becomes a collector coroutine: you start launch { flow.collect { ... } }, and as long as that coroutine is alive, it receives events.
This is very convenient for lifecycle management: “unsubscription” is just cancelling the Job. No searching for “that exact lambda,” no remove(listener). A subscription lives exactly as long as the coroutine lives, and that matches the idea of structured concurrency perfectly: child tasks live inside the parent scope.
A minimal example: “two subscribers receive all events”:
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect
fun main() = runBlocking {
val bus = EventBus()
val a = launch {
bus.events.collect { println("A got: $it") }
}
val b = launch {
bus.events.collect { println("B got: $it") }
}
bus.emit(TaskAdded("Buy milk")) // A got: TaskAdded(text=Buy milk), B got: TaskAdded(text=Buy milk)
a.cancel() // unsubscribed
b.cancel() // unsubscribed
}
The key idea here: collect is a “subscription” that typically does not end on its own. If you don’t cancel the job, the collector will keep waiting for new events. For an “event bus,” that’s normal: events are potentially infinite.
“Unsubscription” in SharedFlow: cancel the Job, don’t search for a lambda
In the lecture about unsubscribe tokens, we emphasized the “reference nature of lambdas”: you can unsubscribe only with the same reference. In SharedFlow, this layer of problems almost disappears, because a subscription is not an “element in a list,” but a running coroutine.
So the practical unsubscribe recipe looks like this: keep the Job from launch, then call cancel() (sometimes also join() — but today, a simple cancellation is enough). It’s also nice because coroutine cancellation is a familiar tool you already know from previous days about structured concurrency.
A super-short example: “subscribed → got one event → unsubscribed”:
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect
fun main() = runBlocking {
val bus = EventBus()
val job = launch {
bus.events.collect { println("Audit: $it") }
}
bus.emit(TaskAdded("Read Kotlin docs")) // Audit: TaskAdded(text=Read Kotlin docs)
job.cancel() // subscription stopped
}
If you need a “temporary listener” (for example, “log only while a scenario is running”), Job.cancel() is exactly what the doctor ordered.
3. replay and “late” subscribers
The most common learning pain with event streams sounds like this: “I did emit, but the subscriber got nothing. Did Kotlin break?” In fact, no: you accidentally demonstrated the classic late-subscriber scenario — the subscriber started listening after the event had already flown by.
By definition, a regular Flow produces values only when it’s being collected (collect), and SharedFlow distributes values to all active collectors. But if there were no collectors at the moment of the event, then for a SharedFlow with replay = 0, that event is essentially “not needed by anyone,” and it isn’t stored.
That’s why the replay parameter is so useful specifically in learning examples: it lets you store the last N events and “replay” them to new subscribers. In this lecture we use replay = 1 so a new subscriber is guaranteed to see at least the “last fact,” and the example doesn’t depend on timing.
Let’s look at two behaviors: first, (mentally) imagine replay = 0, then replay = 1. We’ll keep our EventBus with replay = 1, but we’ll show why it matters.
Late subscription: event first, subscriber later
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect
fun main() = runBlocking {
val bus = EventBus()
bus.emit(TaskAdded("EARLY")) // no subscribers, but replay=1 will remember the last one
val job = launch {
bus.events.collect { println("Late got: $it") }
}
// Thanks to replay=1, the subscriber will immediately receive "EARLY"
job.cancel()
}
If replay were 0, the line Late got: ... might not appear at all, and the student (and sometimes the instructor) would start suspecting a conspiracy of the timer and scheduler.
What replay = 1 does in human terms
You can think of replay = 1 as a “safety cushion for the last event.” It doesn’t turn events into “state” (we won’t go deep into that topic), but for demos and simple scenarios where “the last fact matters,” it makes behavior predictable: a subscriber connects and immediately sees the latest.
It’s important to decide what exactly you’re building: “pure events” with no memory, or “events where the latest value is important to show a new observer.” In learning code, the second option is often better because it’s more robust against coroutine start order.
4. Errors in collect handlers and “dying” subscriptions
When you wrote your own Event<T>, we wrapped each handler in try/catch so that one listener crashing wouldn’t break broadcasting for everyone. In SharedFlow, a similar problem shows up differently: if inside collect { ... } you throw an exception and don’t catch it, the collector coroutine will complete. And once the coroutine completes, the subscription is over.
That is, “one error in the handler” can turn into “the subscriber quietly died and never listens again.” This is especially unpleasant because from the outside it looks like “events stopped coming.”
Therefore, if the handler can fail, it makes sense to isolate errors right inside collect:
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect
fun main() = runBlocking {
val bus = EventBus()
val job = launch {
bus.events.collect { e ->
try {
if (e is TaskAdded && e.text == "BAD") error("broken handler")
println("OK: $e")
} catch (ex: Exception) {
println("handler error: ${ex.message}") // handler error: broken handler
}
}
}
bus.emit(TaskAdded("GOOD")) // OK: TaskAdded(text=GOOD)
bus.emit(TaskAdded("BAD")) // handler error: broken handler
bus.emit(TaskAdded("GOOD2"))// OK: TaskAdded(text=GOOD2)
job.cancel()
}
There’s a nice effect here: the subscription doesn’t die because of one problematic event. You’re essentially repeating the idea of “isolating handler errors,” just in a coroutine style.
5. Example: “commands → events → reactions”
So that SharedFlow doesn’t look like “magic for the sake of magic,” let’s embed it into a familiar real-world scenario. Suppose we have a list of tasks (MutableList<String>) and an add command. Inside the command we want to: add a task and publish an event. Separate subscribers will do auditing and show a user notification.
The process diagram might look like this:
flowchart LR
CLI[CLI command] --> Domain[add/remove logic]
Domain --> Bus[EventBus / SharedFlow]
Bus --> Audit[Audit subscriber]
Bus --> Ui[UI subscriber]
Domain functions publish an event and don’t know about subscribers
import kotlinx.coroutines.runBlocking
fun addTask(tasks: MutableList<String>, text: String, bus: EventBus) = runBlocking {
tasks += text
bus.emit(TaskAdded(text))
}
Yes, runBlocking looks a bit rough here (and in a real application you’d do this more carefully), but for a learning console example the idea is simple: the action happened — the event was published. The event source doesn’t know who is listening.
Subscribers connect in main and live as long as the app lives
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect
fun main() = runBlocking {
val tasks = mutableListOf<String>()
val bus = EventBus()
val auditJob = launch {
bus.events.collect { println("AUDIT: $it") }
}
val uiJob = launch {
bus.events.collect { println("UI: task list changed") }
}
addTask(tasks, "Learn SharedFlow", bus)
// AUDIT: TaskAdded(text=Learn SharedFlow)
// UI: task list changed
auditJob.cancel()
uiJob.cancel()
}
A funny (and realistic) moment: a UI subscriber in a console app is just println. But architecturally it’s the same technique as in large apps: the same fact “a task was added” can matter to different parts of the system.
6. Event<T> vs SharedFlow: when to choose what
Sometimes the right conclusion from this topic is not “SharedFlow is better,” but “these are different tools for different situations.” A homemade Event<T> is great for simple synchronous broadcasting where you want full control and minimal dependencies. SharedFlow is convenient when you already live in coroutines and want subscribe/unsubscribe via Job and a unified style with Flow.
Let’s compare them in a table (without diving into subtle settings):
| Criterion | |
|
|---|---|---|
| Subscription model | list of functions (T) -> Unit | coroutine + collect { ... } |
| Unsubscription | token, remove listener | |
| Listener error | catch try/catch around the call | if you don’t catch it, the collector coroutine completes |
| Storing the “last event” | you need to write it yourself | replay = 1 (and more) |
| Connection to coroutines | not required | native environment |
If somewhere it’s simpler for you to use “a plain list of listeners,” take Event<T>. If you’re already in coroutine code and want the stream model, take SharedFlow. Kotlin is good precisely because it doesn’t force you to pick one style forever.
7. Common mistakes when using SharedFlow as an event bus
Mistake #1: making MutableSharedFlow public.
When you expose MutableSharedFlow itself, any piece of code can start emitting events. As a result, events stop being “truth from the source” and turn into “anything goes.” It’s much more stable to keep MutableSharedFlow private and expose SharedFlow so external parts can only subscribe.
Mistake #2: forgetting that collect usually doesn’t finish on its own.
A beginner often writes launch { flow.collect { ... } } and expects the coroutine to “finish” after one event. But collect is a subscription, and it will keep waiting. If you don’t cancel the Job, the subscriber stays active. This is not a bug; it’s the contract. Just keep the lifecycle in mind.
Mistake #3: expecting past events with replay = 0.
If replay is 0, late subscribers are not obliged to receive past events — they will only see future ones. For learning examples and demos, it’s better to set replay = 1; otherwise you start depending on timing: whoever starts first wins.
Mistake #4: not catching exceptions inside the collect handler.
If the handler inside collect throws an exception, the collector coroutine may complete, and the subscription will “die.” Then you’ll spend a long time thinking about why events stopped coming. If the handler is potentially fragile, isolate errors via try/catch right inside collect, like we did earlier for individual listeners in Event<T>.
Mistake #5: confusing “events” and “state” and setting replay without understanding.
replay = 1 makes behavior more stable, but it’s already “a bit of memory.” If you mindlessly increase replay or start relying on it like “storage,” you can easily get strange effects: a new subscriber sees old events and reacts as if they just happened. In this lecture, replay = 1 is a deliberate learning setting, not a universal recommendation for all cases.
Mistake #6: thinking that SharedFlow automatically makes handlers safe from reentrancy.
In the manual event implementation we solved “modifying the listener list during broadcasting” via a snapshot (toList()), and that worked because we controlled the data structure. In SharedFlow you have a different model: handlers live in coroutines, and it’s important to watch what they do inside. If a handler publishes new events in response to an event, you can get a cascade. That’s not necessarily bad, but it should be a conscious design, not an accidental reaction like “oh, let’s emit here too.”
GO TO FULL VERSION