Надійне розсилання подій: snapshot, try/catch, реентерабельність

1. Чому наївне розсилання ламається в реальній програмі

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

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

class Event<T> {
    private val listeners = mutableListOf<(T) -> Unit>()

    fun subscribe(listener: (T) -> Unit): () -> Unit {
        listeners += listener
        return { listeners.remove(listener) }
    }

    fun emit(value: T) {
        for (l in listeners) l(value)
    }
}

На перший погляд усе виглядає добре. Але тепер додамо два «звичайні» сценарії з реального життя.

Сценарій: відписатися всередині обробника

Нехай обробник «одноразовий»: після першої події він знімає підписку.

fun main() {
    val e = Event<Int>()

    var unsubscribe: (() -> Unit)? = null
    unsubscribe = e.subscribe { x ->
        println("одноразовий отримав $x")   // одноразовий отримав 1
        unsubscribe?.invoke()
    }

    e.emit(1)
    e.emit(2)
}

Тут навіть без заглиблення видно: під час emit список слухачів змінюється. На JVM такі речі часто закінчуються неприємно (на кшталт ConcurrentModificationException) або «тихими» логічними багами: когось пропустили, когось викликали зайвий раз.

Сценарій: один обробник падає, інші мають вижити

Припустімо, у вас є журнал аудиту й сповіщення. Сповіщення впали (наприклад, усередині сталася помилка форматування), але аудит має записатися завжди.

Якщо emit не захищений, виняток перериває цикл, і решта обробників не викличеться. А це вже не «ой, неприємно», а «чому в нас звіти не збігаються?».

2. Snapshot слухачів: копія списку перед розсиланням

Якщо ви коли-небудь фотографували кота, який носиться кімнатою, то вже розумієте ідею snapshot: ви робите знімок «ось так воно було в цей момент», і далі кіт може бігати як завгодно — фото не зміниться. У розсиланні подій snapshot потрібен рівно для цього: ми фіксуємо, який набір слухачів отримає поточну подію, а зміни підписок під час розсилання впливають лише на майбутні emit.

Ключовий інструмент Kotlin тут — toList(). Він створює копію списку елементів у конкретний момент часу; подальші зміни вихідної MutableList не впливають на цю копію.

Поганий обхід: «живий» список

fun emitUnsafe(value: T) {
    for (l in listeners) l(value)
}

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

Хороший обхід: snapshot першим кроком

fun emit(value: T) {
    val snapshot = listeners.toList()
    for (l in snapshot) l(value)
}

І тут дуже важливе правило (майже як «спочатку вдягніть маску на себе»): snapshot має бути першою дією в emit. Якщо ви почнете розсилання, а потім раптом вирішите зробити snapshot «посередині», отримаєте поведінку, яку важко пояснити без шаманського бубна.

Контракт snapshot-підходу

Із snapshot-розсиланням ми ніби підписуємо «договір» із користувачами нашого Event<T>.

• Якщо обробник підпишеться під час emit, він почне отримувати події лише з наступного emit.
• Якщо обробник відпишеться під час emit, він усе одно може отримати поточну подію (бо вже потрапив у snapshot), але не отримає наступних.

Цей контракт передбачуваний, простий і легко пояснюється в одному абзаці. У навчальних застосунках це майже завжди найкращий вибір.

3. try/catch навколо кожного обробника

З винятками в обробниках є тонкий момент: ми хочемо, щоб розсилання було стійким, але не хочемо перетворювати помилки на «чорну діру», у якій нічого не видно й ніхто не винен. Тому правильна стратегія зазвичай така: перехоплюємо виняток для кожного слухача окремо, фіксуємо факт помилки (хоча б через println) і продовжуємо розсилання.

Чому саме так? Бо try/catch навколо всього розсилання зловить помилку, але зупинить цикл. Тобто один слухач зламає інших. А якщо ми обгортаємо кожного окремо, то помилка буде ізольованою.

Неправильний захист: один try/catch на все

fun emit(value: T) {
    try {
        val snapshot = listeners.toList()
        for (l in snapshot) l(value)
    } catch (e: Exception) {
        println("Розсилання зірвалося: ${e.message}")
    }
}

Виглядає «захищено», але на практиці, якщо перший обробник упав, другий і третій уже не будуть викликані.

Правильний захист: try/catch на кожного

fun emit(value: T) {
    val snapshot = listeners.toList()
    for (l in snapshot) {
        try {
            l(value)
        } catch (e: Exception) {
            println("Обробник упав: ${e.message}")
        }
    }
}

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

4. Реентерабельність: обробник може викликати emit знову

Слово «реентерабельність» звучить так, ніби зараз ми будемо будувати космічний корабель. Насправді все простіше: під час обробки події може статися так, що ми знову викличемо emit — напряму або опосередковано.

Наприклад, ви додали витрату → обробник вирішив, що сума надто велика → він автоматично додає «комісію» окремою витратою → додавання комісії теж викликає подію «витрату додано». Бум: вкладене розсилання.

Нам важливо, щоб від такого сценарію програма не перетворювалася на хаос. Snapshot-підхід уже дає основу: кожен emit бере свою копію слухачів. Але варто чітко розуміти, які ефекти можливі, а які пастки все ще лишаються.

Приклад реентерабельності: вкладений emit

fun main() {
    val e = Event<Int>()

    e.subscribe { x ->
        if (x == 1) e.emit(2)
        println("A отримав $x")
    }
    e.subscribe { x -> println("B отримав $x") }

    e.emit(1)
}

Один із можливих виводів буде таким (порядок рядків залежить від того, де ви друкуєте):

A отримав 2
B отримав 2
A отримав 1
B отримав 1

І це нормально: ви справді запустили emit(2) усередині emit(1). Тобто порядок стає «вкладеним», як виклики функцій. Саме це й означає «синхронне розсилання».

Що snapshot гарантує за реентерабельності

Snapshot гарантує, що навіть якщо всередині emit(1) обробник підпишеться або відпишеться, це не зламає поточний обхід. Кожен emit працює зі своєю копією.

Чого snapshot не розвʼязує

Snapshot не рятує вас від нескінченної рекурсії. Якщо обробник для будь-якого значення викликає emit знову без зупинки, ви отримаєте переповнення стека. Це не баг Event<T>, а логіка обробника.

Тому в межах сьогоднішньої лекції зафіксуємо простий і зрозумілий контракт: emit синхронний і допускає вкладені виклики, а відповідальність «не піти в нескінченність» лежить на обробниках.

5. Збираємо надійний Event<T>

Давайте тепер акуратно зберемо підсумкову версію Event<T>, яку можна використовувати в навчальному застосунку. Продовжимо ідею консольного трекера витрат: ви пишете «кава 4.5», а застосунок друкує «у вас знову кава, тримайтеся».

Модель події: «витрату додано»

Для простоти зробимо невелику модель:

data class ExpenseAdded(
    val id: Int,
    val title: String,
    val amount: Double
)

Підсумковий Event<T>

Зробимо subscribe так, щоб він повертав токен відписки. Причому повторна відписка буде безпечною (не падає і не намагається «видалити двічі»).

class Event<T> {
    private val listeners = mutableListOf<(T) -> Unit>()

    fun subscribe(listener: (T) -> Unit): () -> Unit {
        listeners += listener
        var active = true
        return {
            if (active) {
                listeners.remove(listener)
                active = false
            }
        }
    }

    fun emit(value: T) {
        val snapshot = listeners.toList()
        for (l in snapshot) {
            try {
                l(value)
            } catch (e: Exception) {
                println("Обробник упав: ${e.message}")
            }
        }
    }
}

Тут snapshot робиться через toList(), і саме це дає нам «знімок» слухачів на момент початку розсилання.

try/catch стоїть навколо кожного слухача, щоб один збій не обривав розсилання для всіх інших.

Вбудовуємо подію в «мінітрекер»

Нехай у нас є сервіс, який додає витрату й емітить подію:

class ExpenseService {
    private var nextId = 1
    val expenseAdded = Event<ExpenseAdded>()

    fun addExpense(title: String, amount: Double) {
        val e = ExpenseAdded(nextId++, title, amount)
        expenseAdded.emit(e)
    }
}

Два незалежні слухачі: аудит і «сповіщення»

fun main() {
    val service = ExpenseService()

    service.expenseAdded.subscribe { e ->
        println("AUDIT: додано #${e.id} '${e.title}'") // AUDIT: додано #1 'Coffee'
    }

    service.expenseAdded.subscribe { e ->
        if (e.amount > 1000) error("Надто велике!")     // може впасти
        println("UI: +${e.amount} за ${e.title}")       // UI: +4.5 за Coffee
    }

    service.addExpense("Coffee", 4.5)
    service.addExpense("Laptop", 2000.0)
    service.addExpense("Sandwich", 6.0)
}

Очікувана логіка така: «Laptop» зібʼє другий обробник, але аудит має продовжувати працювати, і третя витрата теж має обробитися.

Приклад можливого виводу:

AUDIT: додано #1 'Coffee'
UI: +4.5 за Coffee
AUDIT: додано #2 'Laptop'
Обробник упав: Надто велике!
AUDIT: додано #3 'Sandwich'
UI: +6.0 за Sandwich

Зверніть увагу на поведінку: виняток не «приклеївся» до всього emit, він ізольований усередині одного слухача.

Схема: що відбувається всередині emit

Іноді корисно бачити не лише код, а й «картинку процесу». Уявімо розсилання як конвеєр:

flowchart TD
    A["emit(value) викликано"] --> B["snapshot = listeners.toList()"]
    B --> C["for listener in snapshot"]
    C --> D["try { listener(value) }"]
    D --> E["успіх → наступний listener"]
    D --> F["catch Exception → логуємо помилку"]
    F --> E
    E --> G["кінець розсилки"]

Сенс діаграми простий: спочатку ми фіксуємо список слухачів, а потім проходимо по копії, ізолюючи помилки кожного обробника.

6. Типові помилки

Помилка № 1: робити snapshot не на початку emit.
Іноді трапляється підхід «спочатку викличу пару слухачів, потім зроблю копію». Це майже гарантовано призведе до дивної поведінки: частина слухачів буде викликана за старим списком, частина — за новим. Якщо ви обираєте snapshot-стратегію, то знімок має бути першим кроком у розсиланні. Інакше ви втрачаєте головний плюс — передбачуваність.

Помилка № 2: обгортати весь emit одним try/catch.
Такий код виглядає «надійно», але він захищає не те, що потрібно. У разі винятку перший обробник, який упав, перерве цикл, і решта слухачів не отримають подію. Правильний принцип тут — ізоляція: try/catch має стояти навколо кожного обробника, щоб один збій не ламав увесь процес.

Помилка № 3: «ковтати» винятки мовчки.
Якщо ви перехоплюєте виняток і нічого не робите (навіть не друкуєте повідомлення), ви створюєте собі дуже дорогу загадку на майбутнє: «чому один із модулів перестав реагувати?». У навчальному коді достатньо println, у більш серйозному — варто хоча б фіксувати message і, за можливості, стек. Головне — не перетворювати помилки на невидимок.

Помилка № 4: очікувати, що відписка всередині emit миттєво скасує виклик у поточному розсиланні.
За snapshot-підходу це не так: поточне розсилання вже йде по копії. Тому обробник може отримати подію «ще раз» у межах поточного emit, навіть якщо він щойно відписався. Це не баг, а частина контракту. Плюс у тому, що контракт простий і стабільний: зміни підписок впливають на наступні розсилання.

Помилка № 5: випадково влаштувати нескінченну реентерабельність.
Якщо обробник на будь-яку подію викликає emit знову без умови зупинки, ви отримаєте рекурсію і зрештою — переповнення стека. Snapshot не зобовʼязаний (і не може) це виправляти: це логічна помилка обробника. Тому для реентерабельних сценаріїв завжди тримайте в голові «умову виходу»: перевірку значення, прапорець, ліміт або хоча б здоровий глузд.