Дочерние задачи и стримы

Корневая Job может разделить работу на дочерние:

SoloJob<String> sync(int item) => run<Ready, String>(
      key: _Op.sync,
      (ctx) async {
        // Создан контроллером, запущен родителем и минует очередь:
        // очередь родитель держит за обоих.
        final upload = job<Ready, String>(
          key: _Op.upload,
          (child) => child.join(() => api.push(item)),
        );
        final path = await ctx.run(upload);

        // Ребёнок на стриме. Родитель ждёт его и без await, а отмена
        // родителя отменяет и его.
        ctx.each(
          device.events,
          (child, event) async =>
              child.emit(child.state.copyWith(done: event)),
        );
        return path;
      },
    );

Ребёнок запускается сразу, минуя очередь, с проверкой собственных стартовых правил. Родитель удерживает очередь до завершения всех детей, даже если его тело вернулось раньше.

ctx.run(child) возвращает Future<T>. Он ждёт ребёнка, его детей и освобождения ресурсов. При успехе он проверяет отмену и правила состояния родителя, прежде чем вернуть значение. Ошибка или отмена ребёнка бросается в тело родителя со своим стеком вызовов.

Сохраните исходный child, чтобы отменить его или посмотреть child.done. Для одновременной работы родителя и ребёнка обрабатывайте возвращённую future отдельно. Тогда их записи состояния могут чередоваться. ctx.run(child).ignore() явно игнорирует результат этой future, при этом родитель продолжает ждать детей. Одного child.ignore() недостаточно для обработки ошибок future, которую вернул run.

Принятая отмена родителя передаётся детям. Это относится и к случаю, когда родитель бросает Cancelled, включая неперехваченную отмену из await ctx.run(child). Если тело родителя завершается ошибкой, оно позволяет детям закончить и ждёт их. Ребёнок может отклонить отмену; ctx.run всё равно ждёт его и проверяет родителя после успеха ребёнка.

Ребёнок, отклонённый стартовыми правилами, всё равно получает родителя, уровень вложенности и наблюдателя, но дальнейшего ожидания не требует. Ошибка стартового правила завершает ребёнка с ошибкой и передаётся через ctx.run.

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

ctx.each(stream, onData) создаёт и возвращает дочерний Job<void>, который владеет подпиской. Каждый колбэк события получает контекст этого ребёнка. Используйте его для доступа к состоянию и операций с учётом отмены:

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

Этот прикладной пример копирует позиции устройства в состояние Ready. Возврат .value ребёнка передаёт ошибки стрима и колбэка в тело родителя. Для разбора исхода используйте .done, а для остановки только этой подписки сохраните возвращённую Job и вызовите её cancel().

События обрабатываются по порядку. Асинхронный колбэк завершается до начала следующего; первая ошибка стрима или колбэка прекращает обработку. Отмена сразу снимает подписку, прекращает дальнейшую доставку и ждёт текущий колбэк перед завершением ребёнка. Для ожиданий используйте контекст колбэка; обычный await может удерживать ребёнка и родителя неопределённо долго.

Родитель ждёт этого ребёнка и без явного await, поэтому открытый стрим без событий всё равно удерживает родителя работающим. Принятая отмена родителя, в том числе при close(), отменяет ребёнка. Ребёнок отменяем, даже если родитель нет. После возврата из тела родителя он сохраняет его W и keepWhile, но не повторяет canStart. Наблюдатели видят его отдельной Job.

Отмена только ребёнка напрямую не отменяет родителя. Неперехваченный Cancelled из .value ребёнка отменяет родителя через его тело. Не ждите завершения самого ребёнка или его cancel() внутри колбэка события: ребёнок уже ждёт этот колбэк.

Future, которую возвращает cancel() самой подписки на стрим, не ожидается. При необходимости отдельно дождитесь асинхронной уборки источника. Обычное завершение требует onDone от источника. Как и ctx.run, each не может запустить ребёнка после конца тела родителя, во время уборки или из работы unattended.

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

Контроллер, предназначенный для слежения за другим контроллером, может держать Job над его стримом. Сначала прочитайте текущее состояние, поскольку стрим содержит только последующие обновления:

final class ScreenController extends Solo<Screen> {
  final Solo<Session> session;

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

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

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

Здесь Screen и Session являются состояниями приложения со свойством signedIn. Подписка удерживает очередь следящего контроллера до своего завершения. Если этому контроллеру нужно обрабатывать и другие Job, используйте внешний слушатель, который ставит короткую Job на каждое обновление. Если обновление немедленно меняет допустимость текущей работы, рассмотрите правила страницы Внешнее состояние.

Цепочки завершённой работы

// Запускается после успеха источника, со своим обычным JobContext.
Job<void> syncAndReport(int item) =>
    sync(item).then((ctx, path) => analytics.send(path));

job.then((ctx, value) => ...) создаёт Job, которая запускается после успеха источника, включая его детей и освобождение ресурсов. Колбэк получает результат и новый контекст ядра JobContext, а вернуть может значение или future. Ошибка источника передаётся дальше без вызова колбэка.

Продолжение не наследует контекст состояния, правила, наблюдателя или позицию в очереди контроллера. У него есть собственный необязательный наблюдатель. Для изменения состояния контроллера вызовите метод, ставящий новую Job в очередь; между этими операциями могут выполниться другие Job. Используйте детей одного родителя, если вся последовательность должна удерживать очередь без запуска другой корневой Job между шагами.

Отмена передаётся вперёд продолжениям и назад незавершённым источникам с учётом правил отмены каждой Job. Отмена конца цепочки ждёт эти источники и освобождение их ресурсов, включая источник, отклоняющий отмену. close() достигает продолжения через незавершённый источник, но не владеет продолжением, уже работающим после завершения источника.