Стримы

За стримом задача контроллера следует в ребёнке, которому принадлежит подписка: ctx.each запускает этого ребёнка и отдаёт колбэку каждое событие вместе с контекстом ребёнка. Подписка живёт ровно столько, сколько ребёнок: он кончается, когда кончился стрим, а отмена ребёнка или задачи снимает подписку. Эта страница о том, что к такому ребёнку добавляет контроллер: очередь, которую он держит, close(), стартовые правила и стрим другого контроллера.

hw принадлежит приложению, из которого взяты примеры, а в состоянии Ready, с полем position, работает его контроллер.

Обработка стрима

Контроллер переносит в состояние позиции, о которых сообщает его устройство.

Первая попытка

// Первая попытка: задача кончается до первой позиции.
Job<void> track() => enqueue<Ready, void>(key: 'track', (ctx) async {
  hw.positions.listen((p) => ctx.emit(ctx.state.copyWith(position: p)));
});

Тело подписывается и возвращается, поэтому задача кончается раньше, чем придёт первая позиция, и очередь идёт дальше. Подписка остаётся и держит контекст задачи, которая уже кончилась. На каждой позиции, которую она доставляет, emit бросает ошибку, и та уходит в зону, а состояние не меняется:

Bad state: Job(track) has already finished, cannot emit

Снять эту подписку тоже нечем: задачи, которую можно отменить, уже нет, а close() о подписке ничего не знает. Регистрация sub.cancel через ctx.onDispose этого не чинит: уборка идёт сразу, как тело вернулось, и отменяет подписку раньше, чем придёт первая позиция.

Ребёнок, которому принадлежит подписка

Job<void> track() => enqueue<Ready, void>(
  key: 'track',
  (ctx) => ctx.each(hw.positions, (childCtx, p) {
    childCtx.emit(childCtx.state.copyWith(position: p));
  }).value,
);

ctx.each(stream, onData) создаёт и возвращает дочернюю SoloJob<void>, которая владеет подпиской. Каждый колбэк события получает контекст этого ребёнка. Обращайтесь через него к состоянию и ждите через него же: так ожидание слышит отмену. Возврат .value ребёнка передаёт ошибки стрима и колбэка в тело родителя. Для разбора исхода используйте .done. Чтобы остановить только эту подписку, сохраните возвращённую Job, вызовите её cancel() и ждите в теле .done, а не .value: Cancelled из .value отменяет и родителя, об этом ниже.

Ребёнок завершается, когда источник пришлёт onDone, поэтому при открытом стриме без событий ребёнок всё ещё работает. А детей родитель ждёт и без явного await, так что работает и он, и очередь занята. Принятая отмена родителя, в том числе при close(), отменяет ребёнка; дренаж, close(mode: SoloCloseMode.drain), ничего не отменяет и ждёт конца стрима. Ребёнок отменяем, даже если родитель нет. После возврата из тела родителя он сохраняет его рабочий тип и keepWhile, но не повторяет canStart. Наблюдатели видят его отдельной Job, и имя ей даёт сам ctx.each: Job(each), если вызов не передал key или describe. Раз ребёнок сохранил бы рабочий тип и keepWhile родителя, ctx.each сначала спрашивает их, как это делает ctx.state. Если родитель сам опубликовал через emit состояние, которое они отклоняют, ctx.each заканчивает родителя отменой и бросает её, а не возвращает ребёнка, который так и не начал слушать стрим.

Отмена только ребёнка напрямую не отменяет родителя. Неперехваченный Cancelled из .value ребёнка отменяет родителя через его тело. Ребёнок, исход которого никто не читает, как прогресс в _sync в начале страницы «Дочерние задачи и цепочки», родителя по-прежнему держит, но его провал провалом родителя не становится. Родитель кончается Done, а провал уходит в onUnanswered и по умолчанию в зону, как всякий провал, на который никто не посмотрел; о них раздел «Обработанные и необработанные ошибки» страницы об ошибках. ignoreFailure() на ребёнке не пускает его ни туда, ни туда.

События доставляются по одному, а отмена ждёт колбэк, который уже начался. Поэтому ожидания и делаются контекстом колбэка: childCtx.abandonable заканчивается отменой в тот же миг, когда она пришла, и оставляет действие доигрывать само, а обычный await кончается только вместе со своей future. Пока колбэк не вернулся, ребёнок и родитель ещё работают, очередь занята, а close() не возвращается. Не ждите внутри колбэка события ни завершения самого ребёнка, ни его cancel(): это взаимная блокировка. Ребёнок ждёт возврата из колбэка, колбэк ждёт ребёнка, а родитель ждёт ребёнка, и с места уже никто не сдвинется. Чтобы остановить подписку изнутри колбэка, вызовите cancel() и не ждите его.

Ребёнок не ждёт источник: future, которую возвращает cancel() подписки, несёт уборку самого источника, а ею задача не владеет. Если эта уборка упадёт, её ошибка уйдёт в onError контроллера и дальше в его onUnanswered, как всякая ошибка без исхода. Задача, которой нельзя кончиться раньше, чем закрылся её источник, дожидается этого сама, как в разделе «Дождаться закрытия источника» страницы async_job о стримах. Остальное на той странице верно и здесь, только вместо Job.deferred контроллер пишет job(...) или ctx.child. Там описано, что делают в теле await for и listen, привязанный к задаче, как колбэк в несколько шагов останавливается на отмене и где место уборке одного события.

Слежение за другим контроллером

Экран показывает, вошёл ли пользователь, а сессия живёт в состоянии другого контроллера.

Первая попытка

// Первая попытка: экран не видит текущую сессию.
final class ScreenController extends Solo<Screen> {
  final SoloStream<Session> session;

  ScreenController(this.session) : super(const Screen());

  Job<void> follow() => enqueue<Screen, void>(
    key: 'follow',
    (ctx) => ctx.each(session.stream, (childCtx, next) {
      childCtx.emit(childCtx.state.copyWith(signedIn: next.signedIn));
    }).value,
  );
}

Стрим контроллера присылает только изменения состояния: текущее состояние подписчик при подписке не получает. Сессия, в которую уже вошли к моменту старта follow(), ничего не присылает, и экран показывает signedIn: false, пока сессия не изменится.

Сначала состояние, потом стрим

Job<void> follow() => enqueue<Screen, void>(key: 'follow', (ctx) async {
  void take(SoloContext<Screen, Screen> target, Session next) =>
      target.emit(target.state.copyWith(signedIn: next.signedIn));

  final following = ctx.each(session.stream, take); // что случится дальше
  take(ctx, session.currentState); // то, что уже случилось
  await following.value;
});

Screen и Session здесь состояния приложения, и у обоих есть свойство signedIn. Подписка и чтение стоят на двух строках без await между ними, поэтому ни одно изменение сессии между ними не попадает. Подписка идёт первой: take пишет состояние, слушатели экрана слышат запись сразу, и один из них может тут же изменить сессию. Ребёнок, который уже подписан, это изменение услышит, а подписка, сделанная после чтения, его бы пропустила.

Подписка удерживает очередь следящего контроллера до своего завершения: ни одна другая Job этого контроллера не начнётся, пока сессия не закроет стрим, и дренаж ждёт того же. Поэтому такая задача подходит контроллеру, который только следит. Контроллер, у которого есть и другие задачи, держит вместо неё собственную подписку, созданную в конструкторе и снятую в onClose. На каждое обновление этой подписки он либо ставит короткую Job, либо, когда обновление сообщает факт, который идущая работа должна услышать сразу, записывает его через externalSetState. Раздел «Внешнее состояние» страницы о состоянии показывает такую подписку и говорит, какой из двух путей нужен обновлению.