Потоки (Streams)

Future отдаёт один результат или одну ошибку, один раз; Stream отдаёт сколько угодно значений и ошибок со временем и потом может завершиться. Future подходит для разовой операции, Stream нужен, когда значения продолжают приходить. **Future**: загрузить пользователя, отправить форму, прочитать файл целиком. Вызывающий код пишет await, а ошибка просто выбрасывается **Stream: сообщения вебсокета, показания датчика, прогресс загрузки, смена состояния авторизации. Вызывающий код реагирует на каждое значение и владеет подпиской, которую нужно отменить Преобразование**: stream.first и stream.toList() дают Future; future.asStream() даёт стрим из одного события Тревожный знак: стрим, который всегда отдаёт ровно одно событие и закрывается, на деле оказывается замаскированным Future

Stream<T> представляет собой асинхронную последовательность событий: данные, ошибки и финальное завершение (done). Single-subscription стрим допускает одного слушателя за всю жизнь и копит события, пока тот не появится; broadcast допускает много слушателей, но теряет события, которые никто не слушает. Single-subscription (по умолчанию): один слушатель навсегда; слушатель может поставить источник на паузу. Чтение файла, тело HTTP-ответа, сокет, генераторы async* Broadcast: сколько угодно слушателей, которые подключаются и отключаются; поздний слушатель видит только события с момента подписки. UI-события, шина событий, StreamController.broadcast() Проверка и адаптация: stream.isBroadcast говорит, какой у вас стрим; asBroadcastStream() позволяет нескольким слушателям делить single-subscription стрим Как слушать: listen возвращает StreamSubscription, которую потом нужно отменить; await for перебирает события внутри async-функции

Напечатается только world: broadcast-контроллер без слушателей сразу выбрасывает hello. Без .broadcast() напечатается hello, а затем world, потому что single-subscription контроллер копит события, пока не появится его единственный слушатель. Почему: у broadcast-стрима нет памяти; слушатель видит только события, добавленные после подписки Если позднему слушателю нужно последнее значение: храните его в поле, возьмите BehaviorSubject из rxdart или держите состояние в Bloc или Riverpod Доставка всё равно асинхронная: world печатается в следующей микрозадаче, уже после возврата из add

FutureBuilder ждёт один Future и перестраивается, когда тот завершится; StreamBuilder подписывается на Stream и перестраивается на каждом событии, пока стрим не закроется. Оба передают в builder AsyncSnapshot с connectionState, data и error. Состояния: у future: waiting → done; у стрима: waiting → active (перестройка на каждое событие) → done, когда он закрылся Подписка: StreamBuilder отменяет подписку, когда уходит из дерева или получает другой экземпляр стрима Общая ловушка: создавайте future или стрим в initState, в поле или провайдере, но не прямо в build, иначе каждая перестройка перезапускает запрос или переподписывается Выбор по источнику: разовый запрос → FutureBuilder; Firestore snapshots(), состояние авторизации, вебсокет → StreamBuilder

StreamController объединяет sink, в который вы пушите события (add, addError, close), и stream, который вы отдаёте наружу. Чтобы обернуть колбэк-API, регистрируйте колбэк в onListen, пересылайте каждый вызов через add и снимайте колбэк в onCancel. Запуск по требованию: регистрация колбэка в onListen значит, что ничего не работает и батарея не тратится, пока никто не слушает Остановка с последним слушателем: у broadcast-контроллера onListen и onCancel срабатывают на первом и последнем слушателе, поэтому источник может запускаться и останавливаться много раз Правильные сигналы: сбои отправляйте через addError, а не throw; вызывайте close(), когда источник закончился насовсем; проверяйте isClosed перед add, потому что добавление после закрытия бросает StateError Выбор вида: обычный контроллер буферизует события и поддерживает паузу для одного владельца, например репозитория; .broadcast() берут, когда слушают несколько виджетов, с учётом того, что события без слушателей теряются Скрывайте контроллер: возвращайте Stream<T>, чтобы вызывающий код не мог пушить в него или закрыть его Когда не нужен: если источник уже представляет собой цикл по future, генератор async* короче и сам освобождает ресурсы в finally

StreamSubscription и есть хэндл, который возвращает listen: через него доставку ставят на паузу, возобновляют и отменяют. В State её хранят в поле, создают в initState и отменяют в dispose, иначе колбэк продолжит срабатывать после того, как виджета уже нет. **cancel()**: останавливает доставку; у single-subscription стрима заодно останавливает источник (генератор async* выполняет свой finally, чтение файла закрывает файл). У broadcast-стрима просто отписывает вас **cancel() возвращает Future: ждите его, когда очистка должна завершиться раньше следующего шага, например закрыть файл перед удалением; повторный вызов безопасен pause() / resume(): на паузе события копятся; у single-subscription стрима на паузу встаёт и источник Виджет получил новый стрим**: в didUpdateWidget отмените старую подписку и подпишитесь на новый стрим Чем грозит забытый cancel: ошибки setState() called after dispose(), а замыкание держит в памяти весь State вместе с context, то есть утечка

await for обрабатывает события по одному внутри async-функции и не берёт следующее, пока не закончится тело цикла; listen регистрирует колбэк, сразу возвращает управление и даёт StreamSubscription, чтобы потом отменить подписку. **await for**: последовательно и с backpressure: пока тело цикла ждёт, single-subscription источник стоит на паузе. Ошибки приходят как исключения, их ловят через try/catch, и первая же ошибка завершает цикл **listen**: инструмент для виджетов и сервисов: есть onError, onDone и хэндл, который отменяют в dispose. async-колбэк никто не ждёт, поэтому следующее событие может начаться, пока предыдущий обработчик ещё работает Досрочный выход: break или return внутри await for отменяют подписку, но остановить цикл снаружи функции нельзя Частый баг: при await for по стриму, который никогда не закрывается (сокет, состояние авторизации), код после цикла не выполнится никогда

Напечатается created, listened, start, 1, 2. Вызов async*-функции только создаёт стрим; тело начинает выполняться после подписки, и даже тогда в следующей микрозадаче, поэтому синхронный код после listen выполняется раньше. Ленивый, или «холодный»: без слушателя start не напечатается никогда; работа начинается на listen, а не на вызове функции Асинхронная доставка: даже первый yield доходит до слушателя в микрозадаче, а не внутри вызова listen **Пауза на yield**: пока слушатель на паузе, генератор ждёт на своём yield; отмена подписки заставляет этот yield вернуться и выполняет блоки finally Single-subscription: второй stream.listen бросит исключение; чтобы запустить заново, вызовите numbers() ещё раз

StreamBuilder сам подписывается на стрим, перестраивает builder на каждом событии и отменяет подписку, когда виджет уходит из дерева. Всё, что он знает, приходит в AsyncSnapshot. **connectionState**: waiting до первого события, active, пока идут события, done, когда стрим закрылся. done не значит «успех»: у закрытого пустого стрима данных не было **data и error**: последнее событие; snapshot не хранит оба сразу, поэтому событие-ошибка сбрасывает data в null. Если последнее хорошее значение должно пережить ошибку, храните его сами **initialData: заполняет первый кадр, чтобы UI не мигал спиннером Частая ошибка**: создавать стрим прямо в аргументе stream:: на каждой перестройке получается новый экземпляр, StreamBuilder переподписывается, заново запускает запрос и откатывается в waiting; если новый стрим оборачивает single-subscription источник, вторая подписка бросит исключение. Держите стрим в поле или провайдере

Стрим single-subscription: он допускает ровно одного слушателя за всю жизнь, и отмена подписки место не освобождает. Падает listen второго виджета. Подпиской считается больше, чем кажется: listen, await for, first, toList, forEach и каждый StreamBuilder занимают ту самую единственную подписку, поэтому ошибку ловит даже один виджет, который подписывается повторно, например через listen внутри build или через StreamBuilder, которому передали новый объект стрима поверх того же single-subscription источника. Откуда такие стримы: обычный StreamController(), генератор async*, чтение файлов и сокетов, тела HTTP-ответов Лучшее решение: подписаться один раз и кормить хранилище: один слушатель обновляет ValueNotifier, Bloc или провайдер, а виджеты смотрят на него; тогда и поздние виджеты видят текущее значение **StreamController.broadcast(): если контроллер ваш и правда нужны несколько независимых слушателей asBroadcastStream()**: чтобы адаптировать чужой стрим; только broadcast не бесплатен: опоздавшие не получат прошлые события, а backpressure нет

Ошибки передаются как события: addError или throw внутри тела async* отправляют их по тому же конвейеру, что и данные, в onError слушателя. Если onError у слушателя нет, ошибка уходит в обработчик неперехваченных ошибок зоны (во Flutter это PlatformDispatcher.instance.onError). **onError в listen**: получает ошибку и её стектрейс; один listen(onData) обрабатывает данные и выпускает наружу каждую ошибку **handleError**: ловит ошибки посреди конвейера; колбэк test позволяет проглатывать только ожидаемые типы **await for**: ошибка выбрасывается на цикле, поэтому оберните его в try/catch. await for, first и forEach останавливаются на первой ошибке, а listen продолжает работу, если не передать cancelOnError: true **throw в async* не роняет вызывающий код; он превращается в событие-ошибку, и генератор завершается Ошибки как состояние**: на границе репозитория или Bloc превращайте сбои транспорта в состояние вроде Failed(e), чтобы UI показал повтор, а не мёртвый стрим **Не FlutterError.onError**: этот обработчик получает ошибки фреймворка из build, layout и жестов, а не ошибки стримов

StreamController никогда не закрывается, поэтому стрим так и не отправляет done: после 100% await for продолжает ждать, и showDone() не вызывается, как и всё, что идёт после await finish(...) (скрытие индикатора, блок finally). Почему: стрим заканчивается, только когда об этом скажет источник: close() у контроллера или конец тела async*. Завершение Future загрузки стриму ничего не сообщает, а await for не отличает «событий пока нет» от «событий больше не будет» Как работает исправление: whenComplete(progress.close) закрывает контроллер и после успеха, и после ошибки, так что цикл либо завершается штатно, либо выбрасывает ошибку загрузки Где ещё всплывает: toList(), last, length, drain() и StreamBuilder, который ждёт ConnectionState.done, тоже ждут done Как поймать в следующий раз: включите линт close_sinks: он подсвечивает StreamController, созданный в функции, где нигде не вызывается close; в тесте expect(stream, emitsInOrder([..., emitsDone])) зависает до таймаута, если done не приходит

StreamTransformer<S, T> превращает один стрим в другой через stream.transform(...); utf8.decoder и LineSplitter тоже являются трансформерами. Своё: fromHandlers для логики по событиям, метод-расширение с async* для логики с состоянием. **fromHandlers**: handleData и handleError получают событие и EventSink, а handleDone получает только sink, так что одно входное событие может дать ноль, одно или несколько выходных **Расширение с async*** читается лучше всего для операторов с состоянием вроде scan; результатом будет single-subscription стрим, backpressure достаётся бесплатно, а первая ошибка источника его завершает **StreamTransformer.fromBind или StreamTransformerBase: когда нужен полный контроль: таймеры, сохранение broadcast-режима, своя логика отмены Соблюдайте контракт: пробрасывайте ошибки, которые не обрабатываете, отменяйте источник при отмене выхода, закрывайте выход, когда вход завершился Сначала встроенные**: map, where, expand, asyncMap, asyncExpand, distinct, take, skip и timeout покрывают большинство задач

merge пропускает каждое событие каждого стрима по мере прихода; zip спаривает первые события всех стримов, потом вторые и так далее; combineLatest отдаёт последние значения всех стримов каждый раз, когда любой из них что-то выдаёт. **merge**: один выход в порядке прихода для нескольких источников одного типа, например двух чат-каналов; завершается, когда завершились все входы. Rx.merge в rxdart или StreamGroup.merge из package:async **zip: спаривает по индексу и завершается вместе с самым коротким входом, поэтому темп задаёт самый медленный combineLatest**: производное состояние, например «форма валидна, когда валидны email и пароль»; ничего не выдаёт, пока каждый вход не отдал хотя бы одно значение, поэтому задавайте входам начальное значение через startWith или BehaviorSubject.seeded Частая ошибка: zip для состояния: если один вход выдаёт вдвое чаще, пары разъезжаются, а несовпавший хвост копится в памяти

rxdart добавляет BehaviorSubject (стрим с текущим значением), debounceTime (ждать паузу) и switchMap (отменять предыдущий внутренний стрим); все они по-прежнему возвращают обычный Stream. **BehaviorSubject**: broadcast-контроллер, который помнит последнее значение и отдаёт его каждому новому слушателю; .value читает его синхронно. Именно это делает стрим пригодным для состояния: виджет, подписавшийся поздно, всё равно видит текущее значение **PublishSubject и ReplaySubject: первый работает как обычный broadcast-контроллер; второй переигрывает последние N или все события, полезен для логов и кэшей debounceTime: выдаёт событие только после паузы; оператор для поисковой строки switchMap: превращает каждое событие во внутренний стрим и отменяет предыдущий, чтобы устаревший результат для «flu» не пришёл после «flut» Ещё**: combineLatest, withLatestFrom, startWith, throttleTime, bufferTime Когда не добавлять: с Riverpod или Bloc большая часть этого уже есть (в Bloc это restartable() и свои трансформеры событий); для одного маленького конвейера хватит оператора на async* Частая ошибка: subject является контроллером, которым владеете вы, поэтому закрыть его в dispose должны вы

List (и любая модель без ==) сравнивается по ссылке, поэтому каждый свежераспарсенный результат опроса считается новым значением, и distinct() его пропускает. Сравнивайте по содержимому своим equals или дайте элементам равенство по значению. Списки сравниваются по ссылке: передайте distinct(const ListEquality<Order>().equals) из package:collection (для него нужен == у Order) или дешёвый ключ: distinct((a, b) => a.etag == b.etag) лучше глубокого сравнения тысячи строк Дайте модели равенство по значению: переопределите == и hashCode, используйте record или сгенерируйте через Equatable или freezed Сначала сужайте: сделайте map ровно в тот срез, который рисует виджет, и только потом distinct: бейдж следит за int, а не за списком У каждой подписки своя память: distinct помнит предыдущее событие для каждого слушателя, поэтому первое событие позднего подписчика проходит всегда

Вы решили, что у стрима есть текущее значение. Его нет: first подписывается и ждёт следующее событие после этого момента, поэтому на broadcast-стриме, который уже отдал цену, вы ждёте следующую, а она может не прийти никогда. **Что делает first: подписывается, берёт первое пришедшее событие, отменяет подписку и завершает future; на broadcast-стриме всё, что было до этого, потеряно Другие ловушки**: если стрим закроется без событий, first бросит StateError: No element; на single-subscription стриме он съедает единственную подписку Храните значение там, где его можно прочитать: поле репозитория, ValueNotifier, BehaviorSubject с .value или состояние Bloc/Riverpod Проектируйте API парой: Future<T> current() для «что сейчас» и Stream<T> changes() для «скажи, когда изменится» Как это держать в голове: стрим представляет собой последовательность произошедшего, а не переменную, которую можно прочитать.

Правило такое: builder только рисует. Flutter может вызвать его сколько угодно раз, поэтому побочный эффект вроде снекбара или навигации должен жить в колбэке подписки, а не в build. Почему три раза: builder выполняется на каждой перестройке, а не только на событии стрима: перестройка родителя, смена темы, кадр анимации перехода или hot reload запускают его снова с тем же snapshot, а snapshot держит значение «вошёл», пока стрим открыт **Почему не спасает addPostFrameCallback: он обходит исключение «called during build», но срабатывает столько же раз Где это размещать**: подпишитесь один раз в initState, храните последнее состояние в поле и показывайте снекбар в колбэке listen; в bloc это BlocListener, в Riverpod это ref.listen **Проверяйте mounted**: только если колбэк чего-то ждёт (await) перед обращением к context Эмпирическое правило: builder отвечает «как выглядит это состояние», а listener отвечает «что сделать теперь, когда это случилось».

open перезаписывает _sub, не отменяя предыдущую подписку, поэтому каждый открытый до этого чат продолжает слушать, и его сообщения попадают в тот же список. dispose отменяет только последнюю подписку, остальные утекают, пока живы их стримы. Почему: присвоение нового значения полю ничего не отменяет: старая StreamSubscription остаётся активной, а её замыкание держит в памяти контроллер и ValueNotifier Как работает исправление: _sub?.cancel() в начале open останавливает предыдущий чат до того, как начнётся следующий; повторная отмена или отмена null безвредны Тот же баг в других местах: listen внутри refresh(), в didUpdateWidget без отмены старой подписки или на каждое нажатие кнопки: каждый вызов добавляет слушателя, и события приходят по два-три раза Пусть переключением занимается стрим: представьте открытый чат как стрим id и используйте switchMap из rxdart: он сам отменяет предыдущий внутренний стрим Как поймать в следующий раз: линт cancel_subscriptions здесь молчит, потому что _sub отменяется в dispose; тест, который открывает чат A, затем B, пушит сообщение в A и ожидает, что список B не изменился, падает сразу

Стримы описывают состояние как последовательность значений: источник выдаёт новое неизменяемое состояние, виджеты перестраиваются через StreamBuilder. У голого стрима нет текущего значения, воспроизведения для поздних виджетов и владельца, поэтому берут Bloc, Riverpod или BehaviorSubject. Реактивная идея: UI является функцией от состояния; виджеты не спрашивают «что сейчас», им сообщают, когда что-то изменилось Чего нет у голого стрима: синхронного текущего значения для первого кадра, воспроизведения последнего значения для виджетов, подписавшихся позже, пропуска одинаковых состояний и того, кто отвечает за закрытие контроллера Bloc: bloc.stream представляет собой broadcast Stream<State> плюс синхронный геттер state; BlocBuilder добавляет buildWhen и освобождение ресурсов Riverpod: StreamProvider превращает стрим в AsyncValue с состояниями загрузки и ошибки; основное состояние живёт в Notifier или AsyncNotifier Где голые стримы всё ещё к месту: источники данных, которые по природе стримы: Firestore snapshots(), watch-запросы Drift, вебсокеты, состояние сети. Их один раз адаптируют в слой состояния

Backpressure возникает, когда медленный потребитель заставляет быстрый источник ждать. В Dart он работает только у single-subscription стримов с источником, который умеет ждать, например async* на yield; broadcast-стрим источник не тормозит, и медленный слушатель просто копит события. Как передаётся пауза: subscription.pause() или await for, пока тело цикла ждёт, ставят контроллер на паузу; срабатывает onPause, и генератор останавливается на следующем yield. Тела HTTP-ответов и File.openRead так же перестают читать **controller.add никогда не блокирует**: если ваш код продолжает вызывать add на контроллере на паузе, события копятся в памяти; настоящий backpressure требует источника, который проверяет isPaused или реагирует на onPause / onResume **async-колбэк в listen его не даёт**: стрим не ждёт колбэк, поэтому следующее событие доставляется сразу; используйте await for или asyncMap, который ставит источник на паузу, пока выполняется каждый future Broadcast: пауза одного слушателя копит события только для него; источник продолжает работать. Годится для тапов, не годится для больших объёмов данных Когда источник нельзя замедлить: датчики, скролл, лента котировок: сбрасывайте нагрузку через throttleTime, sampleTime или bufferTime из rxdart или храните только последнее значение Симптом в продакшене: во время большого импорта память стабильно растёт: неограниченный буфер между быстрым источником и потребителем, который ни разу не притормозил его

Добавьте stream.timeout, чтобы тишина стала событием: если за окно не пришло ни одной цены, конвейер получит TimeoutException, а UI сможет показать «устарело» и переподключиться. Больше ничто в конвейере не заметит, что события кончились. **Что делает timeout**: отсчёт стартует при подписке и сбрасывается на каждом событии; по истечении эмитит ошибку или то, что ваш колбэк onTimeout положит в sink Источник он не трогает: сокет остаётся открытым, запоздавшее событие всё равно пройдёт; если соединение нужно рвать, рвите сами Heartbeat: мёртвое мобильное соединение может молча висеть минутами, поэтому шлите ping (pingInterval у IOWebSocketChannel) и требуйте ответ Переподключение означает новый стрим: закрытый канал не откроется снова; держите UI на стриме, которым владеете вы, и меняйте сокет под ним с экспоненциальной задержкой Показывайте «устарело» как состояние: цена без пометки, которая может быть неверной, и есть баг, о котором напишет пользователь

Нет. Асинхронная доставка входит в контракт: add планирует событие, а слушатель выполняется в следующей микрозадаче, поэтому код после add всегда завершается первым. sync: true прячет симптом, запуская чужой код слушателя прямо посреди вашего add. Зачем доставка асинхронная: можно делать add до появления слушателя (контроллер буферизует), код слушателя никогда не выполняется посреди метода источника, и все слушатели видят события в порядке отправки **Что на самом деле делает sync: true**: слушатель выполняется в стеке add, но только если подписка есть и не на паузе; иначе событие буферизуется как обычно. «Иногда синхронно» оказывается худшим из контрактов Повторный вход (reentrancy): при sync: true слушатель работает в стеке источника и может обратиться к нему же (setState, событие в bloc), пока источник на середине обновления своего состояния; у broadcast-контроллера со sync вложенный add бросает Bad state: Cannot fire new event. Исключение слушателя до add не доходит: оно уходит в обработчик неперехваченных ошибок зоны Как чинить тест: дайте очереди провернуться через await pumpEventQueue(); лучше проверяйте сам стрим через expectLater и матчеры стримов, а при таймерах вроде debounce берите fakeAsync