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("one-time got $x") // one-time got 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("Emit failed: ${e.message}")
}
}
Выглядит «защищённо», но на деле: если первый обработчик упал, второй и третий уже не будут вызваны.
Правильная защита: try/catch на каждого
fun emit(value: T) {
val snapshot = listeners.toList()
for (l in snapshot) {
try {
l(value)
} catch (e: Exception) {
println("Listener failed: ${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 got $x")
}
e.subscribe { x -> println("B got $x") }
e.emit(1)
}
Один из возможных выводов будет таким (порядок строк зависит от того, где вы печатаете):
A got 2
B got 2
A got 1
B got 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("Listener failed: ${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: added #${e.id} '${e.title}'") // AUDIT: added #1 'Coffee'
}
service.expenseAdded.subscribe { e ->
if (e.amount > 1000) error("Too big!") // может упасть
println("UI: +${e.amount} for ${e.title}") // UI: +4.5 for Coffee
}
service.addExpense("Coffee", 4.5)
service.addExpense("Laptop", 2000.0)
service.addExpense("Sandwich", 6.0)
}
Ожидаемая логика такая: «Laptop» уронит второй обработчик, но аудит должен продолжать работать, и третий расход тоже должен обработаться.
Пример возможного вывода:
AUDIT: added #1 'Coffee'
UI: +4.5 for Coffee
AUDIT: added #2 'Laptop'
Listener failed: Too big!
AUDIT: added #3 'Sandwich'
UI: +6.0 for 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 не обязан (и не может) это чинить: это логическая ошибка обработчика. Поэтому для реентерабельных сценариев всегда держите в голове «условие выхода»: проверку значения, флаг, лимит, или хотя бы здравый смысл.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ