Стримы

За стримом задача следует в ребёнке, которому принадлежит подписка: ctx.each запускает этого ребёнка, отдаёт события колбэку по одному и снимает подписку вместе с задачей. Job.each делает то же для задачи, которой больше нечего делать.

Job<void>((ctx) async {
  final saving = ctx.each(messages, (childCtx, message) {
    return childCtx.join(() => store.saveBody(message));
  });
  await saving.value;
});

Раздел «Обработка стрима» приходит к этому коду после трёх попыток, к которым ведёт обычная подписка.

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

Задача сохраняет каждое сообщение стрима: по одному, с провалом, когда сохранение упало, и с остановкой, когда её отменили.

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

// Первая попытка: молчащий поток держит отменённую задачу.
Job<void>((ctx) async {
  await for (final message in messages) {
    await ctx.join(() => store.saveBody(message));
  }
});

Пока идёт сохранение, join служит контрольной точкой: отмена, пришедшая в это время, бросается, когда сохранение кончилось, и цикл отпускает стрим. Между двумя сообщениями контрольной точки нет. Тело стоит внутри await for, который о задаче ничего не знает, поэтому задача, отменённая при молчащем стриме, продолжает идти и держать подписку, пока не придёт следующее сообщение. job.cancel() до этого тоже не возвращается. На стриме, который замолчал насовсем, этого не случится никогда.

Вторая попытка

// Вторая попытка: сохранения идут внахлёст, а их ошибки уходят в зону.
Job<void>((ctx) async {
  final subscription = messages.listen((message) async {
    await store.saveBody(message);
  });
  ctx.onCancel(subscription.cancel);
  await ctx.abandonable(subscription.asFuture<void>);
});

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

listen не ждёт асинхронный колбэк. Второе сообщение приходит, пока сохраняется первое, и его сохранение начинается сразу: оба идут рядом, и какое кончится первым, не сказано нигде.

И колбэк зовёт стрим, а не тело. Ошибка, брошенная в нём, уходит в зону мимо задачи: задача продолжает идти и держать подписку, её исход о провале ничего не знает, и тот, кто ждёт её value, тоже. onError у listen этого не меняет: он слышит ошибки самого стрима, а не то, что бросил колбэк.

Третья попытка

// Третья попытка: задача кончается, пока сохранение ещё идёт.
Job<void>((ctx) async {
  final subscription = messages.asyncMap(store.saveBody).listen((_) {});
  ctx.onCancel(subscription.cancel);
  await ctx.abandonable(subscription.asFuture<void>);
});

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

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

Job<void>((ctx) async {
  final saving = ctx.each(messages, (childCtx, message) {
    return childCtx.join(() => store.saveBody(message));
  });
  await saving.value;
});

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

Одно за другим не значит, что стрим ограничен. Пока колбэк работает, ребёнок держит подписку на паузе, а источник, который слушателя не ждёт (broadcast-стрим как раз такой), продолжает слать: события копятся в приостановленной подписке, сколько бы их ни пришло. Источник, который устойчиво быстрее сохранений, растит эту очередь без предела. Ограничить её, слить события или отбрасывать их решает приложение, на стриме, ещё до того, как он дошёл до each.

saving.value завершается, когда стрим закончился и последнее сохранение выполнено; его ожидание бросает провал ребёнка в тело, а брошенный им Cancelled идёт путём отмены: неперехваченный, он отменяет родителя по обычным правилам исходов дочерней Job. saving.done не бросает: его ожидание возвращает Outcome ребёнка, будь то Done, Failed или Cancelled.

Родитель ждёт ребёнка, даже если его тело вернулось, не ожидая ребёнка. Поэтому открытый стрим держит родителя до конца стрима или отмены ребёнка. Принятая отмена родителя передаётся ребёнку. Наблюдатель видит этого ребёнка как отдельную Job.

Как и run, each запускает детей только пока тело родителя активно; вызовы из unattended или уборки отвергаются. Обычное завершение стрима зависит от отправки onDone источником.

Колбэк, который возвращает задачу Job.deferred, которую никто не запустил, кончает ребёнка Failed с ArgumentError: здесь эту задачу никто не запускает, а событие считалось бы обработанным. Что ребёнок делает с future, которую возвращает cancel самой подписки, — в «Дождаться закрытия источника».

Колбэк в несколько шагов

Сообщение сохраняется в два шага: сперва само сообщение, потом его вложения, и отмена должна остановить сохранение между ними.

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

// Первая попытка: отменённый колбэк всё равно доходит до конца.
ctx.each(messages, (childCtx, message) async {
  await store.saveBody(message);
  await store.saveAttachments(message);
});

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

Контрольные точки самого ребёнка

ctx.each(messages, (childCtx, message) async {
  await childCtx.join(() => store.saveBody(message));
  await childCtx.join(() => store.saveAttachments(message));
});

Контекст ребёнка колбэк получает именно для того, чтобы его шаги могли остановиться по отмене. childCtx.join дожидается того шага, который обёрнут (сохранение на полпути всё равно сохранение), а затем выпускает отмену вместо значения, поэтому второй шаг не начинается вовсе, а уборка родителя идёт сразу, как только кончился первый. childCtx.abandonable остаётся другим выбором, для шага, результат которого можно бросить. Чего внутрь класть нельзя, так это завершения самого ребёнка: ожидание value задачи, которую вернул each, или её cancel() из колбэка не кончится никогда, потому что ребёнок уже ждёт этот колбэк.

Родительский ctx на месте childCtx слышит только отмену родителя. Отмена родителя останавливает шаги так же, а вот ребёнок, отменённый отдельно, как в разделе «Остановить один стрим», доигрывает оба шага до конца.

Что открывает одно событие

Каждое сообщение пишется через черновик, и черновик надо закрыть, когда его сообщение записано.

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

// Первая попытка: черновики закрываются только с концом потока.
ctx.each(messages, (childCtx, message) async {
  final draft = await childCtx.join(
    () => store.openDraft(message),
    dispose: (draft) => draft.close(),
  );
  await childCtx.join(draft.write);
});

Колбэк отдаёт закрытие в dispose, как это делает тело. dispose кладёт его на стек уборки ребёнка, а этот стек разматывается, когда кончается ребёнок, а не когда возвращается колбэк. После двух сообщений оба черновика записаны и ни один не закрыт; закроются они вместе, когда кончится стрим, а на стриме без конца не закроются никогда. ctx.onDispose и discard в колбэке регистрируют на том же стеке. ctx.onCancel в колбэке копится так же, только в своём списке, если колбэк не снимет его функцией, которую onCancel возвращает.

Ребёнок на событие

ctx.each(messages, (childCtx, message) {
  return childCtx.run(
    Job.deferred<void>((eventCtx) async {
      final draft = await eventCtx.join(
        () => store.openDraft(message),
        dispose: (draft) => draft.close(),
      );
      await eventCtx.join(draft.write);
    }),
  );
});

То, что открывает одно событие, принадлежит ребёнку, запущенному для этого события. Колбэк возвращает future от run, поэтому следующее сообщение ждёт этого ребёнка, а его стек разматывается раньше: каждый черновик закрыт до того, как открыт следующий. Отменённый посреди записи, ребёнок всё равно закрывает свой черновик до конца задачи.

Остановить один стрим

Job, которую возвращает each, позволяет родителю прекратить подписку и идти дальше. Этот пример печатает событие каждую секунду, прекращает подписку через 2,5 секунды и продолжает тело родителя:

Future<void> watchTicks() async {
  final job = Job<void>((ctx) async {
    final ticks = ctx.each(
      Stream<int>.periodic(const Duration(seconds: 1), (index) => index + 1),
      (childCtx, tick) => print('Tick $tick'),
    );

    await ctx.pause(const Duration(milliseconds: 2500));
    await ticks.cancel();
    print('Parent continues');
  });

  await job.value;
}

Вывод: Tick 1, Tick 2, затем Parent continues. Стрим ещё не завершён, когда ticks.cancel() снимает подписку. await ticks.cancel() ждёт, пока ребёнок ticks кончится, и только потом тело идёт дальше; отмена ticks сама родителя не отменяет.

Дождаться закрытия источника

final feed = Feed();
ctx.onDispose(feed.close);
await ctx.each(feed.messages, (childCtx, message) => save(message)).value;

В future, которую возвращает cancel() самой подписки, идёт уборка источника: закрывается соединение, сбрасывается на диск файл. Ребёнок отпускает стрим и кончается, не дожидаясь её, поэтому задача может кончиться, пока источник ещё закрывается. Если эта future упадёт, ошибка уйдёт в onError наблюдателя ребёнка и, если он не перехватит её в onUnanswered, в зону. await for, из которого вышли раньше конца стрима, эту future, наоборот, ждёт: код после цикла выполняется, когда источник уже закрылся, а провал этой future бросает сам цикл.

Задача, которой нельзя кончаться раньше источника, кладёт его закрытие на свой стек уборки, как выше. feed.close() здесь возвращает future, которая завершается, когда соединение закрыто. Уборка задачи её ждёт, поэтому задача кончается после неё, а cancel() и родитель, которые ждут задачу, ждут и источник.

Задача, которая только следует за стримом

Future<void> saveAll(
  Stream<String> messages,
  Future<void> Function(String message) save,
) async {
  final saving = Job.each(messages, (ctx, message) {
    return ctx.join(() => save(message));
  });

  await saving.value;
}

Задача из раздела «Обработка стрима» состоит из двух: родителя, которому больше нечего делать, и ребёнка, которому принадлежит подписка. Job.each создаёт одну задачу, которая следует за стримом сама. Это корень, а не ребёнок, и колбэк получает контекст этой задачи. За стримом она следует так же, как ctx.each; key, describe, cancellable и observer те же, что у Job(...).

От ребёнка её отличают три вещи:

• Она стартует внутри вызова. Задача, созданная Job(...), запускает тело следующей микротаской. Job.each к возврату уже подписана, поэтому событие, которое broadcast-стрим отправил сразу после вызова, не теряется. По той же причине onStart наблюдателя выполняется раньше, чем вызывающий получил хэндл, а источник, отдающий событие прямо из listen, доходит до колбэка до возврата из вызова. Колбэк, которому нужна своя задача, читает ctx.job.
• Она корень, где бы её ни создали. Вызванная в теле другой задачи, она не становится её ребёнком: тело её не ждёт и не отменяет, и она не ChildJob, поэтому ctx.run её не принимает. Внутри тела за стримом следуют через ctx.each. За провал, который никто не прочёл, отвечают, как у любого корня.
• С cancellable: false снаружи её не остановить. Такая задача не принимает ни одной отмены, и cancel() не возвращается, пока задача не кончится сама: стрим кончился или упал либо колбэк бросил, в том числе Cancelled. На стриме без конца другого выхода нет.

Оба each берут стрим с одной подпиской один раз, как и listen. Второй each на стриме, который уже слушают, не бросает из вызова: задача, которую он создаёт, кончается Failed с StateError от listen, Bad state: Stream has already been listened to., и этот провал уходит туда же, куда любой другой.