SharedFlow як шина подій: collect, replay і життєвий цикл підписки

1. Публікація та підписка в SharedFlow

Якщо ви щойно написали охайний Event<T> зі snapshot і токеном відписки, логічно виникає запитання: «А навіщо мені ще якийсь SharedFlow? Я ж щойно навів лад». І це справді слушне запитання: ваш Event<T> добре підходить для синхронної розсилки в межах звичайного коду.

Але щойно ми опиняємося у світі корутин, з’являється інший стиль роботи: підписники стають корутинами, життєвий цикл підписки зручніше контролювати через Job, а сама розсилка подій природно вкладається в модель потоків даних (Flow). У Kotlin є кілька способів обміну даними між корутинами, і SharedFlow якраз описують як механізм, що «ділиться кожним значенням з усіма активними колекторами».

Мета лекції — навчитися використовувати SharedFlow як «шину подій» у корутинному коді, розуміти, як улаштовані підписка й «відписка», а також чому параметр replay робить навчальні приклади стабільними й незалежними від того, «встигли / не встигли».

MutableSharedFlow і SharedFlow: «пульт» і «екран»

Коли ви вперше бачите MutableSharedFlow<T>, може виникнути спокуса видати його всім підряд: «хай хто хоче — той і emit-ить». Це приблизно як залишити кнопку «пуск ракети» на кухонному столі поруч із печивом. Формально зручно, але потім ви пів дня шукаєте, хто й чому «вистрілив» зайву подію.

У корутинному світі заведено розділяти ролі: MutableSharedFlow<T> — це точка публікації (туди «надсилають» події), а SharedFlow<T> — безпечний варіант «лише для читання» (звідти події тільки читають через collect). Цей підхід добре узгоджується із загальною ідеєю розподілу обов’язків і зменшення зв’язності: назовні ми даємо можливість підписатися, але не даємо змоги публікувати події звідки завгодно.

Зробімо мінімальну «шину подій» для нашого консольного застосунку (нехай це буде невеликий CLI-трекер завдань, де є команди додавання/видалення, а події потрібні для аудиту та сповіщень).

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)
}

Зверніть увагу на дві речі. По‑перше, _events — приватний і «мутабельний» (його можна змінювати: у нього можна emit). По‑друге, назовні доступний events: SharedFlow<AppEvent>: підписуйтеся скільки хочете, але напряму «публікувати» не можна.

Підписка через collect: «слухач» живе в корутині

У нашій саморобній системі подій слухач був звичайною функцією (T) -> Unit, яку ми складали в список. У SharedFlow слухач перетворюється на корутину-колектор: ви запускаєте launch { flow.collect { ... } }, і доки ця корутина жива, вона отримує події.

Це дуже зручно для керування життєвим циклом: «відписка» — це просто скасування Job. Жодних пошуків «тієї самої лямбди», жодних remove(listener). Підписка живе рівно стільки, скільки живе корутина, і це чудово узгоджується з ідеєю structured concurrency: дочірні задачі живуть усередині батьківського scope.

Мінімальний приклад «два підписники отримують усі події»:

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 отримав: $it") }
    }
    val b = launch {
        bus.events.collect { println("B отримав: $it") }
    }

    bus.emit(TaskAdded("Купити молоко")) // A отримав: TaskAdded(text=Купити молоко), B отримав: TaskAdded(text=Купити молоко)

    a.cancel() // відписалися
    b.cancel() // відписалися
}

Тут важлива думка: collect — це «підписка», яка зазвичай не завершується сама по собі. Якщо ви не скасуєте Job, колектор чекатиме на нові події. Для «event bus» це нормально: події потенційно нескінченні.

«Відписка» в SharedFlow: скасовуємо Job, а не шукаємо лямбду

У лекції про токени відписки ми підкреслювали «посилальну природу лямбд»: відписатися можна лише тим самим посиланням. У SharedFlow цей пласт проблем майже зникає, тому що підписка — це не «елемент у списку», а корутина, що виконується.

Тобто практичний рецепт відписки виглядає так: зберегли Job від launch, потім викликали cancel() (інколи ще join(), але сьогодні нам достатньо просто скасування). Це зручно ще й тим, що скасування корутини — звичний інструмент, який ви вже знаєте з попередніх тем про structured concurrency.

Дуже короткий приклад «підписалися — отримали одну подію — відписалися»:

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("Аудит: $it") }
    }

    bus.emit(TaskAdded("Прочитати документацію Kotlin")) // Аудит: TaskAdded(text=Прочитати документацію Kotlin)
    job.cancel() // підписку зупинено
}

Якщо вам потрібен «тимчасовий слухач» (наприклад, «вести журнал лише поки триває сценарій»), Job.cancel() — саме те, що треба.

3. replay і «пізні» підписники

Найпоширеніший навчальний біль із потоками подій звучить так: «Я зробив emit, але підписник нічого не отримав. Kotlin зламався?». Насправді ні: ви випадково продемонстрували класичний сценарій late subscriber — підписник почав слухати після того, як подія вже «полетіла».

За визначенням звичайний Flow продукує значення лише тоді, коли його збирають (collect), а SharedFlow роздає значення всім активним колекторам. Але якщо колекторів у момент події не було, то для SharedFlow з replay = 0 ця подія, по суті, «нікому не потрібна», і вона не зберігається.

Ось чому параметр replay такий корисний саме в навчальних прикладах: він дає змогу зберігати останні N подій і «перегравати» їх новим підписникам. У цій лекції ми використовуємо replay = 1, щоб новий підписник гарантовано побачив хоча б «останній факт», а приклад не залежав від таймінґів.

Подивімося на дві поведінки: спершу уявіть replay = 0, а потім — replay = 1. Ми залишимо наш EventBus з replay = 1, але покажемо, чому це важливо.

Пізня підписка: подія раніше, підписник пізніше

import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.collect

fun main() = runBlocking {
    val bus = EventBus()

    bus.emit(TaskAdded("EARLY")) // підписників немає, але replay=1 запамʼятає останнє

    val job = launch {
        bus.events.collect { println("Пізній отримав: $it") }
    }

    // Завдяки replay=1 підписник одразу отримає "EARLY"
    job.cancel()
}

Якби replay дорівнював 0, рядок Пізній отримав: ... міг би не з’явитися взагалі. І тоді студент (а інколи й викладач) почав би підозрювати змову таймера та планувальника.

Що робить replay = 1 у людських термінах

Можна сприймати replay = 1 як «подушку безпеки для останньої події». Це не перетворює події на «стан» (ми не заглиблюємося в цю тему), але для демонстрацій і простих сценаріїв, де «останній факт важливий», робить поведінку передбачуваною: підписник під’єднався — одразу побачив останнє.

Важливо домовитися із собою, що саме ви будуєте: «чисті події» без пам’яті чи «події, де останнє значення важливо показати новому спостерігачеві». У навчальному коді часто вигідніший другий варіант, бо він стійкіший до порядку запуску корутин.

4. Помилки в обробниках collect і підписки, що «помирають»

Коли ви писали свій Event<T>, ми обгортали кожен обробник у try/catch, щоб падіння одного слухача не ламало розсилку всім. У SharedFlow схожа проблема проявляється інакше: якщо всередині collect { ... } ви кинете виняток і не спіймаєте його, корутина-колектор завершиться. А раз корутина завершилася — підписка закінчилася.

Тобто «одна помилка в обробнику» може перетворитися на «підписник мовчки помер і більше ніколи не слухає». Це особливо неприємно, тому що ззовні здається, ніби «події перестали приходити».

Тому, якщо обробник може падати, варто ізолювати помилки прямо всередині 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("зламаний обробник")
                println("OK: $e")
            } catch (ex: Exception) {
                println("помилка обробника: ${ex.message}") // помилка обробника: зламаний обробник
            }
        }
    }

    bus.emit(TaskAdded("GOOD")) // OK: TaskAdded(text=GOOD)
    bus.emit(TaskAdded("BAD"))  // помилка обробника: зламаний обробник
    bus.emit(TaskAdded("GOOD2"))// OK: TaskAdded(text=GOOD2)

    job.cancel()
}

Тут є приємний ефект: підписка не «помирає» через одну проблемну подію. Ви ніби повторюєте ідею «ізоляції помилок обробників», але вже в корутинному стилі.

5. Приклад: «команди → події → реакції»

Щоб SharedFlow не виглядав як «магія заради магії», вбудуймо його в зрозумілий побутовий сценарій. Нехай у нас є список задач (MutableList<String>) і команда add. Усередині команди ми хочемо: додати задачу й опублікувати подію. Окремі підписники робитимуть аудит і показуватимуть користувачеві сповіщення.

Схема процесу може виглядати так:

flowchart LR
    CLI[Команда в CLI] --> Domain[Логіка add/remove]
    Domain --> Bus[EventBus / SharedFlow]
    Bus --> Audit[Audit subscriber]
    Bus --> Ui[UI subscriber]

Доменні функції публікують подію й не знають про підписників

import kotlinx.coroutines.runBlocking

fun addTask(tasks: MutableList<String>, text: String, bus: EventBus) = runBlocking {
    tasks += text
    bus.emit(TaskAdded(text))
}

Так, тут runBlocking виглядає грубувато (і в реальному застосунку ви б зробили це охайніше), але для навчального консольного прикладу ідея проста: дія відбулася — подію опубліковано. Джерело події не знає, хто слухає.

Підписники під’єднуються в main і живуть, доки живе застосунок

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("АУДИТ: $it") }
    }
    val uiJob = launch {
        bus.events.collect { println("UI: список задач змінився") }
    }

    addTask(tasks, "Learn SharedFlow", bus)
    // АУДИТ: TaskAdded(text=Learn SharedFlow)
    // UI: список задач змінився

    auditJob.cancel()
    uiJob.cancel()
}

Кумедний (і життєвий) момент: UI‑підписник у консолі — це просто println. Але з погляду архітектури це той самий прийом, що й у великих застосунках: один і той самий факт «задачу додано» може бути важливим для різних частин системи.

6. Event<T> vs SharedFlow: коли що обирати

Іноді правильний висновок із цієї теми звучить не як «SharedFlow краще», а як «це різні інструменти для різних ситуацій». Саморобний Event<T> чудово підходить для простої синхронної розсилки, коли ви хочете повний контроль і мінімум залежностей. SharedFlow зручний, коли ви вже працюєте з корутинами й хочете підписки/відписки через Job та єдиний стиль із Flow.

Порівняймо в таблиці (без занурення в тонкі налаштування):

Event<T>

SharedFlow<T>

job.cancel()

Якщо десь вам простіше «звичайний список слухачів», беріть Event<T>. Якщо ви вже в корутинному коді й хочете потокову модель, беріть SharedFlow. Kotlin якраз і добрий тим, що не змушує обирати один стиль назавжди.

7. Типові помилки під час роботи з SharedFlow як із шиною подій

Помилка №1: робити MutableSharedFlow публічним.
Коли назовні «стирчить» саме MutableSharedFlow, будь-який фрагмент коду може почати «емітити» події. У результаті події перестають бути «істиною від джерела», а стають принципом «хто на що здатен». Набагато надійніше тримати MutableSharedFlow приватним, а назовні віддавати SharedFlow, щоб зовнішні частини могли лише підписуватися.

Помилка №2: забувати, що collect зазвичай не завершується сам.
Новачок часто пише launch { flow.collect { ... } } і чекає, що корутина «закінчиться» після однієї події. Але collect — це підписка, і вона чекатиме далі. Якщо ви не скасуєте Job, підписник залишиться активним. Це не баг — це контракт. Просто тримайте в голові життєвий цикл.

Помилка №3: очікувати минулі події за replay = 0.
Якщо replay дорівнює 0, пізні підписники не зобов’язані отримувати минулі події — вони побачать лише майбутні. Для навчальних прикладів і демонстрацій краще ставити replay = 1, інакше ви починаєте залежати від таймінґів: у кого підписка запустилася першою, той і отримав подію.

Помилка №4: не ловити винятки всередині обробника collect.
Якщо обробник усередині collect кидає виняток, корутина-колектор може завершитися, і підписка «помре». Потім ви довго думатимете, чому події перестали приходити. Якщо обробник потенційно «крихкий», ізолюйте помилки через try/catch прямо всередині collect, як ми робили раніше для окремих слухачів у Event<T>.

Помилка №5: плутати «події» і «стан» та ставити replay без розуміння.
replay = 1 робить поведінку стабільнішою, але це вже «трошки пам’ять». Якщо ви бездумно збільшите replay або почнете покладатися на нього як на «сховище», легко отримати дивні ефекти: новий підписник бачить старі події й реагує так, ніби вони щойно сталися. У цій лекції replay = 1 — усвідомлене навчальне налаштування, а не універсальна рекомендація на всі випадки.

Помилка №6: думати, що SharedFlow автоматично робить обробники безпечними щодо реентерабельності.
У ручній реалізації подій ми розв’язували проблему «модифікація списку слухачів під час розсилки» через snapshot (toList()), і це працювало, бо ми контролювали структуру даних. У SharedFlow вже інша модель: обробники живуть у корутинах, і важливо стежити за тим, що вони роблять усередині. Якщо обробник у відповідь на подію запускає нову публікацію, ви можете отримати каскад. Це не обов’язково погано, але має бути усвідомленим дизайном, а не випадковою реакцією «ой, давайте тут ще emit зробимо».

1

Опитування

Події/Observer, рівень 57, лекція 4

Недоступний

Події/Observer

Події/Observer

Відкрити