Kotlin Flow

Flow описывает асинхронный поток значений, построенный на suspend-функциях. Холодный flow ничего не делает до вызова collect и заново запускает продюсер для каждого подписчика; горячий flow (StateFlow, SharedFlow, Channel) эмитит независимо от того, слушает ли его кто-нибудь. Холодный: flow { }, Room-запрос @Query, возвращающий Flow. Два подписчика дают два запроса в базу или два сетевых вызова. Горячий: StateFlow хранит текущее значение, SharedFlow рассылает всем, кто подписан прямо сейчас. Поздний подписчик пропускает прежние значения, если flow их не реплеит. **Flow и Channel**: Channel работает как горячая очередь между корутинами: каждый элемент получает ровно один потребитель. Flow представляет собой декларативный тип потока с операторами; сырой Channel из ViewModel наружу почти не отдают. Из холодного в горячий: stateIn и shareIn собирают холодный flow один раз и раздают его многим подписчикам.

Оба выдают значения лениво, по одному, но Sequence работает синхронно: потребитель вытягивает каждое значение через next(), а его поток выполняет продюсер до следующего yield. Flow построен на suspend-функциях, поэтому его продюсер может ждать таймер или сеть, не блокируя поток. **Почему delay там не компилируется**: блок sequence { } выполняется на SequenceScope, помеченном @RestrictsSuspension, поэтому в нём можно вызывать только yield и yieldAll. Всё остальное падает с ошибкой «Restricted suspending functions can invoke member or extension suspending functions only on their restricted coroutine scope». Зачем такое правило: next() является обычной функцией, поэтому у вызывающего нет корутины, которую можно приостановить; подождать внутри можно было бы, только заблокировав его поток. Что есть только у Flow: отмена вместе с собирающей корутиной, смена диспетчера через flowOn и операторы по времени вроде debounce. Что выбрать: последовательность для работы с данными в памяти на CPU (ленивый пайплайн над большим списком, генератор); flow для всего, что чего-то ждёт. asFlow() превращает последовательность во flow.

Терминальные операторы запускают сбор: collect, first, toList, fold и подобные являются suspend-функциями, а launchIn собирает flow в новой корутине. map, filter и onEach лишь оборачивают flow в новый, поэтому цепочка, которая заканчивается на onEach { }, ничего не выполняет. Холодный, пока не собран: промежуточный оператор возвращается сразу, и код продюсера не запускается. Каждый терминальный вызов заново прогоняет всю цепочку от источника. Как их различить: промежуточные операторы являются обычными функциями, которые возвращают Flow, и их можно вызвать где угодно; терминальные помечены suspend (или, как launchIn, принимают scope), поэтому им нужна корутина. Ранний выход: first() берёт одно значение и отменяет апстрим. На flow, который завершился пустым, он бросает NoSuchElementException, а firstOrNull() возвращает null. Ожидание состояния: session.first { it.isReady } приостанавливается, пока StateFlow не получит подходящее значение, и возвращается сразу, если оно уже там; оберните вызов в withTimeoutOrNull(5_000), чтобы перестать ждать. **launchIn(scope)**: сокращение для scope.launch { collect() }. Он возвращает Job, отмена которого останавливает сбор. onEach и catch ставят перед ним, потому что после него ничего быть не может.

StateFlow нужен для стейта экрана: он всегда хранит текущее значение, требует начальное и пропускает равные значения. SharedFlow даёт настраиваемую рассылку без текущего значения, для событий, которые должен увидеть каждый подписчик. **StateFlow**: .value читается синхронно; медленный подписчик видит только последнее значение; равные значения отбрасываются, поэтому обновляйте копиями, а не изменяйте список на месте. **SharedFlow**: MutableSharedFlow(replay, extraBufferCapacity, onBufferOverflow). При replay = 0 значение, отправленное, пока никто не собирает, теряется. Связь между ними: StateFlow ведёт себя как SharedFlow с replay = 1, DROP_OLDEST, начальным значением и distinctUntilChanged. Оба горячие: их сбор сам по себе никогда не завершается. Наружу отдают только read-only типы: приватный MutableStateFlow, публичный asStateFlow(). Одноразовые события (навигация, snackbar): SharedFlow теряет их, когда UI не собирает; многие команды моделируют их как стейт или используют Channel с receiveAsFlow() для одного подписчика.

StateFlow отбрасывает новое значение, равное текущему, а _cart.value = items присваивает тот же самый список, который уже в нём лежит. Ничего не эмитится, поэтому экран продолжает показывать старую корзину. Почему: перед эмитом MutableStateFlow сравнивает новое значение с текущим через equals. Здесь это один и тот же объект, поэтому они всегда равны, как бы ни изменился список внутри. Исправление: храните во flow неизменяемый List и заменяйте его целиком: _cart.update { it + item } создаёт новый список, а update работает как compare-and-set, поэтому два одновременных вызова add не потеряют товар. Та же причина, второй симптом: подписчик держит тот же MutableList, который меняет ViewModel, поэтому DiffUtil или список в Compose, читающий его во время add, может увидеть наполовину изменённый список или упасть с ConcurrentModificationException. Как поймать в следующий раз: не кладите в StateFlow MutableList или data class с полями var; обновляйте состояние через copy. Тест ViewModel с Turbine (cart.test { vm.add(x); awaitItem() }) упадёт на пропущенном эмите.

collect на StateFlow никогда не возвращается, потому что горячий flow не завершается, и корутина застревает на первой строке: до второго collect дело не доходит. Запускайте каждый сбор в своём launch внутри блока repeatOnLifecycle. Почему это выглядит нормально: collect является обычным suspend-вызовом, и код после него выполнится, только когда flow завершится. StateFlow.collect, унаследованный от SharedFlow, даже объявлен с типом возврата Nothing, поэтому компилятор помечает вторую строку предупреждением «Unreachable code»: именно его и стоит искать. Почему иногда работает: холодный flow, который завершается, например одноразовый flow { } или flowOf в тестовом фейке, даёт выполниться следующей строке, и баг может прятаться, пока не подключат настоящий flow репозитория. Исправление: два дочерних launch работают параллельно; repeatOnLifecycle отменяет оба на onStop и снова запускает оба на onStart. Ещё лучше: если экрану всегда нужны оба значения, объедините их через combine в один UiState во ViewModel и собирайте один flow. В Compose у каждого вызова collectAsStateWithLifecycle() свой сборщик, поэтому там эта ловушка не возникает. Та же причина в других местах: toList() или count() на StateFlow зависают навсегда, а код после collect в LaunchedEffect не выполняется никогда.

Channel представляет собой горячую очередь между корутинами: send кладёт элемент, receive забирает его, и каждый элемент достаётся ровно одному получателю. Flow является холодным описанием, которое запускается заново для каждого сборщика, а канал существует сам по себе, даже когда никто не получает. **RENDEZVOUS** (по умолчанию для Channel(), ёмкость 0): send приостанавливается, пока получатель не заберёт элемент: передача из рук в руки. trySend не срабатывает, если получатель не ждёт. **BUFFERED**: буфер на 64 элемента по умолчанию; send приостанавливается, только когда он полон. **CONFLATED**: хранит один элемент; новый send заменяет непрочитанный и никогда не приостанавливается. Для случаев «важно только последнее». **UNLIMITED**: send никогда не приостанавливается, поэтому при медленном получателе память растёт без ограничений. Распределение, а не рассылка: две корутины, читающие один канал, делят элементы между собой; два сборщика SharedFlow получают каждое значение. Закрытие: после close() цикл for (x in channel) завершается, когда вычитает остаток; receive() на закрытом и уже опустевшем канале бросает ClosedReceiveChannelException. Канал, который никто не закрыл, держит получателей приостановленными навсегда. Где он уместен: очередь задач с одним или несколькими потребителями или одноразовые UI-события через receiveAsFlow(); consumeAsFlow() можно собрать только один раз. Сам Flow использует каналы внутри buffer, flowOn, channelFlow и callbackFlow.

flowOn меняет контекст только для операторов выше себя, до предыдущего flowOn. Поэтому db.readAll() и map (1) выполняются на IO, filter (2) на Default, а collect (3) на Main, где была запущена собирающая корутина. Только апстрим: контекст сборщика никогда не меняется. Поэтому во Flow нет observeOn: поток даунстрима определяется просто тем местом, где вызван collect. Как это устроено: на границе диспетчеров flowOn запускает апстрим в отдельной корутине и передаёт значения через канал, поэтому заодно добавляет буфер (по умолчанию 64). **Никакого withContext вокруг emit**: билдер flow { } обязан эмитить из корутины сборщика. Эмит изнутри withContext падает с IllegalStateException: Flow invariant is violated. Вызвать withContext для вычисления и эмитить снаружи можно. Несколько продюсеров: channelFlow { } разрешает send из других корутин и контекстов.

emit вызывается из корутин, запущенных через launch, а не из корутины, которая собирает flow, и flow { } это запрещает: «Emission from another coroutine is detected». Используйте channelFlow { }: его send можно вызывать из любого числа дочерних корутин. Зачем это правило: во flow { } вызов emit напрямую выполняет код сборщика. Две корутины, эмитящие одновременно, возможно на двух потоках, запустили бы даунстрим конкурентно, а FlowCollector не потокобезопасен. Проверка в рантайме сообщает именно это и предлагает channelFlow. **Как это чинит channelFlow**: продюсеры делают send в канал, а сборщик читает из него в своей корутине. Flow завершается, когда закончились блок и все его дочерние корутины, а ошибка в одной из них отменяет другую. Порядок: результаты приходят в том порядке, в каком завершились страницы, поэтому вторая может прийти первой. Если порядок важен, запустите обе через async и эмитьте результаты по очереди из собственной корутины flow; emit прямо внутри coroutineScope { } разрешён, а внутри launch или async нет. Другие варианты: merge(flow { emit(api.page(1)) }, flow { emit(api.page(2)) }) тоже запускает оба конкурентно. callbackFlow устроен так же, на канале, и предназначен для API со слушателями.

zip выводит 1A 2B: он составляет пары по позиции и останавливается, как только заканчивается один из flow, поэтому 3 остаётся без пары. combine выводит 1A 2A 2B 3B: когда у обоих flow уже есть значение, он выдаёт последнюю пару при каждом эмите любого из них. **zip: каждая пара ждёт по одному новому значению с каждой стороны, поэтому быстрый flow замедляется до темпа медленного, а лишние значения более длинного отбрасываются. Он подходит для значений, связанных по позиции, а в UI-коде такое бывает редко. combine**: молчит, пока каждый источник не выдаст хотя бы одно значение (здесь на 100 мс ничего нет), а потом пересчитывает результат на каждый эмит любого источника. Он подходит для стейта экрана из нескольких источников, например товаров и текущего фильтра. Если источник может молчать, дайте ему стартовое значение через onStart, иначе экран так и не выйдет из начального состояния. Завершение: zip завершается, когда завершается любой из flow, и отменяет второй; combine завершается, когда завершились все. **merge**: для flow одного типа, когда нужно каждое значение в момент прихода, без пар: локальные правки и push-сообщения в один обработчик синхронизации, несколько источников кликов в один поток событий.

flatMapLatest: когда приходит новый запрос, он отменяет запрос для старого, и устаревшие результаты не могут перезаписать свежие. Вокруг него debounce ждёт паузы в наборе, а distinctUntilChanged отбрасывает повторный запрос. **flatMapConcat: ждёт завершения каждого внутреннего flow по порядку, так что поиск по каждому нажатию клавиши выполняется один за другим. flatMapMerge: запускает внутренние flow конкурентно и перемешивает их результаты; медленный старый ответ может прийти после нового. flatMapLatest / mapLatest: отменяют предыдущий внутренний flow или преобразование, когда приходит новое значение. collectLatest**: та же идея в конце цепочки: отменяет тело сборщика при новом значении, тогда как обычный collect обрабатывает каждое значение по очереди. Пайплайн: debounce(300), затем filter для слишком коротких запросов, distinctUntilChanged(), flatMapLatest и stateIn для экрана. Ошибки: ставьте catch внутри внутреннего flow. catch после flatMapLatest превращает первую ошибку в значение и завершает весь поток поиска.

debounce(300) выдаёт значение, только если за 300 мс не пришло более нового, sample(300) раз в 300 мс выдаёт самое свежее значение, а throttle-first выдаёт первое значение и игнорирует остальные в течение окна. Строка поиска: debounce. Датчик: sample. Кнопка «Оплатить»: throttle-first. **debounce**: побеждает последнее значение перед паузой, что и означает «пользователь перестал печатать». Источник без пауз не выдаёт ничего: датчик на 60 Гц через debounce(300) молчит. **sample: снимок с фиксированной частотой, который отбрасывает значения между тиками. Он подходит для перерисовки по быстрому источнику (датчик, прогресс загрузки) с частотой, которую UI потянет; последнее значение теряется, если flow завершился раньше следующего тика. Throttle-first**: первое нажатие срабатывает сразу, а дубли внутри окна отбрасываются. debounce на кнопке задержал бы каждую оплату на длину окна, а два нажатия с интервалом 400 мс всё равно оплатили бы дважды. В kotlinx.coroutines нет throttleFirst; он пишется в несколько строк через flow { }. Не только задача flow: ещё блокируйте кнопку, пока идёт запрос, и делайте оплату идемпотентной на сервере, чтобы проскочившее нажатие не списало деньги дважды. Детали: debounce и sample помечены @FlowPreview, поэтому без @OptIn(FlowPreview::class) компилятор выдаёт предупреждение. В RxJava у sample есть второе имя throttleLast, а throttle-first называется throttleFirst.

По умолчанию ничего не теряется: emit приостанавливается, пока сборщик не обработает предыдущее значение, и продюсер замедляется до его темпа. buffer даёт им работать одновременно, conflate оставляет только новейшее значение, collectLatest отменяет медленный обработчик при новом. **buffer(capacity, onBufferOverflow)**: продюсер и сборщик работают в разных корутинах с каналом между ними, и общее время равно более медленной стороне, а не сумме. Политика переполнения: SUSPEND, DROP_OLDEST или DROP_LATEST. **conflate(): пропускает промежуточные значения. Подходит для прогресса, датчиков и геолокации; не подходит для событий, где важно каждое. collectLatest { }: перезапускает тело на каждое новое значение; хорошо, когда важна только последняя отрисовка или сохранение. Встроенные политики**: StateFlow сливает значения (conflation) по своей природе; MutableSharedFlow принимает extraBufferCapacity и политику переполнения. Побочный эффект: flowOn добавляет буфер на границе диспетчеров.

catch обрабатывает только исключения апстрима (операторов выше себя), а исключение в collect { } находится ниже и уходит прямо в собирающую корутину. Стройте цепочку так: источник, retryWhen, map, catch, а рискованную работу сборщика переносите в onEach выше catch. Прозрачность исключений (exception transparency): flow никогда не должен ловить исключения даунстрима, поэтому try/catch вокруг emit внутри билдера flow { } нарушает контракт и может упасть с IllegalStateException. **retry / retryWhen**: заново собирают апстрим. retry(3) { it is IOException } служит короткой формой; retryWhen { cause, attempt -> } с delay внутри даёт экспоненциальную задержку. **catch { emit(fallback) }: превращает ошибку в значение, но flow после этого всё равно завершается. Долгоживущий поток вроде поиска должен ловить ошибки внутри каждого вложенного flow, чтобы оставаться живым. Порядок важен**: catch выше retryWhen не оставляет ничего для повтора, потому что ошибка уже стала значением. **onCompletion { cause -> }**: работает как finally: видит причину и пропускает её дальше, но не обрабатывает.

onStart выполняется до запуска апстрима и может выдать первое значение, onEmpty срабатывает, когда апстрим завершился, не выдав ничего, а onCompletion выполняется после окончания flow по любой причине: успех, исключение или отмена. Флаг загрузки, поднятый в onStart, сбрасывают в onCompletion. **onStart**: срабатывает при каждом сборе, поэтому каждый retry, стоящий после него, и каждый перезапуск шаринга с WhileSubscribed запускают его заново, и спиннер появляется снова. **onCompletion { cause -> }**: cause равен null при успехе, исключению при ошибке и CancellationException при отмене. Он видит и ошибки из даунстрима, но не обрабатывает их: как finally, он пропускает исключение дальше, к catch или к сборщику. Почему не в другом месте: сброс флага в onEach или после последнего значения пропускает путь ошибки, а сброс в catch пропускает отмену, которая ошибкой не считается: синхронизация, которую пользователь остановил кнопкой «Отмена», оставит спиннер на экране. Бесконечные flow: Room-запрос или сокет никогда не завершаются, поэтому их onCompletion срабатывает только при отмене. Там сбрасывайте флаг на первом значении или моделируйте загрузку как стейт, который заменит следующее значение. **onEmpty { emit(NothingFound) }**: срабатывает при нуле эмитов, а не при пустом списке. Room-запрос, вернувший emptyList(), эмит сделал, поэтому пустой список проверяйте в map.

Возвращайте Flow<T>, когда данные могут меняться, пока на них смотрят, и вызывающий должен видеть каждое изменение: Room-запрос, настройка в DataStore, сокет. Делайте suspend fun для разовой операции с одним результатом: сетевой запрос, запись, команда. Flow для наблюдения: Room-запрос, возвращающий Flow, эмитит заново после каждой записи в его таблицы, поэтому экран, наблюдающий за базой, обновляется сам после любого обновления или правки, без логики перезапроса. Так и работает паттерн single source of truth. suspend для одного результата: suspend fun placeOrder(cart: Cart): OrderId читается как вызов, падает исключением, которое ловит вызывающий, и выполняется ровно один раз. Завёрнутый в flow { emit(api.order(cart)) }, он выполняется заново для каждого сборщика и скрывает, что значение будет только одно. Обычная пара: fun observeArticles(): Flow<List<Article>> плюс suspend fun refresh(), который берёт данные из сети и пишет их в Room; новые строки приходят на экран через flow. **Никаких suspend fun, возвращающих Flow**: создание холодного flow ничего не делает, поэтому suspend там лишь зря заставляет вызывающих заходить в корутину; приостанавливающуюся работу кладите внутрь flow. UI-состояния не в репозитории: Loading и баннеры ошибок относятся к ViewModel (onStart, catch); репозиторий, который эмитит Resource.Loading, навязывает всем вызывающим представление одного экрана о загрузке.

stateIn и shareIn один раз собирают холодный апстрим и раздают его многим подписчикам; SharingStarted решает, когда этот сбор начинается и заканчивается. С WhileSubscribed(5_000) апстрим переживает поворот экрана и останавливается через 5 секунд после ухода последнего подписчика. Три стратегии Eagerly: подписывается сразу и никогда не останавливается; опрос сервера идёт ещё до того, как кто-то открыл экран. Lazily: подписывается на первом подписчике и больше не останавливается; flow продолжает работать после ухода пользователя. WhileSubscribed(5_000): подписывается на первом подписчике и останавливается через 5 секунд после ухода последнего. Поворот экрана пересоздаёт подписчика внутри этого окна, поэтому Room-запрос или сокет не перезапускаются; настоящий уход окно перекрывает, и апстрим останавливается. replayExpirationMillis (по умолчанию никогда) задаёт, сколько кэшированное значение живёт после остановки; 0 заставляет при возврате перезагрузить данные, а не показать устаревшие. Баги, которые списывают не на то WhileSubscribed работает, только если UI действительно отписывается: collectAsStateWithLifecycle() или repeatOnLifecycle. Обычный collectAsState() никогда не даёт счётчику подписчиков дойти до нуля. Свойство с get(), возвращающим flow.stateIn(...), при каждом обращении создаёт новый StateFlow и новый сбор апстрима. Инициализируйте его один раз. stateIn и shareIn stateIn требует начальное значение, сливает значения (conflation) и отбрасывает равные значения; shareIn принимает replay и не отбрасывает равные значения. Практическое правило: WhileSubscribed(5_000) для стейта экрана, а Eagerly только для данных, нужных приложению независимо от того, открыт ли экран.

Обычно внутри flow, отданного через stateIn(viewModelScope, WhileSubscribed(5_000), initial): загрузка стартует, когда экран впервые начинает сбор, переживает поворот и останавливается, когда никто не смотрит. init { } запускает её при создании, а LaunchedEffect повторяет на каждом повороте. init { } Загрузка стартует при создании ViewModel, до того как кто-либо начал сбор, даже если экран так и не отрисуется. Сборщик, запущенный там, работает до onCleared(), поэтому сокет или поток геолокации продолжают работать, пока приложение в фоне. В тесте конструктор уже запустил загрузку или поставил её в очередь, раньше, чем тест успел что-то подготовить, а повтор требует функции load() с защитой от двойной загрузки. Сильная сторона: простой императивный код, который легко сочетать с действиями пользователя, вызывающими _state.update { }. LaunchedEffect(Unit) { vm.load() } Он срабатывает каждый раз, когда composable входит в композицию. Поворот пересоздаёт композицию, но сохраняет ViewModel, как и возврат на экран из back stack, поэтому загрузка каждый раз запускается снова. Теперь UI решает, когда грузить данные, вопреки однонаправленному потоку данных, а ViewModel нужен флаг «уже загружено»: оба признака говорят о неправильном месте. stateIn с загрузкой внутри flow Лениво: создание ViewModel ничего не стоит, а тест запускает загрузку сбором (test { } из Turbine). WhileSubscribed(5_000) сохраняет результат при повороте и останавливает работу в фоне. Возврат больше чем через 5 секунд показывает закэшированное значение и запускает загрузку снова; если это не нужно, берите SharingStarted.Lazily. Цена: действия пользователя приходится превращать во входы цепочки. Повтор становится триггер-flow, за которым идёт flatMapLatest, фильтр становится MutableStateFlow, подмешанным через combine. То, что императивно пишется легко, здесь требует больше продумывания.

У SharedFlow один буфер на всех подписчиков, и слот освобождается, только когда его забрал самый медленный. У MutableSharedFlow() буфера нет совсем, поэтому emit приостанавливается, пока значение не заберут все подписчики: эмиттер идёт в темпе медленного подписчика, а быстрые ждут вместе с ним. Без буфера Каждый подписчик хранит свою позицию в общем буфере. Если один подписчик тратит на значение 300 мс, а другой быстрый, emit возвращается, только когда значение забрал медленный, а быстрый уходит вперёд максимум на одно значение (это видно в примере). tryEmit никогда не приостанавливается, поэтому возвращает false всякий раз, когда emit пришлось бы приостановиться. У MutableSharedFlow<T>() с любым подписчиком так происходит при каждом вызове: так возникает классический баг «события не доходят», и поэтому шины событий объявляют extraBufferCapacity. Без подписчиков он возвращает true, а значение просто пропадает. Что меняют параметры extraBufferCapacity = 64: эмиттер и быстрые подписчики могут уйти вперёд самого медленного на 64 значения; дальше emit снова приостанавливается. Буфер вмещает replay + extraBufferCapacity значений. onBufferOverflow решает, что делать с полным буфером: SUSPEND (по умолчанию) приостанавливает emit, DROP_OLDEST выбрасывает самое старое значение, DROP_LATEST выбрасывает новое. С любой политикой отбрасывания emit никогда не приостанавливается, а выброшенное значение теряется для всех подписчиков, а не только для медленного. Политике отбрасывания нужен буфер: MutableSharedFlow<Int>(onBufferOverflow = DROP_OLDEST) без replay и extraBufferCapacity бросает IllegalArgumentException. Как не дать медленному подписчику тормозить остальных Пусть медленный подписчик буферизует сам: bus.buffer(Channel.UNLIMITED).collect { slowWrite(it) } собирает шину в быстрой корутине и ставит работу в очередь за ней, так что эмиттер и остальные не ждут. StateFlow так не тормозит: он сливает значения (conflation), и каждый подписчик читает последнее. Берите его, когда важно только последнее значение.

Flow является интерфейсом с одной функцией, suspend fun collect(collector: FlowCollector<T>), и у FlowCollector она тоже одна: suspend fun emit(value: T). Сбор выполняет код продюсера, а каждый emit напрямую вызывает код сборщика, без очереди и без потока между ними. Как выполняется цепочка flow { } возвращает flow, чей collect выполняет ваш блок с коллектором даунстрима в качестве ресивера. Оператор вроде map возвращает новый flow, чей collect собирает flow выше и эмитит преобразованные значения вниз. Поэтому source.map { }.filter { }.collect { } сводится к вложенным вызовам, и каждое значение проходит через цепочку вызовов emit на одном стеке до финальной лямбды. Отсюда бесплатный backpressure по умолчанию: продюсер остаётся приостановленным внутри emit, пока даунстрим обрабатывает значение. Канал между ними ставят только операторы, которым нужна конкурентность: buffer, flowOn, channelFlow, combine, flatMapMerge. Что добавляет flow { } Сборщик оборачивается в SafeCollector, который проверяет каждый emit: он должен идти из собирающей корутины, в том же контексте (context preservation), иначе падает с «Flow invariant is violated». Он же проверяет отмену на каждом emit. flowOf(...) и asFlow() ради скорости эту проверку пропускают, поэтому цикл по ним после отмены доходит до конца; .cancellable() её добавляет. Реализация Flow через object компилируется, но обходит эти проверки; документация называет интерфейс нестабильным для наследования и предлагает flow { } или AbstractFlow. Как написать оператор Оператор пишется как расширение для Flow<T>, которое возвращает flow { collect { value -> ... emit(...) } }. Для операторов без состояния, выдающих ноль или больше значений на каждое входное, есть сокращение transform { }. Состояние одного сбора держите внутри блока flow { }. var, объявленная в теле функции снаружи, оказывается общей для всех сборов возвращённого flow: второй toList() в примере начинается с последнего значения первого. Никогда не ставьте catch вокруг emit (он проглотит исключения даунстрима) и не эмитьте из другой корутины или контекста: для этого есть channelFlow и flowOn.