Цепочки

Цепочка продолжает задачу её результатом. Здесь приложение загружает число текстом, разбирает его и сохраняет. then возвращает новую задачу. Она стартует, когда источник кончился успехом вместе с детьми и уборкой, и передаёт колбэку значение и собственный контекст:

final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = loaded.then<int>((ctx, text) => int.parse(text));
final saved = parsed.then<void>(
  (ctx, number) => ctx.join(() => saveNumber(number)),
);

await saved.value;

В блоках вывода на этой странице:

• loaded:, parsed:, saved:, shown: исход этой задачи;
• parent:, child: исходы после tail.cancel() из раздела «Продолжение ребёнка».

Отмена цепочки

Текст ещё загружается, когда пользователь уходит с экрана, и экран отменяет последнее звено:

// Пользователь ушёл с экрана, пока текст загружался.
await saved.cancel();
print('loaded: ${loaded.outcome}');
print('parsed: ${parsed.outcome}');
print('saved: ${saved.outcome}');

loaded: Cancelled(chain)
parsed: Cancelled(chain)
saved: Cancelled(manual)

Отмена идёт по цепочке в обе стороны. Незавершённые звенья перед отменённым она просит отмениться, а звенья после него отменяет. Каждое звено, до которого она доходит, получает ChainCancelReason, и в его поле cause лежит Cancelled звена, от которого она пришла:

Что отменяете	Просит отмениться, если не кончились	Отменяет, когда отменённое звено примет отмену
loaded	никого	parsed и saved
parsed	loaded	saved
saved	loaded и parsed	никого
parsed, сделанная на ручке loaded.shared()	ручку, которая кончается сразу; loaded не просят	saved

Звено, которое отмену отвергает, на себе её и останавливает: отмените loaded, созданную с cancellable: false, и parsed и saved пойдут дальше. Кончившиеся звенья сохраняют свои исходы. О ручке из последней строки — раздел «Два продолжения одного источника».

Звенья после источника узнают о его отмене через whenCancelled: продолжение и ручка shared регистрируются там при создании. Слушатель, зарегистрированный на источнике раньше них, может застать их ещё не отменёнными.

Продолжение стартует только после того, как кончился источник, поэтому у работающего звена позади нет ничего незавершённого. Пока работает parsed, loaded уже кончилась, и loaded.cancel() ничего не делает: parsed идёт дальше, а за ней saved. Пока работает saved, отмена loaded или parsed её не трогает. Чтобы остановить работающее звено, отмените его само или одно из следующих.

await saved.cancel() возвращается, когда кончились незавершённые звенья перед saved вместе с их детьми и уборкой и когда saved кончила начатую работу и уборку. Источник сохраняет свои правила: cancellable: false отклоняет запрос, а uncancellable придерживает его. Отменённое продолжение всё равно ждёт такой источник, а потом кончается, не вызвав колбэк. Пока оно ждёт, у него isCancelled, но ещё не isFinished. Его исход содержит started: false, если отмену приняли, пока оно ждало источник.

Два продолжения одного источника

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

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

// Первая попытка: второе продолжение бросает.
final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = loaded.then<int>((ctx, text) => int.parse(text));
final shown = loaded.then<void>((ctx, text) => showText(text));

Bad state: Job() already has a continuation: a job is continued once; call shared() for each branch

Задача продолжается один раз, и второй then бросает, прежде чем что-либо создано. Продолжение просит отмениться свой незавершённый источник, как любое звено просит звенья перед собой, а источник не знает, сколько продолжений его ждёт. Будь их два, уходящее число отменило бы loaded, и до экрана, который остался, текст так и не дошёл бы.

Ручка на каждую ветку

final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = loaded.shared().then<int>((ctx, text) => int.parse(text));
final shown = loaded.shared().then<void>((ctx, text) => showText(text));

// Число ушло с экрана, пока текст загружался.
await parsed.cancel();
await shown.done;
print('parsed: ${parsed.outcome}');
print('loaded: ${loaded.outcome}');
print('shown: ${shown.outcome}');

parsed: Cancelled(manual)
loaded: Done(21)
shown: Done(null)

shared() делает ручку, которая кончается исходом loaded и до неё не дотягивается. parsed просит отмениться свой источник, а её источник — ручка: ручка кончается сразу и оставляет loaded в покое. Каждый вызов делает новую ручку, так что у каждой ветки свой источник. Отмена, которую приняла loaded, по-прежнему доходит до каждой ручки, и каждая кончается Cancelled(chain), не дожидаясь уборки loaded.

Провал loaded доходит до каждой ручки, которая ещё есть, и отвечают за него один раз, сколько бы ручек его ни несли. Если и parsed, и shown отменили до того, как loaded упала, ручки, которая понесла бы провал, не остаётся. Тогда за него отвечает сама loaded, если её никто не наблюдает, как источник во второй строке таблицы раздела «Когда звено падает». Она отвечает в onUnanswered своего наблюдателя, если тот подмешивает JobAnswerer, а иначе в зоне, где создана loaded, а не там, где вызвали then.

Задачи, которые ждут источник

final loaded = Job<String>((ctx) => ctx.join(loadText));
final parsed = Job<int>(
  (ctx) async => int.parse(await ctx.abandonable(() => loaded.value)),
);
final shown = Job<void>(
  (ctx) async => showText(await ctx.abandonable(() => loaded.value)),
);

// Число ушло с экрана, пока текст загружался.
await parsed.cancel();
await shown.done;
print('parsed: ${parsed.outcome}');
print('loaded: ${loaded.outcome}');
print('shown: ${shown.outcome}');

parsed: Cancelled(manual)
loaded: Done(21)
shown: Done(null)

parsed и shown — отдельные задачи, которые ждут loaded, а не её продолжения, поэтому отмена одной из них прекращает её собственное ожидание и дальше не идёт. loaded принадлежит тому, кто её создал, и этот код отменяет её, когда текст никому не нужен. Провал loaded, пришедший после того, как parsed ушла, достаётся parsed как поздняя ошибка действия, которое ctx.abandonable перестал ждать: см. «Куда идёт каждая ошибка».

Когда звено падает

Упавшее звено передаёт ошибку и стек дальше, не вызывая следующий колбэк. Провал идёт по цепочке вниз и заканчивает каждое звено после упавшего, поэтому тот, кто наблюдает последнее звено через value, done или ignoreFailure, обрабатывает провал из любого места цепочки выше. JobObserver последнего звена слышит его как провал этого звена.

Куда уходит ошибка, зависит от того, что упало:

Что упало	Куда уходит ошибка
Тело звена или колбэк продолжения	Вниз по цепочке, к тому, кто наблюдает последнее звено. Если его никто не наблюдает, за ошибку отвечает последнее звено: в onUnanswered своего наблюдателя, если тот подмешивает JobAnswerer, а иначе в зоне, где вызвали последний then
Источник, который отверг отмену своего продолжения и потом упал	За ошибку отвечает источник, если его никто не наблюдает: в onUnanswered своего наблюдателя, если тот подмешивает JobAnswerer, а иначе в зоне, где создан источник. Отменённое продолжение ошибку не передаёт и поэтому не наблюдает
Работа ctx.unattended или уборщик, ошибку которых не несёт ни один исход	К наблюдателю звена, где она случилась, а потом, если он не перехватит её в onUnanswered, в зону, где звено создано. У продолжения это зона, где вызвали then

Что возвращает колбэк

Колбэк возвращает значение или future, и продолжение ждёт эту future. Здесь число сохраняет отдельная задача.

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

// Первая попытка: продолжение не ждёт задачу.
final stored = parsed.then<void>((ctx, number) {
  final saving = Job<void>((ctx) => ctx.join(() => saveNumber(number)));
  return saving;
});

Код компилируется. Этот колбэк отмечает линт void_checks, который включает package:lints, но стрелочную форму (ctx, number) => saving он не отмечает, а компилятор не отмечает ни ту ни другую. stored не ждёт saving. Вот что происходит:

1. then ждёт future, а Job — не future. Продолжение берёт saving своим значением и сразу кончается Done.
2. saving стартует сама и работает дальше одна, и её никто не ждёт.
3. saving, созданная через Job.deferred, не стартовала бы никогда, потому что её здесь никто не запускает. У продолжения типа void ядро это проверяет, и stored падает с ArgumentError. Продолжение более широкого типа берёт задачу своим значением: задача может быть тем значением, которое оно хочет отдать.

Задача, которую продолжение запускает своим ребёнком

final stored = parsed.then<void>((ctx, number) {
  final saving = Job.deferred<void>(
    (ctx) => ctx.join(() => saveNumber(number)),
  );
  return ctx.run(saving);
});

ctx.run запускает saving и возвращает future её значения, и продолжение ждёт эту future.

Продолжение — не ребёнок

Каждое продолжение — корневая задача ядра, где бы ни вызвали then. Созданное в теле другой задачи, оно не становится её ребёнком, и эта задача его не ждёт. У него есть необязательный аргумент observer. Ни наблюдателя источника, ни того, что движок навесил на источник, оно не наследует.

	Ребёнок	Продолжение
Создаётся	Job.deferred	then
Стартует	когда родитель запускает его через ctx.run	само, когда источник кончился успехом
Его ждёт	родитель	никто: это корневая задача
ctx.run его принимает	да, это ChildJob	нет, это простая Job

then не запускает свой источник. Цепочка, навешенная на Job.deferred, стоит незапущенной, каждое её звено, пока источник не запустят: вызовом start() или через ctx.run(source) в теле, которое берёт его ребёнком.

Что отдаёт источник

opened открывает строки и отдаёт их, а archived пишет их в архив:

final opened = Job<Source>(
  (ctx) => ctx.abandonable(openRows, discard: (rows) => rows.close()),
);
final archived = opened.then<void>((ctx, rows) async {
  ctx.onDispose(rows.close);
  await ctx.join(() => writeArchive([rows]));
});

Уборка, которую зарегистрировал источник, к передаче значения уже выполнена: ресурс, закрытый через onDispose источника, в этот момент уже закрыт. С discard всё наоборот: он выполняется, только если его задача кончилась не Done. opened кончилась Done, поэтому её discard не выполнялся и не выполнится. Получателем становится продолжение: rows приходит к нему аргументом, и закрытие оно регистрирует в своём теле, первой строкой.

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

Как идёт цепочка	Кто закрывает строки
archived получает строки	archived, через ctx.onDispose
archived отменяют, пока opened открывает, и opened принимает отмену	opened, через свой discard, когда строки приходят
То же, но opened её отвергает: она создана с cancellable: false	Вызывающий, через opened: (await opened.value).close()
То же, но opened кончилась к приходу запроса	Вызывающий, через opened, так же

Вызывающий закрывает строки через opened так же, как тело группы закрывает значение ветки, отвергающей остановку группы.

Источник, продолженный через shared() в несколько веток, отдаёт каждой ветке одни и те же строки. Каждую ручку отменяют отдельно, но значение у них одно: ветка, которая регистрирует закрытие, закрывает строки из-под другой, а вторая такая ветка закрывает их второй раз. Когда веток несколько, строки принадлежат коду, который ждёт их все:

final opened = Job<Source>(
  (ctx) => ctx.abandonable(openRows, discard: (rows) => rows.close()),
);
final archived = opened.shared().then<void>(
  (ctx, rows) => ctx.join(() => writeArchive([rows])),
);
final warmed = opened.shared().then<void>(
  (ctx, rows) => ctx.join(() => cache.warm(rows)),
);
await [archived.done, warmed.done].wait;
if (await opened.done case Done(:final value)) value.close();

Ручка не отменяет opened, поэтому закрытие ждёт саму opened так же, как ветки. Если opened кончилась не Done, строки закрыл её discard или они так и не открылись.

Продолжение ребёнка

Родителю нужны строки, а доклад о них может кончиться позже родителя:

final child = Job.deferred<int>((ctx) => ctx.abandonable(load));
final tail = child.then<void>((ctx, rows) => report(rows));

final parent = Job<void>((ctx) async {
  // Источник остаётся ребёнком: родитель его запускает и дожидается.
  ctx.log(await ctx.run(child));
});

await parent.value; // Done, что бы ни делал tail.
await tail.value; // tail наблюдать вам.

Родитель ждёт своих детей, а не то, что на них навешано. Как только ctx.run(child) вернулся и тело закончилось, родитель завершается Done, пока медленное продолжение ещё работает. Что останавливает tail, зависит от того, ждёт ли оно ещё ребёнка:

Что	Пока tail ждёт ребёнка	Пока tail работает
Отмена родителя	Останавливает: родитель отменяет ребёнка, а отмена ребёнка идёт вперёд как ChainCancelReason	Не доходит
tail.cancel()	Останавливает	Останавливает
ctx.onCancel(() => tail.cancel().ignore()) в теле родителя, которое ещё идёт	Останавливает	Останавливает

В обратную сторону продолжение до источника доходит, пока источник работает. tail.cancel(), пока ребёнок ещё загружает, идёт назад по цепочке, как описано в разделе «Отмена цепочки», и просит ребёнка отмениться. ctx.run(child) бросает отмену ребёнка в тело родителя, и родитель тоже кончается Cancelled, с BodyCancelReason:

parent: Cancelled(body: Job(): Cancelled(chain)), child: Cancelled(chain)

Ошибка продолжения касается только его самого. Наблюдайте его через value, done или ignoreFailure; за ошибку, которую никто не наблюдал, отвечает продолжение, в onUnanswered своего наблюдателя или в зоне, где вызвали then, уже после того, как родитель закончился. Исход продолжения возвращается вам, а не родителю.

Шаги подряд

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

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

then читается как «а затем», поэтому доклад становится продолжением, и тело берёт обоих детьми:

// Первая попытка: ctx.run не принимает продолжение.
final child = Job.deferred<int>((ctx) => ctx.abandonable(load));
final tail = child.then<void>((ctx, rows) => report(rows));

final parent = Job<void>((ctx) async {
  ctx.log(await ctx.run(child));
  await ctx.run(tail);
});

Второй ctx.run не компилируется:

The argument type 'Job<void>' can't be assigned to the parameter type 'ChildJob<dynamic>'.

ctx.run принимает ChildJob, задачу, которую запускает её родитель, а продолжение стартует само, когда кончился источник: then возвращает простую Job. Длина цепочки ничего не меняет: child.then(...).then(...) вешает одно продолжение на другое, и ни одно звено цепочки не ребёнок.

Два ребёнка подряд

final parent = Job<void>((ctx) async {
  final rows = await ctx.run(
    Job.deferred<int>((ctx) => ctx.abandonable(load)),
  );
  await ctx.run(Job.deferred<void>((ctx) => ctx.join(() => report(rows))));
});

Последовательность, которая принадлежит операции, строится из детей. Второй ctx.run стартует, когда первый вернул значение, родитель ждёт обоих, и его отмена доходит до того из них, который сейчас работает. Цепочка нужна для обратного случая, когда шаг живёт дольше операции или вне операции: «Продолжение ребёнка».

Источник, который ждёт своё продолжение

Продолжение стартует, когда его источник кончился вместе с уборкой, поэтому источник, который ждёт собственное продолжение, ждёт вечно:

late final Job<int> parsed;
final loaded = Job<String>((ctx) async {
  final text = await ctx.join(loadText);
  await parsed.value;
  return text;
});
parsed = loaded.then<int>((ctx, text) => int.parse(text));

parsed стартует, когда loaded кончилась, а loaded ждёт parsed. Не кончается ни одна, и loaded.cancel() тоже не возвращается. parsed.done на месте parsed.value ничего не меняет, как и перенос ожидания в уборку loaded.