Задачи и очередь

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

Контроллер профиля

Страницы этого пакета строятся на одном контроллере: контроллере профиля из раздела «Быстрый старт» README пакета с тем, что добавляет к нему контроллер в приложении. ProfileApi возвращает имя после небольшой задержки; настоящий клиент бросает ApiException, когда вызов не удался:

import 'package:solo/solo.dart';

sealed class ProfileState {
  const ProfileState();
}

final class Initial extends ProfileState {
  const Initial();
}

final class Loading extends ProfileState {
  const Loading();
}

final class Loaded extends ProfileState {
  final String name;

  const Loaded(this.name);
}

final class Failure extends ProfileState {
  final Object error;

  const Failure(this.error);
}

final class ApiException implements Exception {
  final String message;

  const ApiException(this.message);

  @override
  String toString() => 'ApiException: $message';
}

class ProfileApi {
  Future<String> fetchName() =>
      Future.delayed(const Duration(milliseconds: 20), () => 'Ada Lovelace');
}

final class ProfileController extends Solo<ProfileState> {
  final ProfileApi api;

  ProfileController(this.api) : super(const Initial());

  Job<String> load() => enqueue<ProfileState, String>(
    key: 'load',
    policy: Policy.droppable,
    ifFailed: (state, error, stackTrace) => Failure(error),
    isExpectedError: (error) => error is ApiException,
    ifCancelled: (state, cancelled) => const Initial(),
    (ctx) async {
      ctx.emit(const Loading());
      final name = await ctx.abandonable(api.fetchName);
      ctx.emit(Loaded(name));
      return name;
    },
  );
}

enqueue<ProfileState, String> называет два типа. ProfileState — рабочий тип W задачи: состояния, с которыми работает её тело; он может быть уже типа состояния контроллера S. String — тип результата задачи. Если их опустить, как в быстром старте, тип результата — то, что возвращает тело, а рабочим типом задачи служит весь S; но в теле ctx.state тогда имеет тип Object?, поэтому тело, которое его читает, называет оба типа. Остальное — то, чего в быстром старте нет:

Параметр	Что делает	Где объяснено
key: 'load', policy: Policy.droppable	Второй load(), пока первая загрузка стоит в очереди или выполняется, получает её задачу	Очередь и политики
ifFailed, ifCancelled	Состояние для публикации, когда задача падает, — Failure, или когда начатую задачу отменяют, — Initial	Состояние после ошибки или отмены
isExpectedError	Провал с ApiException, который никто не читает, заканчивается на Failure; любая другая ошибка, которую никто не читает, идёт дальше в зону	Назвать ожидаемые ошибки

Задачи и результаты

Job<T> представляет одну операцию и её будущий исход. Метод контроллера ставит работу в очередь и отдаёт хэндл; profile.load() ниже взят из раздела «Контроллер профиля». await у done или value хэндла читает, чем операция кончилась, а не запускает её. Хэндл при этом не Future: Job можно отменить, а её ошибка доходит до контроллера независимо от того, ждёт ли её кто-нибудь.

switch (await profile.load().done) {
  case Done(:final value):
    print('loaded $value');
  case Failed(:final error):
    print('not loaded: $error');
  case Cancelled(:final reason):
    print('gave up: $reason');
}

У Outcome<T> эти три варианта: Done содержит значение, Failed хранит ошибку и стек вызовов, Cancelled называет причину отмены.

Член API	Назначение
job.value	Дождаться T; бросает исключение при ошибке или отмене.
job.done	Дождаться Outcome<T> без исключения.
job.outcome	Прочитать исход синхронно; null, пока Job не закончилась.
job.cancel()	Запросить отмену и дождаться завершения.
job.ignoreFailure()	Не сообщать об ошибке как о необработанной, когда исхода Job никто не ждёт; ожидаемые ошибки задача вместо этого называет в своём isExpectedError.
job.whenCancelled(callback)	Узнать об отмене в тот момент, когда Job её приняла, работает она или ещё стоит в очереди.
job.whenFinished(callback)	Узнать о конце с исходом, не наблюдая ошибку.

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

Каждая ошибка передаётся хуку onError контроллера и хуку onError у Solo.observer. Если к исходу провалившейся Job никто не обратился через done, value или ignoreFailure() до микрозадачи после её завершения, за ошибку также отвечают, если только Job не называет её в isExpectedError. onUnanswered контроллера отдаёт её Solo.unansweredHandler, если он задан, а иначе передаёт как необработанную в зону Dart, где Job была создана. Используйте ignoreFailure(), когда достаточно обработки ошибок в другом месте: отсутствие await само по себе не помечает ошибку обработанной, и чтение outcome тоже. Подробности есть в разделах «Обработанные и необработанные ошибки» и «Назвать ожидаемые ошибки» страницы об ошибках.

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

Создание и запуск задач

enum _Op { load, save, zoom, seek, stop, pause, shoot }

// Сначала собрать, поставить потом: два шага, если Job нужна в руках.
final saving = job<Ready, void>(
  key: _Op.save,
  (ctx) => ctx.join(store.save),
);
enqueueJob(saving);

// Либо всё сразу: так обычно и делает метод контроллера.
Job<void> setZoom(double zoom) => enqueue<Ready, void>(
  key: _Op.zoom,
  // Лениво и только для диагностики: строится, когда спросит лог.
  describe: () => 'zoom: $zoom',
  (ctx) => ctx.join(() => camera.zoom(zoom)),
);

Код этого раздела и следующих принадлежит одному контроллеру камеры, Solo<CameraState>. Два аргумента типа у job и enqueue задают рабочий тип и тип результата, как в разделе «Контроллер профиля»; рабочий тип здесь — Ready, одно из состояний камеры. О рабочем типе говорит раздел «Состояние и правила» страницы о состоянии.

По key политики очереди ищут родственную работу, а describe задаёт подпись для логов, наблюдателей и toString(): Job(key: label). Без описания используется представление Job(key).

job, enqueueJob и enqueue возвращают SoloJob<T>, который реализует Job<T> и добавляет isQueued. Все три защищённые, как и collect с accumulate: контроллер предоставляет предметные методы вроде load() или setZoom(), и вызывающий код видит их, а не члены, из которых они собраны. Предметный метод объявляет Job<T>, как setZoom выше: вызывающему коду этого хватает, чтобы дождаться результата или отменить Job. SoloJob<T> он объявляет только для вызывающего кода, который читает isQueued. Приватный метод, который делает ребёнка для ctx.run, тоже объявляет SoloJob<T>: ctx.run берёт ChildJob<T>, а SoloJob<T> им является, Job<T> — нет.

Job, которую запускает очередь, выполняет тело в зоне Dart того вызова, который её создал: enqueue или, как у saving выше, job, но не enqueueJob. Зона не зависит от того, что разбудило очередь: Job, поставленная прямо перед ней из другой зоны, конец предыдущей Job или таймер окна накопления. Таймеры тела и значения его зоны, срок Job, хуки onStart и ошибки future, которых тело не ждёт, остаются в этой зоне. Туда же уходит и ошибка Job, если её исход никто не прочитал и за неё больше никто не ответил. Внутри ctx.unattended это зона, откуда работу запустили, как и в ядре. Ребёнка запускает не очередь: он работает там, где его запустил ctx.run, как любая Job.deferred.

Очередь и политики

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

// sequential, по умолчанию: одна за другой, в порядке обращения.
Job<void> save() =>
    enqueue<Ready, void>(key: _Op.save, (ctx) => ctx.join(store.save));

// droppable: повторная загрузка того же профиля вернёт первую Job.
Job<Profile> load(String id) => enqueue<Ready, Profile>(
  key: (_Op.load, id),
  policy: Policy.droppable,
  (ctx) => ctx.abandonable(() => api.load(id)),
);

// keepLatest: ожидающий зум уходит, работающий остаётся нетронутым.
Job<void> setZoom(double zoom) => enqueue<Ready, void>(
  key: _Op.zoom,
  policy: Policy.keepLatest,
  (ctx) => ctx.join(() => camera.zoom(zoom)),
);

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

Политика	Когда Job с таким ключом уже существует
Policy.sequential	Добавить новую Job в очередь.
Policy.droppable	Вернуть существующую Job, из очереди или работающую корневую; новую отбросить.
Policy.keepLatest	Добавить новую Job в очередь, затем удалить из неё отменяемые Job с этим ключом: их отмена уже видит новую в очереди.
Policy.restart	Сделать то же, что keepLatest, и запросить отмену работающей корневой Job с этим ключом.

Корневая Job удерживает очередь, пока не закончатся её тело, её дети, освобождение ресурсов и обработчик состояния, то есть ifFailed или ifCancelled, которые ей дали. До тех пор другая корневая Job не начинается. Дети, которых она ждёт, её собственные: они работают внутри неё и в очередь не встают; они разобраны на странице «Дочерние задачи и цепочки». Политика не касается дочерних Job.

Job, которую убирают keepLatest и restart, кончается Cancelled(replaced) с причиной ReplacedCancelReason: и зум из очереди, который удаляет keepLatest, и работающая перемотка, которую отменяет restart. droppable отбрасывает новую Job, и она кончается Cancelled(duplicate), как показано ниже. Для droppable работающая Job считается существующей всё время, пока её не отменили. Та, что уже приняла отмену, сворачивается и работу не сделает, поэтому вызов, который её застал, ставит в очередь свою Job.

restart запрашивает отмену при добавлении новой Job. Новая Job встаёт в хвост очереди, как любая другая, за тем, что поставили раньше, и всё равно ждёт завершения текущей: их тела не пересекаются. Поэтому Job, которая отклоняет отмену, то есть созданная с cancellable: false, может задержать свою замену. Так же её задержит та, которой отмену некому передать: join вокруг вызова, до которого запрос не доходит, дождётся его до конца, а исход всё равно будет Cancelled. Токен выше и передаёт отмену операции. Второй способ даёт ctx.abandonable: он годится для операции, которую можно оставить доигрывать самой. Оба разобраны в разделе «Остановка самой операции» страницы об отмене.

Любой вызывающий ключа droppable может получить ту же Job. Поэтому простой then на ней бросает StateError, и на первом вызове тоже: then одного вызывающего не оставил бы следующему, чем её продолжить. Вместо него каждый вызывающий продолжает её через shared(), load('ada').shared().then(...): то, что вернул shared(), кончается вместе с Job, а его отмена Job не трогает. См. раздел «Ручка на каждую ветку».

Ключ, который droppable делит с другой политикой, возвращает и Job этой политики, и с этого момента простой then на ней тоже бросает. then, который её вызывающий взял раньше, остаётся, и тогда бросает shared() у вызывающего droppable: ему остаются done и value. Дайте droppable свой ключ.

Ключ на каждый запрос

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

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

// Первая попытка: один ключ на все профили.
Job<Profile> load(String id) => enqueue<Ready, Profile>(
  key: _Op.load,
  policy: Policy.droppable,
  (ctx) => ctx.abandonable(() => api.load(id)),
);

Один член _Op на метод, как у save, setZoom и seek, и droppable, чтобы отбросить повторный запрос. Второй load('ada'), пока первый в пути, получает первую Job, как и задумано. Но ту же Job получает и load('grace'): ключ говорит load и ничего о том, чей профиль, поэтому для политики этот вызов тоже дубликат. Вызывающий код ждёт выданную ему Job и получает профиль Ады, а запрос за профилем Грейс не уходит.

Ключ-запись

Job<Profile> load(String id) => enqueue<Ready, Profile>(
  key: (_Op.load, id),
  policy: Policy.droppable,
  (ctx) => ctx.abandonable(() => api.load(id)),
);

Ключом может быть любой объект со сравнением через ==: запись (_Op.load, id) даёт политике идентичность одного запроса, а не операции вообще. droppable отбрасывает вторую загрузку того же профиля и пропускает загрузку другого.

Job, которую принёс второй вызов, не стартует: она сразу кончается с Cancelled(duplicate) и причиной DuplicateCancelReason. enqueue возвращает один хэндл и не говорит, чья это Job, поэтому метод, которому нужно это знать, сначала собирает Job, а потом сравнивает её с тем, что вернул enqueueJob:

int duplicates = 0;

Job<Profile> load(String id) {
  final mine = job<Ready, Profile>(
    key: (_Op.load, id),
    (ctx) => ctx.abandonable(() => api.load(id)),
  );
  final taken = enqueueJob(mine, policy: Policy.droppable);
  if (!identical(taken, mine)) {
    // `mine` отброшена и уже кончилась с `Cancelled(duplicate)`;
    // `taken` это загрузка, которая стояла до этого вызова.
    duplicates++;
  }
  return taken;
}

load возвращает taken, и оба вызывающих ждут один профиль. Метод, который вместо этого возвращает mine, сообщает второму вызывающему, что его вызов отброшен: тот получает Cancelled(duplicate), а не результат первого вызова.

Используйте разные ключи для разных операций и типов результата. Job<int>, отданная droppable под ключом, который уже занят Job<String>, бросает ArgumentError. Бросает она ещё до того, как задачу взяли: в очередь ничего не встало, ничего не отменилось, и тот же хэндл можно снова поставить в очередь, под Policy.sequential или позже, когда ключ освободится. Сам ключ задаётся при создании Job, поэтому само столкновение разводится там, где обе Job собираются. Удобный способ его не иметь даёт enum, где на каждый метод свой член, сам по себе или внутри записи. Политика, отличная от последовательной, с ключом null тоже бросает ArgumentError, и обе проверки работают в release-сборках.

Работа с очередью напрямую

Сама очередь принадлежит контроллеру, и метод может работать с ней напрямую:

Job<void> stop() {
  // Что ждёт прямо сейчас.
  print(queue.length);
  // Убрать то, что эта команда обесценила...
  queue.removeWhere((job) => job.key == _Op.zoom);
  // ...и пропустить саму остановку без очереди.
  return enqueueJob(
    job<Ready, void>(key: _Op.stop, (ctx) => ctx.join(camera.stop)),
    first: true,
  );
}

first пропускает только то, что ждёт, и ничего больше. Уже работающий зум не заберёт ни removeWhere, ни first: остановка стартует, когда этот зум закончится. Начавшую работу Job останавливает только отмена, и о ней страница «Отмена».

queue предоставляет jobs, length, isEmpty, isNotEmpty, remove, removeWhere и clear. lastJobWhere контроллера находит последнюю Job в очереди, которая проходит его проверку, а если такой нет, то работающую. Саму работающую корневую Job отдаёт current, а если контроллер простаивает, то null; current?.cancel() просит её остановиться. Методы удаления касаются только Job в очереди, работающая им не принадлежит, и сохраняют Job с cancellable: false, если не вызваны с force: true. Удалённая Job заканчивается Cancelled(manual) или с причиной, переданной в reason:. Причину принимает и cancelAll, который очищает очередь и заодно отменяет работающую Job. queue, lastJobWhere и current защищены: они для методов контроллера, а вызывающему снаружи доступен cancelAll().

Порядок в jobs не совпадает с порядком запуска. Job, ждущую окна накопителя, может обойти готовая, стоящая за ней, так что голова очереди не всегда стартует следующей. См. раздел «Выбор момента готовности группы» страницы о правилах накопления.

Пауза очереди

Очереди нужно на время встать: ничего не стартует, пока контроллеру не скажут продолжать, а то, что попросили за это время, идёт потом, в том порядке, в каком просили. Своей паузы у очереди нет, но задача, которая ждёт Completer, держит очередь не хуже любой работы. Эта задача, ворота, работает в CameraState, то есть во всём типе состояния контроллера, тогда как остальные задачи страницы работают в Ready. Задача кончается Cancelled, когда состояние уходит из её рабочего типа, а пауза не должна зависеть от одного состояния.

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

// Первая попытка: ворота ждут за задачами из очереди.
Completer<void>? _gate;

bool get isPaused => _gate != null;

void pause() {
  if (_gate != null) return;
  final gate = _gate = Completer<void>();
  enqueue<CameraState, void>(
    key: _Op.pause,
    (ctx) => ctx.abandonable(() => gate.future),
  );
}

void resume() {
  _gate?.complete();
  _gate = null;
}

enqueue и есть тот вызов, которым метод обычно ставит задачу, и ворота он ставит туда же, куда всё остальное: в хвост. Три задачи, которые уже стояли в очереди, идут до начала паузы, а за воротами ждёт только то, что добавили после pause().

Уйти ворота могут и без resume: их отменяет cancelAll, и close тоже. Поле сбрасывает один resume, поэтому после cancelAll isPaused продолжает отвечать true, пока очередь идёт, а следующий pause() выходит на первой строке.

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

// Вторая попытка: ворота, снятые до старта, остаются в поле.
void pause() {
  if (_gate != null) return;
  final gate = _gate = Completer<void>();
  enqueueJob(
    job<CameraState, void>(key: _Op.pause, (ctx) async {
      // Отмена ещё не resume, а отменённые ворота уже не пауза.
      ctx.onCancel(() => _gate = null);
      await ctx.abandonable(() => gate.future);
    }),
    first: true,
  );
}

Поле, isPaused и resume() остаются прежними. first ставит ворота впереди всего, кроме работающей задачи, поэтому ждёт и то, что уже стояло в очереди, и то, что добавят позже. А ctx.onCancel сбрасывает поле, когда ворота отменяют, если к этому времени они уже работают. Этот колбэк регистрирует тело, а тело ещё не запускалось, пока ворота ждут своей очереди за работающей задачей. queue.clear(), cancelAll() или close() в этом окне снимают ворота с очереди до старта. Колбэка нет, поле осталось, и isPaused снова отвечает true над очередью, которая идёт.

Хэндл ворот

void pause() {
  if (_gate != null) return;
  final gate = _gate = Completer<void>();
  enqueueJob(
    job<CameraState, void>(
      key: _Op.pause,
      (ctx) => ctx.abandonable(() => gate.future),
    ),
    first: true,
  ).whenCancelled((_) {
    // Отмена ещё не resume, а отменённые ворота уже не пауза.
    // После `resume()` и ещё одного `pause()` в поле стоят новые ворота,
    // и они остаются.
    if (identical(_gate, gate)) _gate = null;
  });
}

whenCancelled регистрируется на хэндле, который вернул enqueueJob, а не в теле, поэтому слышит отмену и тех ворот, что не стартовали, и тех, что работают. На Job, которая уже отменена, а именно такую enqueueJob возвращает у закрытого контроллера, колбэк вызывается сразу. Колбэк сбрасывает поле, только пока в поле стоят эти самые ворота: после resume() и нового pause() там уже следующие, и конец старых ворот не должен их сбросить. Случаи:

Что делаешь	Что происходит
pause(), потом три задачи	Все три стоят, ни одного исхода.
pause(), когда три уже стоят в очереди	Они тоже ждут: ворота встают впереди всего, кроме работающей задачи.
resume()	Идут в том порядке, в каком добавлены.
queue.clear(), пока ворота работают	Очередь пустеет, пауза остаётся: ворота сейчас текущая задача, а не стоящая в очереди.
queue.clear(), пока ворота ещё ждут за работающей задачей	Ворота уходят вместе с остальной очередью, и поле сброшено: isPaused отвечает false, и pause() снова работает.
close() под паузой	Возвращается без resume: ворота отменяемы, и их ctx.abandonable кончается в момент отмены.
close(mode: SoloCloseMode.drain) под паузой	Ждёт resume: дренаж не отменяет то, что стоит в очереди, а выполняет, и ворота тоже.
cancelAll()	Ворота уезжают вместе со всем остальным, и очередь идёт дальше, хотя их никто не открывал. Поле сброшено, и pause() снова работает.
Состояние уходит из Ready под паузой	Пауза остаётся: ворота работают в любом состоянии камеры.

pause() не зря ничего не возвращает. Хэндл, который отдаёт enqueueJob, принадлежит воротам и завершается тогда, когда пауза кончается, а не когда начинается: Done на resume, Cancelled(closed) на close. Вернуть его значит позвать привычку, которой учат остальные методы страницы, и тогда await pause() ждёт того самого resume, который вызывающий и собирался сделать.

Чего движок не знает, так это того, что эта задача служит паузой. Наблюдатель видит старт обычной задачи, а pending отвечает воротами, а не работой, так что экран, показывающий текущее, покажет ворота. Признак «на паузе» держишь у себя, и у этого контроллера его держит isPaused.

canStart паузы не делает. Задача, чьё правило не подошло к состоянию, не придерживается: она отменяется, когда приходит её очередь, с Cancelled(rules: canStart), и очередь не наполняется, а пустеет. Пауза, построенная на нём, теряет работу молча.

Отложенный старт

У камеры есть автоспуск: shootIn(delay) делает снимок, когда задержка вышла, а камера тем временем принимает команды — зум, заданный во время отсчёта, ставится сразу. cancelAll() останавливает отсчёт, как и остальную работу, и закрытие контроллера тоже.

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

// Первая попытка: отсчёт держит очередь.
Job<void> shootIn(Duration delay) =>
    enqueue<Ready, void>(key: _Op.shoot, (ctx) async {
      await ctx.pause(delay);
      await ctx.join(camera.shoot);
    });

ctx.pause — то ожидание, которое ядро даёт задаче, и его кончают и cancelAll(), и close(). Но задача, которая ждёт, держит очередь не хуже той, что работает, и ровно для этого служат ворота из «Паузы очереди». При задержке в десять секунд зум, попрошенный через секунду после начала отсчёта, стартует только после снимка, на девять секунд позже.

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

// Вторая попытка: отсчёт не принадлежит ни одной задаче.
void shootIn(Duration delay) => Timer(delay, shoot);

Job<void> shoot() =>
    enqueue<Ready, void>(key: _Op.shoot, (ctx) => ctx.join(camera.shoot));

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

Отсчёт, которым владеет контроллер

Job<void>? _countdown;

void shootIn(Duration delay) {
  if (isClosed) return;
  _countdown?.cancel().ignore();
  final countdown = _countdown = Job<void>((ctx) => ctx.pause(delay));
  countdown.whenFinished((outcome) {
    if (identical(_countdown, countdown)) _countdown = null;
    if (outcome is Done) shoot();
  });
}

@override
Future<void> cancelAll({
  bool force = false,
  CancelReason reason = const ManualCancelReason(),
}) async {
  final countdown = _countdown?.cancel(reason: reason);
  await super.cancelAll(force: force, reason: reason);
  await countdown;
}

@override
void onClose() => _countdown?.cancel().ignore();

Отсчёт — отдельная Job, созданная конструктором ядра, а не через enqueue: она не ждёт своего хода и ни одну очередь не держит. ctx.pause — то же ожидание, что в первой попытке. Старт по сигналу, а не через время, меняет только тело отсчёта: вместо ctx.pause оно ждёт сигнал, например Job<void>((ctx) => ctx.abandonable(() => flash.ready)), где flash.ready — future, который завершается, когда вспышка зарядилась. Когда отсчёт кончается Done, whenFinished ставит снимок в очередь, а отменённый отсчёт не ставит ничего. shoot() — метод из второй попытки.

Контроллер об отсчёте не знает, поэтому класс говорит ему о нём в двух местах. Переопределённый cancelAll отменяет отсчёт с той причиной, которую получил, а потом делает свою работу через super, который очищает очередь и отменяет работающую задачу; возвращённый им future завершается, когда кончился и отсчёт. onClose отменяет отсчёт, когда контроллер закрылся, каким бы режимом его ни закрыли. После этого отменить новый отсчёт уже некому, и проверка в начале shootIn не даёт ему стартовать.

Поле сбрасывается, только пока в нём стоит этот самый отсчёт: второй shootIn отменяет первый и ставит в поле свой, и конец первого не должен его сбросить. whenFinished слышит конец, не наблюдая провал. ctx.pause не падает, а flash.ready упасть может, и ошибка, пришедшая после отмены отсчёта, уходит в зону мимо ignoreFailure(). Поэтому ошибку сигнала, который может упасть, ловят внутри ctx.abandonable и превращают в значение, а снимок делают только по этому значению. Случаи:

Что делаешь	Что происходит
shootIn на десять секунд, потом зум через секунду	Зум ставится сразу, снимок делается на десятой секунде.
Ещё один shootIn во время отсчёта	Отсчёт начинается заново: один снимок, через десять секунд после второго вызова.
cancelAll() во время отсчёта	Снимка нет, и таймера не остаётся.
queue.clear() во время отсчёта	Снимок делается: отсчёт в очереди не стоит.
close() во время отсчёта	Снимка нет, и ни один таймер не переживает контроллер.
close(mode: SoloCloseMode.drain) во время отсчёта	Очередь дорабатывает, а снимка нет.
shootIn после close()	Ничего не стартует.

Если задержка одна на все вызовы, почти всё это даёт накопитель, и держать отсчёт не нужно: accumulate с AccumulationTiming.debounce(delay). Его группа выжидает окно, не держа очередь, повторный add начинает окно заново, а cancelAll() и close() группу снимают. С таблицей он расходится в двух строках: queue.clear() группу снимает, а дренирующий close() дожидается окна и делает снимок. Собственный отсчёт нужен для задержки, которая меняется от вызова к вызову, для старта по сигналу и для ожидания, которого queue.clear() касаться не должен. Как устроено окно — в разделе «Выбор момента готовности группы» страницы о правилах накопления.

Нажатие, которое не ждёт перезагрузки

Экран ленты перезагружает посты, на что у сервера уходит две секунды, а тем временем пользователь лайкает пост. Сердечко должно загореться сразу, лайк должен сразу дойти до сервера, а лайк и снятие лайка одного поста должны дойти до него в том порядке, в каком их нажали. Feed хранит посты, у каждого Post есть id и liked, а withLiked копирует ленту с одним изменённым постом. api — это FeedApi: его fetchFeed() отвечает через две секунды после запроса — с теми лайками, которые были на сервере, когда запрос пришёл, — а его setLiked отвечает через двадцатую долю секунды после запроса или бросает ApiException, когда запрос не удался.

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

// Первая попытка: лайк ждёт перезагрузку.
Job<Feed> reload() =>
    enqueue<Feed, Feed>(key: 'reload', policy: Policy.droppable, (ctx) async {
      final feed = await ctx.abandonable(api.fetchFeed);
      ctx.emit(feed);
      return feed;
    });

Job<void> like(String postId, {required bool liked}) =>
    enqueue<Feed, void>(key: ('like', postId), (ctx) async {
      ctx.emit(ctx.state.withLiked(postId, liked: liked));
      await ctx.join(() => api.setLiked(postId, liked: liked));
    });

У каждого метода свой ключ, и droppable отбрасывает вторую перезагрузку, пока первая в пути. Но ключ решает только, что второй вызов сделает с задачей с тем же ключом, и обойти другую задачу он не даёт: корневые задачи выполняются по одной. Лайк, нажатый через 100 мс после начала перезагрузки, начинается, когда перезагрузка закончилась. Сердечко загорается на 2000 мс, и тогда же лайк доходит до сервера.

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

// Вторая попытка: лайк идёт мимо очереди.
Future<void> like(String postId, {required bool liked}) async {
  externalSetState(currentState.withLiked(postId, liked: liked));
  await api.setLiked(postId, liked: liked);
}

externalSetState пишет состояние, не дожидаясь очереди, и запрос уходит сразу: сердечко загорается на 100 мс, а сервер подтверждает лайк на 150 мс. Но перезагрузка спросила сервер на 0 мс, до лайка, и на 2000 мс публикует то, что ей ответили: сердечко гаснет, хотя лайк на сервере есть, и зажжёт его только следующая перезагрузка. Та же гонка разобрана в разделе «Порядок обновлений общего состояния» на странице, которая сравнивает пакет с bloc.

Контроллер для лайков

/// Лайк, который экран показывает поверх ленты.
final class Mark {
  final bool liked;

  /// Номер подтверждения или `null`, пока лайк в пути.
  final int? confirmation;

  const Mark(this.liked, [this.confirmation]);
}

final class Marks {
  final Map<String, Mark> byPost;

  const Marks([this.byPost = const {}]);

  Marks put(String postId, Mark? mark) => Marks({
    for (final MapEntry(:key, :value) in byPost.entries)
      if (key != postId) key: value,
    postId: ?mark,
  });
}

final class LikesController extends Solo<Marks> {
  final FeedApi api;
  int _confirmations = 0;

  LikesController(this.api) : super(const Marks());

  /// Сколько лайков сервер уже подтвердил.
  int get confirmations => _confirmations;

  Job<void> setLiked(String postId, {required bool liked}) {
    Mark? before;
    return enqueue<Marks, void>(
      // Подтверждения нет: пост показывает то же, что до нажатия.
      ifFailed: (marks, error, stackTrace) => marks.put(postId, before),
      ifCancelled: (marks, cancelled) => marks.put(postId, before),
      isExpectedError: (error) => error is ApiException,
      (ctx) async {
        before = ctx.state.byPost[postId];
        ctx.emit(ctx.state.put(postId, Mark(liked)));
        await ctx.join(() => api.setLiked(postId, liked: liked));
        ctx.emit(ctx.state.put(postId, Mark(liked, ++_confirmations)));
      },
    );
  }

  /// Снимает отметки, которые принесёт перезагрузка, отправленная после
  /// [seen] подтверждений.
  Job<void> settle(int seen) => enqueue<Marks, void>((ctx) async {
    final kept = {
      for (final MapEntry(key: postId, value: mark) in ctx.state.byPost.entries)
        if (mark.confirmation! > seen) postId: mark,
    };
    if (kept.length < ctx.state.byPost.length) ctx.emit(Marks(kept));
  });
}

У лайков свой контроллер, и в его очереди стоят только короткие задачи, лайки и settle, поэтому лайк никогда не ждёт перезагрузку: это принцип раздела «Своя доска» страницы о состоянии. Его состояние — отметка для каждого поста, который нажал пользователь, и экран показывает отметку поверх того, что говорит лента. Лента сообщает лайкам, когда перезагрузка принесла слово сервера:

final class FeedController extends Solo<Feed> {
  final FeedApi api;
  final LikesController likes;

  FeedController(this.api, this.likes) : super(const Feed([]));

  Job<Feed> reload() =>
      enqueue<Feed, Feed>(key: 'reload', policy: Policy.droppable, (ctx) async {
        final seen = likes.confirmations;
        final feed = await ctx.abandonable(api.fetchFeed);
        ctx.emit(feed);
        likes.settle(seen);
        return feed;
      });
}

/// Показывает ли экран лайк у [post].
bool shownLiked(Post post, Marks marks) =>
    marks.byPost[post.id]?.liked ?? post.liked;

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

before — отметка, которая была у поста, когда лайк начался. Если лайк не прошёл или его отменили, ifFailed или ifCancelled возвращает её, и пост показывает то же, что до нажатия: неудачное снятие лайка возвращает подтверждённый лайк. Исход нажатия никто не читает, поэтому isExpectedError называет ApiException неудачного запроса, и дальше в зону он не идёт.

Одна очередь держит по порядку лайк и снятие лайка одного поста, а заодно и лайки разных постов: лайк второго поста ждёт, пока первый в пути. Для запроса в двадцатую долю секунды это приемлемо; работе, которая идёт долго, нужна своя очередь. Так же ждёт и второе нажатие на тот же пост, а с ним и его отметка: сердечко отвечает на нажатие, когда первый лайк подтвердится. Поэтому экран, который переключает сердечко, берёт новое значение из своего последнего нажатия, а не из shownLiked: третье нажатие во время ожидания прочтёт сердечко, которое второе нажатие ещё не повернуло, и повторит второе нажатие. Экран слушает оба контроллера. Во Flutter один виджет над обоими берёт Listenable.merge в ListenableBuilder, а в каждый контроллер подмешан SoloListenable из flutter_solo.

Отсчёт из раздела «Отложенный старт» подсказывает другой путь: ожидание сервера — в Job вне очереди, а в очереди только шаг, который публикует ленту. Лайк в том же контроллере тогда уходит сразу, но перезагрузка всё равно публикует то, что было на сервере до лайка, и отметки нужны всё равно. ctx.unattended не помогает: лайк, который его зовёт, сначала ждёт в очереди. Не помогает и дочерняя задача: она выполняется внутри тела задачи, которая уже идёт. Случаи:

Что вы делаете	Что происходит
Лайк во время перезагрузки	Сердечко загорается сразу, и лайк сразу доходит до сервера. Перезагрузка публикует ленту без лайка, и отметка остаётся поверх неё.
Следующая перезагрузка	Лента приносит лайк, и отметка уходит.
Лайк подтверждён, потом отправлена перезагрузка	Эта перезагрузка снимает отметку.
Лайк и снятие лайка одного поста	Сервер получает их в этом порядке.
Лайк не прошёл	Пост показывает то же, что до нажатия, и в зону ничего не уходит.
Лайк отменили	Когда его запрос ответит, пост показывает то же, что до нажатия, пока перезагрузка не принесёт слово сервера.
Второе нажатие на пост, пока его лайк в пути	Оно ждёт, пока первый лайк подтвердится, и тогда сердечко на него отвечает.
Лайк другого поста во время медленного	Он ждёт, пока первый подтвердится.