Отмена

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

Метод	Если Job принимает отмену во время ожидания
ctx.abandonable(action)	Бросает Cancelled, не дожидаясь завершения операции.
ctx.join(action)	Ждёт операцию до конца, затем вместо успешного результата бросает Cancelled.
ctx.uncancellable(action)	Удерживает обычную отмену до завершения действия, затем возвращает его результат, не бросая Cancelled; код после вызова выполняется до следующей контрольной точки, и уже она бросает отмену.
ctx.pause(duration)	Бросает Cancelled сразу и отменяет свой таймер.
ctx.check()	Бросает Cancelled, если Job уже отменена или её правила больше не выполняются.

Контроллер добавляет к этому две вещи. Его очередь ждёт работающую Job, поэтому выбор контрольной точки решает ещё и то, когда стартует следующая Job. abandonable подходит для запроса, результат которого можно отбросить: запрос может продолжаться после завершения Job и старта следующей. join подходит для работы, которая должна закончиться до продвижения очереди: например, команды устройству или открытия устройства.

Вторая вещь: Job контроллера отменяется двумя способами. Первый — запрос: cancel(), отмена родителя, close(), срок, заданный через timeout. Запрос — это обычная отмена из таблицы: её придерживает секция uncancellable и отклоняет Job, созданная с cancellable: false. Другая отмена приходит от правил состояния самой Job, и её не останавливает ни секция, ни cancellable: false. Оба способа разобраны ниже.

Остановка самой операции

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

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

// Первая попытка: брошенные перемотки идут на устройстве.
Job<void> seek(Duration position) =>
    enqueue<Ready, void>(key: 'seek', policy: Policy.restart, (ctx) async {
      // Кончается в тот момент, когда следующая перемотка отменяет эту.
      await ctx.abandonable(() => _player.seek(position));
      ctx.emit(ctx.state.copyWith(position: position));
    });

restart отменяет работающую Job, и abandonable в тот же момент отпускает вызов: Job завершается, и начинается следующая. Отпущенная перемотка при этом всё ещё идёт на устройстве. Протяните ползунок через три позиции, и все три перемотки пойдут на устройстве одновременно: три начала, потом три конца. Что устройство из этого сделает, зависит от устройства. Состояние при этом всё равно говорит о последней позиции: до emit доходит только последняя Job, поэтому экран показывает позицию, которую просил пользователь, что бы ни сделало устройство на самом деле.

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

// Вторая попытка: устаревшая перемотка идёт до конца.
Job<void> seek(Duration position) =>
    enqueue<Ready, void>(key: 'seek', policy: Policy.restart, (ctx) async {
      // Дождаться конца перемотки: двух перемоток сразу не будет.
      await ctx.join(() => _player.seek(position));
      ctx.emit(ctx.state.copyWith(position: position));
    });

Двух перемоток сразу устройство больше не получает, но и о том, что перемотка устарела, ему никто не говорит. Перемотка, которую пользователь проскочил, идёт до конца; Job, стоявшую за ней в очереди, restart убирает до старта; и только потом начинается последняя перемотка. Job, дождавшаяся первой перемотки, всё равно завершается с Cancelled: дождаться вызова не значит засчитать его результат.

Токен

Job<void> seek(Duration position) =>
    enqueue<Ready, void>(key: 'seek', policy: Policy.restart, (ctx) async {
      final token = CancelToken();
      ctx.onCancel(token.cancel);
      // Дождаться остановки устройства до следующей перемотки.
      await ctx.join(() => _player.seek(position, cancelToken: token));
      ctx.emit(ctx.state.copyWith(position: position));
    });

ctx.onCancel(callback) связывает отмену Job с механизмом отмены самой операции. Здесь это CancelToken из API плеера, а не тип этого пакета. Колбэк вызывается синхронно, когда Job принимает отмену. Токен просит плеер остановить перемотку, а join ждёт, пока тот остановится, поэтому новая перемотка начинается сразу после остановки прежней, а не после её конца. Это зависит от того, действительно ли API плеера реагирует на токен. Если не реагирует, всё возвращается ко второй попытке: проскоченная перемотка идёт до конца, и только потом начинается следующая.

Плеер, который останавливает перемотку, бросая ошибку, отдаёт эту ошибку в join, а join бросает её как есть. Job всё равно кончается Cancelled, поэтому ошибка не становится ничьим исходом, и задача отвечает за неё, как за любой провал своего тела. О каждой проскоченной перемотке узнаёт хук onError контроллера, а затем onUnanswered просят за неё ответить, и по умолчанию он передаёт её дальше, в Solo.unansweredHandler или в зону. Приложение один раз, в main, называет остановку своего плеера:

Job.isCancellationError = (error) => error is SeekStopped;

Названная им ошибка доходит до onError и дальше не идёт. Поле разобрано в разделе «Ошибки отмены» страницы об ошибках пакета async_job. Тело может выпустить вместо ошибки плеера отмену: оберните вызов в try и первой строкой его catch поставьте ctx.check(). Такой catch показывает раздел «Спросить задачу» страницы об отмене пакета async_job.

ctx.onCancel возвращает функцию удаления колбэка.

Защита шага или всей задачи

Платёж, квитанция на экране и запись о платеже в журнале идут вместе. Если платёж прошёл, квитанция должна быть показана, а запись сделана, о чём бы Job ни попросили за это время.

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

// Первая попытка: платёж проходит, а чек нет.
Job<void> commit(String entry) => enqueue<Ready, void>((ctx) async {
  // Каждого вызова дожидаются, что бы ни случилось.
  final receipt = await ctx.join(() => payment.commit());
  ctx.emit(ctx.state.copyWith(receipt: receipt));
  await ctx.join(() => journal.write(entry));
});

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

Обычный await на вызовах не лучше: между ними стоит emit, а он контрольная точка и на отменённой Job бросает.

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

// Вторая попытка: emit внутри шага бросает.
Job<void> commit(String entry) => enqueue<Ready, void>((ctx) async {
  // Дождаться всего шага как одного вызова.
  await ctx.join(() async {
    final receipt = await payment.commit();
    ctx.emit(ctx.state.copyWith(receipt: receipt));
    await journal.write(entry);
  });
});

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

Шаг одной секцией

Job<void> commit(String entry) => enqueue<Ready, void>((ctx) async {
  await ctx.uncancellable(() async {
    final receipt = await payment.commit();
    ctx.emit(ctx.state.copyWith(receipt: receipt));
    await journal.write(entry);
  });
  // ...остальная работа Job, которую отмена по-прежнему может остановить.
});

Ручная отмена, отмена родителя, закрытие и срок удерживаются, пока выполняется действие uncancellable. Job эти запросы пока не принимает, поэтому контрольная точка внутри секции на них не бросает, и квитанция доходит до состояния. Колбэки отмены и передача отмены детям тоже откладываются.

Когда завершается внешняя секция, Job принимает удержанный запрос. Следующая контрольная точка бросает Cancelled; обычный код сразу после вызова ещё может выполниться, и остановить его там может ctx.check() первой строкой после секции. Поместите всю обязательную работу внутрь секции и всегда ожидайте её. Неожидаемая секция может пережить Job и потерять удержанный запрос. Секции могут быть вложенными.

Внутри секции ctx.heldCancellation называет удержанный запрос, тот же, что показывает Solo.pending. Он говорит, что придержано в момент чтения, а почему шаг сохраняет результат, что бы он ни показал, рассказывает страница ядра.

Вся задача

// Job, отклоняющая все запросы, которые ей позволено отклонить.
Job<void> flush() =>
    enqueue<Ready, void>(cancellable: false, (ctx) => device.flush());

cancellable: false у Job полностью отклоняет ручную отмену, отмену родителя и закрытие. timeout такой Job не дают: срок — это отмена, которую она отклонила бы, поэтому вызов, где есть и то и другое, бросает ArgumentError. Пока такая Job ждёт в очереди, queue.remove, queue.removeWhere, queue.clear и cancelAll() оставляют её на месте; с force: true они её убирают. close() тоже убирает её из очереди, а работающую дожидается.

Флаг действует, только когда Job держит движок: Job, которую создал job, а enqueueJob так и не поставил в очередь, отменяется как любая другая. Очередная Job, отклонившая cancel, стартует в свой черёд и доходит до конца, и await job.cancel() возвращается только тогда.

Правила состояния Job не удерживает ни секция, ни cancellable: false. Рабочий тип Job — первый аргумент типа у enqueue, на этой странице Ready. Этот тип и keepWhile отменяют Job, когда состояние перестаёт им подходить, в том числе внутри секции. Если состояние уйдёт из Ready, пока commit выше ждёт платёж, следующая строка секции бросит, и ни квитанции, ни записи не будет. Job, которая должна работать в любом состоянии, берёт рабочим типом весь тип состояния контроллера и обходится без keepWhile. О рабочем типе и о keepWhile говорит раздел «Состояние и правила» страницы о состоянии.

Обычный await и время жизни контекста

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

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

// Первая попытка: каждая половина ждёт так, как нужно было другой.
Job<void> upload(List<int> chunks) => enqueue<Ready, void>((ctx) async {
  ctx.onDispose(() async {
    // Сброс должен закончиться до продвижения очереди.
    await ctx.join(() => device.flush());
  });
  for (final chunk in chunks) {
    await device.write(chunk);
  }
});

Каждая половина ждёт так, как нужно было другой. Цикл ждёт обычным await, который ни на что не отвечает: close() во время второго куска из четырёх ждёт ещё и третий, и четвёртый. Уборка ждёт через join, а он принадлежит телу, и к началу уборки тело уже закончилось: join бросает StateError, и сброса не происходит. От исхода Job это не зависит: загрузка, закончившаяся Done, тоже оставляет буфер несброшенным. Об ошибке узнаёт хук onError контроллера, а затем onUnanswered просят за неё ответить, и по умолчанию он передаёт её дальше, в Solo.unansweredHandler или в зону; весь маршрут описан в разделе «Ответ за ошибку» страницы об ошибках.

Каждое ожидание на своём месте

Job<void> upload(List<int> chunks) => enqueue<Ready, void>((ctx) async {
  ctx.onDispose(() async {
    // Уборка идёт после тела, где методов ожидания уже нет:
    // здесь работает обычный await.
    await device.flush();
  });
  for (final chunk in chunks) {
    // Проверяет и до записи, и после неё.
    await ctx.join(() => device.write(chunk));
  }
});

Теперь цикл ждёт через join, и каждая запись стала контрольной точкой. close() во время второго куска из четырёх ждёт только этот кусок, затем join вместо результата бросает Cancelled, и третий с четвёртым не пишутся вовсе. Уборка ждёт обычным await, и буфер сбрасывается, чем бы загрузка ни кончилась.

Обычный await уместен там, где дождаться операции несмотря на отмену и нужно: в уборке и внутри секции uncancellable.

Задержка, записанная как await Future.delayed(...), тоже такой обычный await: отменённая Job досиживает её до конца, и close() ждёт вместе с ней. ctx.pause(duration) даёт задержку с контрольной точкой: такая задержка кончается в тот момент, когда Job принимает отмену, а таймер задержки отменяется вместе с ней. Три способа выждать время сравнивает раздел «Дать времени пройти» страницы о времени пакета async_job.

Не сохраняйте контекст для запуска работы после конца Job. Методы emit, run, each, abandonable, join, pause и uncancellable тогда бросят StateError. Чтения и check остаются доступны, когда Job закончилась, и остаются контрольными точками: после отмены они бросают тот же Cancelled, а после любого другого исхода по-прежнему спрашивают правила Job. Во время зарегистрированной уборки чтения состояния и операции тела недоступны. Захватите необходимые для освобождения ресурсы замыканием. Во время уборки остаются доступны log, job, регистрация уборки, disown и unattended. Сам log не бросает исключение из-за отмены или завершения Job.

Сведения об отмене

Собственная причина проносит через отмену данные приложения:

final class OutOfRange extends CancelReason {
  final Duration position;

  const OutOfRange(this.position);

  @override
  String get name => 'out of range';
}

// Снаружи Job:
await job.cancel(reason: OutOfRange(position));

// Внутри её тела:
throw Cancelled.by(reason: OutOfRange(position));

// И на выходе:
switch (job.outcome) {
  case Cancelled(reason: OutOfRange(:final position)):
    print('out of range at $position');
  case Cancelled(:final reason, :final started):
    print('cancelled by ${reason.name}, started: $started');
  case _:
}

Cancelled содержит reason, started, необязательный description и стек вызовов отмены. started: false означает, что тело не запускалось. Причины наследуют CancelReason, а встроенные — вот эти; последний столбец показывает, как печатается Cancelled с каждой из них:

Причина	Когда	Как печатается
ManualCancelReason	cancel(), cancelAll(), queue.remove, queue.removeWhere или queue.clear, вызванные без причины	Cancelled(manual)
ParentCancelReason	отменили родителя, и отмена дошла до ребёнка	Cancelled(parent)
BodyCancelReason	тело сдалось: само бросило Cancelled или выпустило отмену ребёнка	Cancelled(body: why) для throw Cancelled('why'); для отмены ребёнка — сам ребёнок и то, чем он кончился: Cancelled(body: Job(_Op.upload): Cancelled(rules: canStart))
ChainCancelReason	отмена пришла по цепочке then	Cancelled(chain)
SiblingCancelReason	другая ветка ctx.runAll упала или была отменена либо сбой случился у самой группы, и группа остановила остальные	Cancelled(sibling)
RulesCancelReason	состояние вышло из рабочего типа, canStart или keepWhile ответили «нет» либо ctx.stateAs нашёл другой тип	Cancelled(rules: is not Ready), Cancelled(rules: canStart), Cancelled(rules: keepWhile)
ClosedCancelReason	close() или Job, поставленная в очередь после него	Cancelled(closed)
DuplicateCancelReason	Policy.droppable нашёл Job с тем же ключом и вернул её	Cancelled(duplicate)
ReplacedCancelReason	Policy.keepLatest или Policy.restart поставили на её место более новую Job с тем же ключом	Cancelled(replaced)
TimeoutCancelReason	истёк срок, заданный через timeout	Cancelled(timeout)

Проверяйте тип, как первая ветка switch выше; name служит подписью, а не ключом для сравнения. При передаче отмены между Job исходная отмена сохраняется в поле cause причины.

job.whenCancelled(callback) регистрирует синхронный слушатель и возвращает функцию его удаления. Слушатель вызывается:

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

Успешные и провалившиеся Job удаляют слушатели без вызова.

То, что слушатель бросил, идёт тем же путём, что ошибка уборки: в хук onError, а затем в onUnanswered, который по умолчанию передаёт её в Solo.unansweredHandler или в зону. Так же идёт и то, что бросил колбэк ctx.onCancel. У async-слушателя future никто не ждёт, и её ошибка уходит прямо в зону, мимо хука.

Срок задачи

Перемотка, которую устройство не закончило за две секунды, должна остановиться, а экран должен показать, что плеер недоступен. Перемотка, которую пользователь проскочил, оставляет состояние как есть.

Job<void> seek(Duration position) => enqueue<Ready, void>(
  key: 'seek',
  policy: Policy.restart,
  timeout: const Duration(seconds: 2),
  ifCancelled: (state, cancelled) =>
      cancelled.reason is TimeoutCancelReason ? const Offline() : state,
  (ctx) async {
    final token = CancelToken();
    ctx.onCancel(token.cancel);
    await ctx.join(() => _player.seek(position, cancelToken: token));
    ctx.emit(ctx.state.copyWith(position: position));
  },
);

timeout даёт задаче собственный срок. Когда срок выходит, задача отменяется так же, как её отменяет restart: токен останавливает перемотку, join ждёт устройство, и задача кончается Cancelled(timeout). Срок устроен как отмена async_job с причиной TimeoutCancelReason, и разбирает его раздел «Срок» на странице о времени в async_job.

Вышедший срок отличается от TimeoutException, который бросает Future.timeout. С .timeout(...) на вызове _player.seek исключение бросается в тело, задача кончается Failed, и состояние отображает ifFailed: у enqueue. timeout: у enqueue кончает задачу Cancelled, поэтому состояние отображает ifCancelled:, а ifFailed: не вызывается. ifCancelled: получает любую отмену. Чтобы отличить срок от проскоченной перемотки, обработчик ветвится по причине, как тот, что выше. Проскоченной перемотке он возвращает полученный state, а тот же объект ничего не публикует. value такой задачи бросает этот Cancelled, и ветка on TimeoutException вокруг него ничего не ловит.

Срок считается от старта тела, а не от вызова: время, которое задача ждёт в очереди, не считается. У collect и accumulate timeout задаёт срок каждой задаче, которую они ставят в очередь, и считается так же, от старта обработчика. Окно, которое группа пережидает до старта обработчика, в срок тоже не входит. Срок отменяет только свою задачу: задачи, стоящие за ней в очереди, остаются, и следующая стартует, когда эта закончит уборку. Вызов с Policy.droppable, который находит живую задачу со своим ключом, возвращает её, и timeout вызова пропадает вместе с остальным вызовом.

Отмена работы и закрытие контроллера

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

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

// Первая попытка: close() отменяет пачки в очереди.
// Экран отправляет пачку на каждое событие...
void onEvent(String batch) => logs.send(batch);

// ...и уходит, пока первая пачка в пути.
Future<void> onLeave() => logs.close();

Три события отправили три пачки, и экран уходит, пока первая в пути. close() отменяет все Job в очереди с Cancelled(closed) и запрашивает отмену работающей. Вторая и третья пачки не уходят вовсе. Первая уходит, потому что уже была в пути, но её исход всё равно Cancelled(closed): Job приняла отмену раньше, чем пачка дошла, и исход записывает это, а не доставку.

Три способа остановиться

// Очистить очередь и отменить работающую Job; контроллер открыт.
await logs.cancelAll();

// То же самое, и после этого новое не принимается.
await logs.close();

// Либо сначала выполнить то, что уже в очереди, и закрыться после.
await logs.close(mode: SoloCloseMode.drain);

cancelAll() удаляет отменяемые Job из очереди, запрашивает отмену работающей Job и ждёт её завершения. Job, которые он завершил, несут Cancelled(manual) или причину, переданную как cancelAll(reason: ...). Контроллер продолжает принимать работу. cancelAll(force: true) удаляет из очереди и неотменяемые Job; force не влияет на то, принимает ли отмену работающая Job.

close() прекращает приём работы, отменяет все Job в очереди с Cancelled(closed) и запрашивает отмену работающей Job. Он ждёт её завершения, включая детей и освобождение ресурсов, даже если отмена отклонена; чего он ждёт, говорит pending, как показано в разделе «Что удерживает контроллер» страницы о диагностике. Повторные вызовы возвращают ту же future. Новые попытки добавления возвращают уже отменённые Job без исключения, поэтому вызывающему коду не нужна проверка isClosed перед добавлением.

SoloCloseMode.drain закрывает контроллер, выполняя очередь, а не сбрасывая её, и это всё, чего не хватает первой попытке. С ним три пачки уходят по порядку, каждая заканчивается Done, а future, которую вернул close(), завершается после третьей. С момента вызова новые корневые Job не принимаются; вызов с Policy.droppable, чей ключ держит Job из этой очереди, получает эту Job, как до закрытия: это не новая работа. Стоящие в очереди Job выполняются по обычным правилам: по порядку, со своими детьми, своей уборкой и с выдержанным окном накопления, если оно есть. Пока эта очередь не опустеет, isDraining остаётся true. Обычный close() поверх идущего дренажа останавливает его, и future, которую вернул close(mode: SoloCloseMode.drain), завершается после этого. Выполнить очередь не значит обещать доставку: Job, которую выполняет дренаж, всё ещё может упасть или быть отклонена своими правилами. Буфер, хранящий события до подтверждения отправки, строится поверх дренажа, а не внутри него.

Закрытие само не публикует никакого состояния: контроллер остаётся на том, которое опубликовали последним, а когда закрытие завершилось, это состояние уже не изменить. Ресурсы приложения оно тоже не освобождает. Уборку и состояние, на котором заканчивает экран, оформите методом контроллера и дождитесь его до close(). Так сделано в разделе «Дождаться освобождения» примера камеры: там dispose() закрывает устройство и публикует Disposed. Обработчик состояния отменённой Job ещё может обновить состояние при закрытии.

Закрытие из задачи

Выход из аккаунта заканчивает сессию: вызов на сервер, а затем контроллер закрывается.

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

// Первая попытка: задача ждёт close(), а close() ждёт задачу.
Job<void> logout() => enqueue<Ready, void>((ctx) async {
  await ctx.join(api.logout);
  await close();
});

Этот вызов не возвращается никогда. close() ждёт работающую Job, включая её детей и уборку, а работающая Job здесь эта самая, и она ждёт close(). Так же ждёт и cancelAll(), и любой из них, ожидаемый из уборки Job. Телу, которому нужно только закончить себя, вызов контроллера не нужен вовсе: оно возвращается или отменяет себя через throw Cancelled('reason'). Такой бросок отменяет Job так же, как cancel(): отмена передаётся её детям, и вызываются её колбэки ctx.onCancel.

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

// Вторая попытка: контроллер принимает работу до самого close().
Future<void> logout() async {
  await enqueue<Ready, void>((ctx) => ctx.join(api.logout)).done;
  await close();
}

Job делает вызов, а метод закрывает контроллер, когда эта Job закончилась. Этот вызов возвращается, но оставляет открытое окно: контроллер принимает работу вплоть до строки close(). Job, поставленная, пока идёт вызов на сервер, принимается, стартует, когда выход из аккаунта закончился, и отменяется уже в пути, так что API слышит вызов, которого никто не хотел. Тот же порядок держит dispose() примера камеры в разделе «Дождаться освобождения», но там окно остаётся пустым: работу ставит только вызывающий код, а он ждёт на этой самой строке.

Поставить Job в очередь и закрыть дренажом

Future<void> logout() async {
  enqueue<Ready, void>((ctx) => ctx.join(api.logout)).ignoreFailure();
  await close(mode: SoloCloseMode.drain);
}

Дренаж ждёт эту Job так же, а дверь закрыта со строки close. Job, поставленная, пока идёт вызов, возвращается с Cancelled(closed), до API не дойдя. Всё, что стояло в очереди раньше, выполняется в обеих версиях: Job выхода из аккаунта встаёт в конец и ждёт своей очереди.

ignoreFailure() стоит там, где вторая попытка читала done. Исхода этой Job никто не ждёт, и без ignoreFailure() неудачный выход из аккаунта дошёл бы до зоны необработанной ошибкой: Solo.unansweredHandler, который ответил бы за него, не задан. Хук onError контроллера слышит этот сбой в обеих версиях, и контроллер закрывается в любом случае.