solo и bloc рядом

Этот документ сравнивает solo с Bloc и Cubit на одиннадцати сценариях контроллеров. Каждый раздел описывает нужное поведение, начинает с кода, к которому ведёт привычка или словарь самого API, и говорит, что этот код делает на самом деле. Только потом он показывает версии на bloc и на solo, которые требование выполняют, вместе с тем, что делает библиотека, а что остаётся кодом приложения.

Этот код и есть первая попытка раздела, а состояния и трассы под ней получены её запуском. Она написана на bloc, как и ответ под заголовком про bloc после неё: этот заголовок добавляет версию, которая требование выполняет. Там, где одной попытки до этого не хватает, за ней идёт вторая под своим заголовком. Прочитать попытки до ответа и есть смысл раздела. Тот, кто ловушку уже знает, может сразу перейти к заголовку про bloc и к ### Solo или к строке, которой раздел кончается, Что даёт solo.

Сценарии по порядку — что пробует в каждом первая попытка и что берёт версия на solo:

Сценарий	Что пробует первая попытка	Что берёт solo
Порядок обновлений общего состояния	Тип события на команду и sequential() у каждого из двух обработчиков	Одна очередь контроллера
Ошибка наблюдателя при обновлении состояния	Наблюдатель и журнал в onChange, оба с вызовом super первой строкой	Solo.observer, чей сбой до Job не доходит
Закрытие и отмена начатой работы	Обработчик с sequential(), который публикует ответ и затем добавляет MarkReplyRead	close(), который отменяет работающую Job на ctx.abandonable
Выход из состояния загрузки при отмене работы	Сброс в Initial в блоке finally	ifCancelled и lastJobWhere, чтобы найти Job для отмены
Перезапуск одной операции в общей очереди	По регистрации на команду и restartable() у seek	Policy.restart у seek, с ctx.onCancel и ctx.join
Типизированные методы с выполнением через очередь	Методы Cubit, которые возвращают future	Методы, которые возвращают Job, с Policy.restart
Удаление выбранной ожидающей работы	Флаг ухода, который проверяет обработчик	queue.removeWhere и cancellable: false у того, что должно выполниться
Ожидание конкретного запроса	Completer на событии и droppable() у оплаты	Ключ на заказ с Policy.droppable и Job, которую ждёт вызывающий
Реакция на независимое внешнее изменение состояния	Одна регистрация на сборку и отзыв сессии	externalSetState и enqueue<SignedIn, void> у сборки
Завершение начатой записи перед перезапуском	restartable() у цикла загрузки	Policy.restart и ctx.join вокруг каждой записи
Освобождение ресурса, полученного после отмены	Проверка emit.isDone перед использованием результата	dispose у ctx.abandonable

Bloc обрабатывает события обработчиками, и каждый вызов on<E> составляет одну регистрацию: обработчик событий типа E и трансформер, который определяет порядок выполнения этих событий. Cubit предоставляет методы, напрямую меняющие состояние; очереди событий у него нет.

В solo метод контроллера обычно создаёт Job<T> и ставит его в очередь контроллера. Job из очереди называются корневыми и выполняются по одной. Вызывающий код может дождаться job.value, чтобы получить значение или исключение, либо job.done, чтобы получить исход Done, Failed или Cancelled. Политика очереди применяется при добавлении отдельной Job.

Остальные API solo вводятся по мере использования. README служит введением, а подробные контракты лежат на страницах, перечисленных в его разделе «Страницы». Приложение, которое остаётся на bloc, может поставить контроллеры этих разделов под свои bloc; страница «solo под bloc» показывает мосты, их цену и три раздела, где контроллер снимает с bloc обходной приём.

Примеры используют отдельные модели приложений: Ready плеера и Ready отчёта разные. Вспомогательные классы состояний, событий и фейковые API здесь не показаны: они лежат рядом с кодом, который извлекает эти примеры из документа, в tool/doc_snippets.py репозитория. Примеры проверены с bloc 9.2.1, bloc_concurrency 0.3.0 и локальным кодом solo 1.0.0. Трассы описывают эти запуски, а не гарантируют время работы произвольного устройства.

Соответствия

Основные соответствия API для того, кто знает bloc:

bloc	solo
Bloc<E, S>, Cubit<S>	Solo<S>; ValueListenable с подмешанным SoloListenable из flutter_solo
Класс события, on<E>, add(E())	Метод, возвращающий Job<T>
EventTransformer	Policy при добавлении Job
emit(next)	ctx.emit(next)
if (emit.isDone) return;	Контрольные точки отмены, например ctx.abandonable и ctx.check
emit.onEach, emit.forEach	ctx.each(stream, onData)
state, stream	currentState, addListener; stream с подмешанным SoloStream
BlocObserver	SoloObserver
BlocBuilder, BlocSelector	ValueListenableBuilder или SoloBuilder и SoloSelector; два последних лежат в flutter_solo
BlocListener для результата операции	Ожидание job.done этой операции; для новости, которая переживает экран, — уведомление, которое принимает экран
BlocProvider	Пакет этим сознательно не занимается: зависимости передают другими способами и пакетами, например scopo
close()	close()
blocTest	test и ожидание исхода Job

sequential, droppable и restartable соответствуют Policy.sequential, Policy.droppable и Policy.restart, но это не те же вещи под другими именами: трансформер распоряжается всеми событиями своей регистрации, а политика только Job с одним ключом. Разделы «Перезапуск одной операции в общей очереди», «Ожидание конкретного запроса» и «Завершение начатой записи перед перезапуском» показывают, где они расходятся. Policy.keepLatest удаляет ожидающую работу с тем же ключом, сохраняя работающую Job. Политики одновременного выполнения корневых Job нет; для независимой параллельной работы используйте детей или отдельные контроллеры, как в разделе «Нажатие, которое не ждёт перезагрузки».

Порядок обновлений общего состояния

Контроллер заметок отправляет заметку и обновляет список с сервера. Если обновление началось до конца отправки, ответ может содержать старый список. Публикация такого ответа после отправки удалит новую заметку из локального состояния, даже если оба обработчика читают актуальное локальное состояние. Нужно упорядочить целые операции, включая ввод-вывод.

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

Тип события на команду, обработчик на событие и на обоих трансформер, название которого и есть требование:

// Первая попытка: две очереди, и побеждает устаревший список.
class SplitNotesBloc extends Bloc<NotesEvent, NotesState> {
  final Api _api;

  SplitNotesBloc(this._api) : super(const NotesState()) {
    on<UploadNote>((e, emit) async {
      emit(state.copyWith(uploading: true));
      await _api.upload(e.note);
      emit(state.copyWith(notes: [...state.notes, e.note], uploading: false));
    }, transformer: sequential());
    on<RefreshList>((e, emit) async {
      final serverNotes = await _api.list();
      emit(state.copyWith(notes: serverNotes));
    }, transformer: sequential());
  }
}

На сервере лежит одна заметка, n0, а экран добавляет UploadNote(n1) и RefreshList сразу друг за другом. Итоговое состояние NotesState([n0], uploading: false): отправленной заметки в нём нет. Причину показывает порядок выполнения. Ниже записи самого сервера и каждая публикация с событием, которое её сделало; локальный список поначалу пуст, потому что его ещё не читали:

UploadNote emits [], uploading: true
upload n1 starts
list reads [n0]
server receives n1 and holds [n0, n1]
UploadNote emits [n1], uploading: false
RefreshList emits [n0]

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

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

У тех же двух регистраций есть и другой способ потерять заметку, и в нём трансформеры не участвуют вовсе. Запись в одну строку emit(state.copyWith(notes: await _api.list())) вычисляет получатель state до ожидания аргумента и публикует состояние, прочитанное, пока отправка ещё шла. Этот вариант заканчивается на NotesState([n0], uploading: true), затирая заодно и флаг отправки.

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

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

// Вторая попытка: одна регистрация, но без очереди.
class ConcurrentNotesBloc extends Bloc<NotesEvent, NotesState> {
  final Api _api;

  ConcurrentNotesBloc(this._api) : super(const NotesState()) {
    on<NotesEvent>((e, emit) async {
      switch (e) {
        case UploadNote(:final note):
          emit(state.copyWith(uploading: true));
          await _api.upload(note);
          emit(state.copyWith(notes: [...state.notes, note], uploading: false));
        case RefreshList():
          final serverNotes = await _api.list();
          emit(state.copyWith(notes: serverNotes));
      }
    });
  }
}

Состояние снова NotesState([n0], uploading: false), и порядок совпадает с приведённым выше строка в строку. Две попытки ошибаются по разным причинам. В первой две очереди, по одной на регистрацию, и друг друга они не ждут. Во второй регистрация одна, но очереди нет вовсе: трансформер по умолчанию запускает каждое событие сразу, не дожидаясь предыдущего. По результату их не различить.

Bloc

Обе половины вместе: одна регистрация общего типа и sequential() на ней:

class NotesBloc extends Bloc<NotesEvent, NotesState> {
  final Api _api;

  NotesBloc(this._api) : super(const NotesState()) {
    on<NotesEvent>((e, emit) async {
      switch (e) {
        case UploadNote(:final note):
          emit(state.copyWith(uploading: true));
          await _api.upload(note);
          emit(state.copyWith(notes: [...state.notes, note], uploading: false));
        case RefreshList():
          final serverNotes = await _api.list();
          emit(state.copyWith(notes: serverNotes));
      }
    }, transformer: sequential());
  }
}

Итоговое состояние NotesState([n0, n1], uploading: false), и в порядке выполнения сдвинулась одна строка:

UploadNote emits [], uploading: true
upload n1 starts
server receives n1 and holds [n0, n1]
UploadNote emits [n1], uploading: false
list reads [n0, n1]
RefreshList emits [n0, n1]

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

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

Solo

Job, которые создают методы одного контроллера, встают в одну его очередь:

final class NotesController extends Solo<NotesState> {
  final Api _api;

  NotesController(this._api) : super(const NotesState());

  Job<void> upload(Note note) => enqueue<NotesState, void>(key: 'upload', (
    ctx,
  ) async {
    ctx.emit(ctx.state.copyWith(uploading: true));
    await ctx.join(() => _api.upload(note));
    ctx.emit(
      ctx.state.copyWith(notes: [...ctx.state.notes, note], uploading: false),
    );
  });

  Job<void> refresh() => enqueue<NotesState, void>(key: 'refresh', (ctx) async {
    final serverNotes = await ctx.abandonable(_api.list);
    ctx.emit(ctx.state.copyWith(notes: serverNotes));
  });
}

ctx.join(action) вызывает операцию и ждёт её завершения. После успешного ответа он проверяет отмену перед возвратом результата. Отправка использует join, чтобы даже отменённая отправка держала очередь, пока сервер не ответит: очередь не запустит следующую Job, пока отправка продолжается. refresh использует ctx.abandonable, который может прекратить ожидание при отмене: в этом примере результат чтения разрешено отбросить.

Итоговое состояние тоже NotesState([n0, n1], uploading: false), и порядок тот же, что у bloc выше, только вместо типов событий в нём ключи Job:

upload emits [], uploading: true
upload n1 starts
server receives n1 and holds [n0, n1]
upload emits [n1], uploading: false
list reads [n0, n1]
refresh emits [n0, n1]

Помнить здесь про трансформеры и порядок обработки событий не нужно: трансформер некуда передать, планирование не выбирают. Корневые Job контроллера выполняются по одной, и по-другому не выполняются. Готовые Job стартуют в порядке добавления. Этот порядок меняют две вещи, и каждая случается только там, где её просит код. enqueueJob с first: true ставит Job впереди ожидающих, а Job, ждущую окна накопителя, может обойти готовая, стоящая за ней. Обе описаны на странице о задачах и очереди, в разделе «Работа с очередью напрямую». Добавление ещё одного метода через очередь сохраняет этот порядок: очередь одна на контроллер, а не на команду, и расположить методы так, чтобы у двух из них оказалось по своей очереди, нельзя. Гарантия относится к корневым Job; дочерние Job могут работать внутри родителя, в том числе параллельно, а независимые внешние изменения имеют отдельный путь, описанный в разделе «Реакция на независимое внешнее изменение состояния».

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

Работа, запущенная без ожидания, может пережить Job; нельзя считать, что она закончится до close(). Так устроен Future в Dart: он ничем не связан с кодом, который его запустил, и ни Job, ни контроллер о нём не знают. Если такая работа должна закончиться до конца Job, её запускают дочерней Job через ctx.run. Родитель ждёт своих детей, даже если тело их не ждёт, очередь до тех пор не запускает следующую Job, и close() тоже ждёт детей. Они разобраны на странице «Дочерние задачи и цепочки».

Что даёт solo: после отправки и обновления состояние равно NotesState([n0, n1], uploading: false), потому что корневые Job контроллера выполняются по одной, а трансформер некуда передать.

Ошибка наблюдателя при обновлении состояния

Диктофон запускает нативную запись, публикует Recording и включает индикатор уровня. Наблюдатель телеметрии бросает ошибку при обработке обновления. Операция диктофона должна завершиться независимо от отправки телеметрии.

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

Наблюдатель, который сообщает о каждом изменении, и контроллер, который ведёт локальный журнал изменений. Оба написаны так, как сами хуки и предлагают, с вызовом super первой строкой:

// Первая попытка: исключение наблюдателя срывает emit.
class RecorderBloc extends Bloc<StartRecording, RecorderState> {
  final Recorder _recorder;
  final Journal _journal;

  RecorderBloc(this._recorder, this._journal) : super(const Idle()) {
    on<StartRecording>((e, emit) async {
      await _recorder.start();
      emit(const Recording());
      _recorder.armMeter();
    });
  }

  @override
  void onChange(Change<RecorderState> change) {
    super.onChange(change);
    _journal.note('${change.nextState}');
  }
}

class TelemetryObserver extends BlocObserver {
  final Telemetry _telemetry;

  TelemetryObserver(this._telemetry);

  @override
  void onChange(BlocBase<dynamic> bloc, Change<dynamic> change) {
    super.onChange(bloc, change);
    _telemetry.send(change.nextState);
  }
}

Телеметрия недоступна, и каждый send бросает. Устройство начинает запись, но состояние остаётся Idle. Индикатор уровня не включается. Локальный журнал тоже пуст, и дело не в неудачном порядке строк: документация самого onChange требует вызывать super.onChange первым, а это ставит глобального наблюдателя впереди собственной записи контроллера. Перенос записи до super спасает журнал, в нём остаётся [Recording], и больше ничего: состояние всё так же Idle, индикатор всё так же не включён. Обработчик кончается ошибкой telemetry unavailable, и она приходит в зону необработанной, а та же ошибка наблюдателя в методе Cubit передаётся вызывающему этот метод коду.

Bloc и Cubit публикуют состояние через BlocBase.emit. Он вызывает onChange до изменения состояния, а ошибку этого вызова передаёт в onError и бросает дальше, в тот самый обработчик, который публиковал. Отправка отчёта, ради которой наблюдатель и заведён, оказывается на одном пути с операцией, о которой она отчитывается.

Bloc и Cubit

Перехватите ошибки телеметрии внутри наблюдателя, чтобы они не прерывали обновление:

class GuardedTelemetryObserver extends BlocObserver {
  final Telemetry _telemetry;
  final Log _fallback;

  GuardedTelemetryObserver(this._telemetry, this._fallback);

  @override
  void onChange(BlocBase<dynamic> bloc, Change<dynamic> change) {
    super.onChange(bloc, change);
    try {
      _telemetry.send(change.nextState);
    } on Object catch (error, stackTrace) {
      _fallback.write(error, stackTrace);
    }
  }
}

С защитой состояние доходит до Recording, устройство получает оба вызова, [native start, arm meter], а в журнале стоит [Recording]. Резервный логгер получает ошибку телеметрии. Подход работает, если все соответствующие колбэки наблюдения обрабатывают свои ошибки, а резервный логгер не бросает исключение.

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

Solo

Наблюдатель, поставленный в Solo.observer, и собственные хуки контроллера вызываются независимо. Исключение каждого хука передаётся в текущую зону Dart:

final class RecorderController extends Solo<RecorderState> {
  final Recorder _recorder;
  final Journal _journal;

  RecorderController(this._recorder, this._journal) : super(const Idle());

  Job<void> start() => enqueue<RecorderState, void>(key: 'start', (ctx) async {
    await ctx.join(_recorder.start);
    ctx.emit(const Recording());
    _recorder.armMeter();
  });

  @override
  void onChange(SoloTransition<RecorderState> transition) =>
      _journal.note('${transition.current}');
}

final class TelemetryObserver extends SoloObserver {
  final Telemetry _telemetry;

  TelemetryObserver(this._telemetry);

  @override
  void onChange(Solo<Object> solo, SoloTransition<Object> transition) =>
      _telemetry.send(transition.current);
}

SoloObserver получает Solo<Object>, поскольку наблюдает контроллеры с разными типами состояний. Собственный хук контроллера использует его конкретный тип состояния.

С бросающим наблюдателем состояние всё равно доходит до Recording, индикатор включается, в журнале стоит [Recording], а Job заканчивается с Done(null): упавший хук не меняет ни исход Job, ни очередь. Ошибка телеметрии уходит в Zone.current.handleUncaughtError, туда же, где приложение уже разбирается с необработанными асинхронными ошибками, и больше никуда. Оставленная там без обработки, она всё ещё может завершить приложение: изоляция хуков сохраняет ход операции, но не берёт на себя отчётность. Тем же путём идёт и собственный хук контроллера: если бы бросил _journal.note, его ошибка тоже ушла бы в зону. Наблюдатель бы её не услышал, потому что его onError получает то, что бросила Job, а не хук.

Что даёт solo: бросающий наблюдатель не меняет ни исход Job, ни очередь: состояние доходит до Recording, индикатор включается, а ошибка наблюдателя уходит в зону и больше никуда.

Закрытие и отмена начатой работы

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

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

Экрана нет, контроллер закрыт, и обработчик написан так, будто закрытие на этом всё и заканчивает:

// Первая попытка: закрытие не останавливает обработчик.
class UnguardedChatBloc extends Bloc<ChatEvent, ChatState> {
  final Api _api;

  UnguardedChatBloc(this._api) : super(const ChatState()) {
    on<SendMessage>((e, emit) async {
      final reply = await _api.send(e.text);
      emit(state.withReply(reply));
      add(const MarkReplyRead());
    }, transformer: sequential());
    on<MarkReplyRead>((e, emit) => _api.markRead(), transformer: sequential());
  }
}

С sequential() закрытие ждёт вызов API, который ещё выполняется, а состояние после него равно ChatState(reply to hi): ответ опубликован в контроллер, который уже закрывался. Последующий add затем бросает Bad state: Cannot add new events after calling close, и эта ошибка приходит в зону необработанной, пока close() ещё ожидают. Её не получает ни тот код, что отправил сообщение, ни тот, что закрывает Bloc. Отметка о прочтении к серверу не уходит, но лишь потому, что бросил тот самый add, который её бы и отправил. Одна половина требования держится случайностью, сломавшей другую.

В версиях bloc, названных в начале страницы, поведение закрытия зависит от трансформера. С sequential() вызов close() проходит очередь до конца: работающий обработчик и каждое событие, ждущее за ним, обрабатываются по очереди, каждый из них ещё может публиковать состояние, и close() возвращается после последнего. С concurrent, который регистрация получает по умолчанию, или с restartable проходить нечего: каждое событие, добавленное до close(), сразу попадает в свой обработчик, потому что его ничто не держит. С droppable события, пришедшие, пока обработчик был занят, отброшены ещё при поступлении. В этих трёх случаях close() возвращается до конца тел, а отменённый emitter каждого из них игнорирует последующие записи. Ни один вариант не прерывает вызов API или оставшийся код тела обработчика, и ни один не выбрасывает событие, которое трансформер уже пропустил.

Bloc

Этот последовательный обработчик проверяет закрытие после await:

class ChatBloc extends Bloc<ChatEvent, ChatState> {
  final Api _api;

  ChatBloc(this._api) : super(const ChatState()) {
    on<SendMessage>((e, emit) async {
      final reply = await _api.send(e.text);
      if (isClosed) return;
      emit(state.withReply(reply));
      add(const MarkReplyRead());
    }, transformer: sequential());
    on<MarkReplyRead>((e, emit) => _api.markRead(), transformer: sequential());
  }
}

Проверка предотвращает и обновление ответа, и событие MarkReplyRead. Закрытие всё ещё ждёт возврата вызова API и обработчика, а с ними и каждого сообщения, ждущего за ними. Каждое всё равно уходит в API, пока идёт close(), а проверка отклоняет только его ответ. Отдельная регистрация MarkReplyRead безопасна в этом примере, поскольку не меняет общее состояние; иначе возникло бы ограничение порядка из раздела «Порядок обновлений общего состояния».

add после закрытия бросает StateError, поэтому вызывающий код, способный поздно отправить событие, должен сам обработать этот случай. Метод Cubit тоже продолжает выполнение после закрытия, но его прямой emit бросает исключение, а не использует отменённый emitter обработчика.

Другой случай: неожидаемая future вызывает emit после обычного завершения обработчика, пока Bloc ещё открыт. Это вызывает assert в отладочной сборке; с отключёнными assert поздняя запись меняет состояние. Это ограничение времени жизни разобрано в issue о завершённом обработчике. Поддержка отмены также обсуждается в предложении отменяемых операций.

Solo

close() прекращает приём Job, отменяет очередь, запрашивает отмену работающей Job и ждёт её завершения, включая освобождение ресурсов:

final class ChatController extends Solo<ChatState> {
  final Api _api;

  ChatController(this._api) : super(const ChatState());

  Job<void> send(String text) =>
      enqueue<ChatState, void>(key: 'send', (ctx) async {
        final reply = await ctx.abandonable(() => _api.send(text));
        ctx.emit(ctx.state.withReply(reply));
        markReplyRead();
      });

  Job<void> markReplyRead() =>
      enqueue<ChatState, void>((ctx) => ctx.abandonable(_api.markRead));
}

Future<void> onScreenClosed(ChatController chat) async {
  await chat.close();
  chat.send('bye'); // не выполнится: Job сразу вернётся с Cancelled(closed)
}

В этом теле ctx.abandonable бросает Cancelled, когда закрытие отменяет Job. Обновление ответа и вызов markReplyRead() не выполняются. Этот метод поставил бы отдельную корневую Job через enqueue контроллера; ребёнком send она не становится.

С обычным await закрытие ждало бы ответа API. ctx.emit после него всё равно бросил бы Cancelled, и ответ опубликован бы не был. Вызовы после закрытия возвращают Job, уже завершённые с Cancelled(closed), поэтому проверка isClosed в месте вызова не нужна.

Сообщение, ждущее за работающим, до API не доходит вовсе: оно кончается Cancelled(closed), не стартовав, а sequential() его бы отправил. Когда ждущая работа должна уйти, close(mode: SoloCloseMode.drain) сначала проходит очередь, как sequential(); его разбирает раздел «Отмена работы и закрытие контроллера» страницы об отмене. markReplyRead(), который зовёт каждый ответ, возвращается там всё равно с Cancelled(closed), потому что закрывающийся контроллер не берёт новых корневых Job.

ctx действителен, пока выполняется его Job, и у неожидаемой future из раздела про bloc выше есть здесь пара. Тело, которое начало future и не стало её ждать, возвращается, а future потом зовёт ctx.emit у законченной Job. Этот вызов бросает Bad state: Job(send) has already finished, cannot emit и в отладочной сборке, и в релизной, а состояние остаётся прежним. Там, где у bloc assert, исчезающий в релизе, здесь обычная ошибка, которая не исчезает.

Что даёт solo: close() отменяет работающую Job на ctx.abandonable, до обновления ответа и markReplyRead(), а ждущее сообщение или вызов после close() кончаются Cancelled(closed) без проверки isClosed.

Выход из состояния загрузки при отмене работы

Отмена может происходить и при открытом контроллере. У индикатора обновления два состояния, Initial и Loading. StartRefresh начинает работу; CancelRefresh отменяет обработчик через ту же регистрацию с restartable(). При отмене обновления индикатор должен вернуться в Initial. Пример отслеживает только активность обновления; API не возвращает данных для отображения.

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

Сброс принадлежит тому месту, где работа кончается, чем бы она ни кончилась, а это finally:

// Первая попытка: finally пишет в отменённый emitter.
class RefreshBloc extends Bloc<RefreshEvent, RefreshState> {
  final RefreshApi _api;

  RefreshBloc(this._api) : super(const Initial()) {
    on<RefreshEvent>((event, emit) async {
      if (event is CancelRefresh) return;
      emit(const Loading());
      try {
        await _api.refresh();
      } finally {
        emit(const Initial());
      }
    }, transformer: restartable());
  }
}

После StartRefresh состояние становится Loading. CancelRefresh заменяет обработчик, но не публикует новое состояние, а future, которую вернул вызов API, продолжает выполняться. Когда она завершается, finally вызывает старый, отменённый emitter, который игнорирует Initial. Поэтому состояние остаётся Loading даже после завершения вызова API. Сама отмена не бросала исключение в старое тело.

Bloc

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

if (event is CancelRefresh) {
  emit(const Initial());
  return;
}

Теперь состояние становится Initial при обработке события отмены. Поздний emit старого обработчика по-прежнему игнорируется и не может перезаписать более новое обновление. С restartable любой тип события этой регистрации может заменить текущий обработчик; отдельная регистрация on<CancelRefresh> сама по себе не отменит StartRefresh.

Solo

Сброс привязывается к итоговому исходу операции:

final class RefreshController extends Solo<RefreshState> {
  final RefreshApi _api;

  RefreshController(this._api) : super(const Initial());

  Job<void> refresh() => enqueue<RefreshState, void>(
    key: 'refresh',
    policy: Policy.restart,
    ifFailed: (state, error, stackTrace) => const Initial(),
    ifCancelled: (state, cancelled) => const Initial(),
    (ctx) async {
      ctx.emit(const Loading());
      await ctx.abandonable(_api.refresh);
      ctx.emit(const Initial());
    },
  );

  Job<void>? cancelRefresh() {
    final refresh = lastJobWhere((job) => job.key == 'refresh');
    unawaited(refresh?.cancel());
    return refresh;
  }
}

cancelRefresh() находит работающее обновление по его ключу и отменяет, так что для отмены экрану нужен только контроллер, как на стороне bloc для отправки CancelRefresh нужен только сам Bloc. Отдаёт он ту самую Job, которую отменяет, а не future, по той же причине, по которой её отдаёт refresh(): вызов controller.cancelRefresh(); составляет законченное предложение, и проекту с включённым discarded_futures придраться не к чему. Кому нужен исход, тот дожидается controller.cancelRefresh()?.done, а null там означает, что отменять было нечего.

lastJobWhere просматривает очередь с конца и только потом берёт работающую Job, то есть отвечает ожидающим обновлением, если оно есть. Этому методу нужен именно такой порядок: обновление оказывается в очереди только потому, что refresh() вызвали второй раз, а Policy.restart в тот же момент отменила работающее. ctx.abandonable прекращает ожидание API, а ifCancelled возвращает Initial до продвижения очереди. Успешное обновление публикует Initial из тела; ifFailed сбрасывает индикатор при ошибке, при этом Job по-прежнему сообщает Failed.

Момент обновления различается: обработчик события отмены в bloc меняет состояние при своём выполнении, а обработчик состояния solo вызывается после уборки отменённой Job. Ни один пример не останавливает саму операцию API. Если независимое внешнее состояние делает Job solo недопустимой, её итоговые обработчики состояния пропускаются. Раздел «Реакция на независимое внешнее изменение состояния» объясняет внешнее состояние, а правила этих обработчиков даёт раздел «Состояние после ошибки или отмены» страницы о состоянии.

Что даёт solo: ifCancelled возвращает индикатор в Initial после уборки отменённой Job, ifFailed сбрасывает его при ошибке, а cancelRefresh() находит обновление для отмены по его ключу.

Перезапуск одной операции в общей очереди

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

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

По регистрации на команду, а на той, которую нужно уметь заменять, стоит трансформер, названный по требованию:

// Первая попытка: три регистрации, и выполняется каждая перемотка.
class SplitPlayerBloc extends Bloc<PlayerCommand, PlayerState> {
  final Player _player;

  SplitPlayerBloc(this._player) : super(const PlayerState()) {
    on<Play>((e, emit) => _player.play(), transformer: sequential());
    on<Pause>((e, emit) => _player.pause(), transformer: sequential());
    on<Seek>((e, emit) async {
      await _player.seek(e.position);
      emit(PlayerState(position: e.position));
    }, transformer: restartable());
  }
}

Пользователь запускает воспроизведение, протаскивает ползунок через три позиции и жмёт паузу: Play, Seek(1ms), Seek(2ms), Seek(3ms), Pause подряд, ничего между ними не дожидаясь.

Трасса устройства: [play start, seek 1 start, seek 2 start, pause start, seek 3 start, play end, pause end, seek 1 end, seek 2 end, seek 3 end], а итоговое состояние PlayerState(3ms), то есть ровно та позиция, которую просил пользователь. Что устройство сделало из пяти команд, идущих одновременно, по состоянию не сказать.

Не сработало сразу двое. Трансформер упорядочивает события одной регистрации и ничего сверх того, а регистраций здесь три: pause ждёт другую pause и никогда не ждёт play. А restartable() отменяет emitter заменённого обработчика, но не останавливает нативный вызов, который тот ждёт: на устройство ушли все три перемотки, причём третья уже после паузы. До состояния дошёл только последний emit, поэтому по состоянию ничего этого не видно.

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

Одна регистрация на все три команды, чтобы очередь была одна:

// Вторая попытка: одна очередь, и все устаревшие перемотки в ней.
class SerialPlayerBloc extends Bloc<PlayerCommand, PlayerState> {
  final Player _player;

  SerialPlayerBloc(this._player) : super(const PlayerState()) {
    on<PlayerCommand>((command, emit) async {
      switch (command) {
        case Play():
          await _player.play();
        case Pause():
          await _player.pause();
        case Seek(:final position):
          await _player.seek(position);
          emit(PlayerState(position: position));
      }
    }, transformer: sequential());
  }
}

Теперь команды доходят до устройства в том порядке, в каком их нажимали. Состояние снова заканчивается на PlayerState(3ms), и в этот раз с ним согласно устройство: оно действительно стоит на паузе на третьей позиции. Трасса: [play start, play end, seek 1 start, seek 1 end, seek 2 start, seek 2 end, seek 3 start, seek 3 end, pause start, pause end].

Очередь не умеет выбрасывать то, что перетаскивание уже обессмыслило. Позиции 1 и 2 устарели, не успев начаться, и устройство перематывает на каждую по очереди; pause ждёт за всеми тремя. Порядок и замена составляют два разных требования, и sequential() отвечает только на первое.

Bloc

Одна очередь остаётся. Эта версия добавляет к ней способ понять, какая перемотка ещё нужна, и остановить ту, что уже выполняется:

class PlayerBloc extends Bloc<PlayerCommand, PlayerState> {
  final Player _player;
  Duration? _newestSeek;
  CancelToken? _seeking;

  PlayerBloc(this._player) : super(const PlayerState()) {
    on<PlayerCommand>((command, emit) async {
      switch (command) {
        case Play():
          await _player.play();
        case Pause():
          await _player.pause();
        case Seek(:final position):
          if (position != _newestSeek) return;
          final token = CancelToken();
          _seeking = token;
          await _player.seek(position, cancelToken: token);
          _seeking = null;
          if (token.cancelled) return;
          emit(PlayerState(position: position));
      }
    }, transformer: sequential());
  }

  @override
  void onEvent(PlayerCommand event) {
    super.onEvent(event);
    if (event is Seek) {
      _newestSeek = event.position;
      _seeking?.cancel();
    }
  }
}

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

Для позиций 1, 2, 3, добавленных вместе, устройство получает [play, seek 3, pause]. Если seek 1 уже выполняется при появлении seek 3, трасса выглядит так: [seek 1 start, seek 1 stopped, seek 3 start, seek 3 end].

Сравнения значений позиции недостаточно, когда перетаскивание повторяет значение: ввод 1, 2, 1 даёт [play, seek 1, seek 1, pause]. Чтобы такие события различались, каждое нужно помечать при поступлении, а не сравнивать по тому, что оно несёт. Эту схему целиком показывает раздел «Удаление выбранной ожидающей работы», вместе с условием, которое к ней прилагается: события должны быть разными объектами.

Три вещи держит на себе приложение: какая позиция последняя (_newestSeek), какой токен принадлежит выполняющейся перемотке (_seeking) и проверка после await, которая не даёт отменённой перемотке опубликовать состояние. Команда с другим правилом замены приносит в тот же обработчик свою ветку и свою бухгалтерию. А add ничего не возвращает, поэтому у пропущенной или прерванной перемотки нет исхода, который мог бы прочитать вызывающий код.

Solo

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

enum PlayerKey { play, pause, seek }

final class PlayerController extends Solo<PlayerState> with SoloStream {
  final Player _player;

  PlayerController(this._player) : super(Ready());

  Job<void> play() => enqueue<Ready, void>(key: PlayerKey.play, (ctx) async {
    await ctx.join(_player.play);
    ctx.emit(ctx.state.copyWith(playing: true));
  });

  Job<void> pause() => enqueue<Ready, void>(key: PlayerKey.pause, (ctx) async {
    await ctx.join(_player.pause);
    ctx.emit(ctx.state.copyWith(playing: false));
  });

  Job<void> seek(Duration position) => enqueue<Ready, void>(
    key: PlayerKey.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));
    },
  );
}

Policy.restart удаляет отменяемые ожидающие перемотки с тем же ключом и запрашивает отмену активной. ctx.onCancel сразу передаёт запрос плееру. ctx.join ждёт возврата операции, прежде чем текущая Job сможет закончиться, а замена начаться. Если операция, останавливаясь, бросает ошибку, тело получает эту ошибку, и исход Job остаётся отменённым. Затем ошибка уходит в хук onError контроллера и дальше в onUnanswered, как любой провал тела отменённой задачи, если только приложение не назвало её один раз в Job.isCancellationError.

with SoloStream добавляет stream состояний, из которого читаются эти трассы; политике он не нужен. Трассы те же, что у ответа на bloc выше: [play, seek 3, pause] для перетаскивания и [seek 1 start, seek 1 stopped, seek 3 start, seek 3 end], когда seek 1 уже выполняется. pause и play остаются в той же очереди с последовательной политикой по умолчанию. Токен лежит в локальной переменной тела seek, а вызывающий код может проверить исход каждой перемотки, включая Cancelled(replaced) для заменённого запроса.

Что даёт solo: с Policy.restart только на seek устройство получает [play, seek 3, pause], начатая перемотка останавливается до старта замены, а исход каждой перемотки можно проверить.

Типизированные методы с выполнением через очередь

Карта предоставляет moveTo и setZoom. Снаружи их должны вызывать как методы самого контроллера, вроде controller.moveTo(point), со своими аргументами и без отдельного класса на каждую команду. Частые обновления при перетаскивании не должны оставлять карту на устаревшей позиции. В сценарии ниже перетаскивание отправляет позицию 1, затем, пока карта ещё движется к ней, позиции 2 и 3, а пользователь меняет масштаб на 4.

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

Cubit напрямую поддерживает методы, возвращающие future, поэтому типизированный интерфейс вызова достаётся даром:

// Первая попытка: вызовы идут разом, и победить может любой.
class MapCubit extends Cubit<MapState> {
  final MapApi _map;

  MapCubit(this._map) : super(const MapState());

  Future<void> moveTo(Point<double> point) async {
    await _map.moveTo(point);
    emit(state.copyWith(center: point));
  }

  Future<void> setZoom(double value) async {
    await _map.setZoom(value);
    emit(state.copyWith(zoom: value));
  }
}

Эти методы не упорядочивают вызовы. При разной длительности нативных вызовов трасса может выглядеть так: [moveTo 1 start, moveTo 2 start, moveTo 3 start, moveTo 3 end, moveTo 2 end, moveTo 1 end]. Состояние может закончиться на любой из трёх позиций: здесь оно заканчивается на MapState(1, z4), самом старом запросе, потому что именно он вернулся последним. В коде нет ничего, что решало бы, какой победит.

Цепочка future может упорядочить вызовы; для отбрасывания устаревших запросов нужен дополнительный учёт. Приложение также должно решить, как закрытие дожидается этой цепочки или делает её недействительной; Cubit не управляет этим автоматически.

Bloc с функциями в качестве событий

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

typedef MapCommand = Future<void> Function(Emitter<MapState>);

class CommandMapBloc extends Bloc<MapCommand, MapState> {
  final MapApi _map;

  CommandMapBloc(this._map) : super(const MapState()) {
    on<MapCommand>((command, emit) => command(emit), transformer: sequential());
  }

  void moveTo(Point<double> point) => add((emit) async {
    await _map.moveTo(point);
    emit(state.copyWith(center: point));
  });

  void setZoom(double value) => add((emit) async {
    await _map.setZoom(value);
    emit(state.copyWith(zoom: value));
  });
}

Теперь вызовы идут по порядку и заканчиваются с MapState(3, z4). Обычный класс события типизирован ничуть не хуже, ведь add(MoveTo(point)) проверяет аргумент, как любой другой конструктор. Но тогда словарь команд живёт рядом с контроллером, а не на нём, и каждой команде нужен свой класс. Замыкание убирает и то и другое, а платой становится опознаваемость события: наблюдатель этого Bloc получает на каждую из четырёх команд (Emitter<MapState>) => Future<void>, без имени и без аргументов, которые можно было бы записать. Все команды по-прежнему используют один трансформер, а показанные методы возвращают void. Для возврата результата нужен дополнительный механизм, как в разделе «Ожидание конкретного запроса».

Solo

Каждый метод возвращает Job и задаёт политику замены:

enum MapKey { moveTo, setZoom }

final class MapController extends Solo<MapState> {
  final MapApi _map;

  MapController(this._map) : super(const MapState());

  Job<void> moveTo(Point<double> point) => enqueue<MapState, void>(
    key: MapKey.moveTo,
    policy: Policy.restart,
    (ctx) async {
      final token = CancelToken();
      ctx.onCancel(token.cancel);
      await ctx.join(() => _map.moveTo(point, cancelToken: token));
      ctx.emit(ctx.state.copyWith(center: point));
    },
  );

  Job<void> setZoom(double value) => enqueue<MapState, void>(
    key: MapKey.setZoom,
    policy: Policy.restart,
    (ctx) async {
      final token = CancelToken();
      ctx.onCancel(token.cancel);
      await ctx.join(() => _map.setZoom(value, cancelToken: token));
      ctx.emit(ctx.state.copyWith(zoom: value));
    },
  );
}

void onMapDrag(MapController map, Point<double> point) => map.moveTo(point);

Трасса: [moveTo 1 start, moveTo 1 stopped, moveTo 3 start, moveTo 3 end]. Первый запрос уже шёл и останавливается до начала последнего, а средний не начинается. И карта, и MapState(3, z4) отражают последний запрос. Вызывающий код может дождаться операции через job.value или job.done либо обойтись без ожидания. Сама возвращённая Job не Future; что происходит с ошибкой Job, которой никто не ждёт, говорит раздел «Обработанные и необработанные ошибки» страницы об ошибках.

Что даёт solo: типизированные методы возвращают Job, а с Policy.restart первое перемещение останавливается до начала последнего, среднее не начинается, и карта заканчивается на MapState(3, z4).

Удаление выбранной ожидающей работы

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

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

Одна регистрация с sequential() упорядочивает все команды устройства. Через API Bloc нельзя перечислить или удалить её ожидающие события, поэтому обработчику приходится проверять, актуально ли ещё чтение. Самым очевидным предметом проверки выглядит флаг ухода: экран ставит его, когда уходит, и снимает, когда возвращается.

// Первая попытка: флаг, который сбрасывает вернувшийся экран.
class FlagDeviceBloc extends Bloc<DeviceEvent, DeviceState> {
  final Ble _ble;
  var _leaving = false;

  FlagDeviceBloc(this._ble) : super(const DeviceState()) {
    on<DeviceEvent>((e, emit) async {
      switch (e) {
        case Connect():
          await _ble.connect();
          emit(state.copyWith(online: true));
        case ReadBattery():
          if (_leaving) return;
          emit(state.copyWith(battery: await _ble.battery()));
        case Rename(:final name):
          await _ble.rename(name);
        case Disconnect():
          await _ble.disconnect();
          emit(const DeviceState());
      }
    }, transformer: sequential());
  }

  @override
  void onEvent(DeviceEvent event) {
    super.onEvent(event);
    if (event is Disconnect) _leaving = true;
    if (event is Connect) _leaving = false;
  }
}

На уходе флаг работает: [_leaving = true, connect, rename kitchen, disconnect]. Трасса несёт записи в _leaving вместе с вызовами устройства, и запись стоит раньше их всех. onEvent выполняется внутри add, а очередь разбирается, когда все четыре команды уже в ней. К тому времени, как обработчик доходит до первой, флаг выставлен, и чтение пропускается.

Этим не выразить экрана, который вернулся раньше, чем разошлись старые события. Те же четыре команды, а за ними Connect открывшегося экрана: [_leaving = true, _leaving = false, connect, battery, rename kitchen, disconnect, connect]. Обе записи ложатся раньше первого вызова устройства, вторая снимает флаг, и чтение, поставленное ушедшим экраном, снова разрешено. Это чтение сделано для экрана, которого уже нет.

Bloc

Флаг говорит, где экран сейчас; очереди же нужно знать, какому экрану принадлежит каждое событие. Эта версия записывает поколение экрана на каждом событии:

class DeviceBloc extends Bloc<DeviceEvent, DeviceState> {
  final Ble _ble;
  final _stampOf = Expando<int>();
  int _screen = 0;

  DeviceBloc(this._ble) : super(const DeviceState()) {
    on<DeviceEvent>((e, emit) async {
      switch (e) {
        case Connect():
          await _ble.connect();
          emit(state.copyWith(online: true));
        case ReadBattery():
          if (_stampOf[e] != _screen) return;
          emit(state.copyWith(battery: await _ble.battery()));
        case Rename(:final name):
          await _ble.rename(name);
        case Disconnect():
          await _ble.disconnect();
          emit(const DeviceState());
      }
    }, transformer: sequential());
  }

  @override
  void onEvent(DeviceEvent event) {
    super.onEvent(event);
    if (event is Disconnect) _screen++;
    _stampOf[event] = _screen;
  }
}

Поколение увеличивается там же, где выставлялся флаг, и так же рано: [_screen = 1, connect, rename kitchen, disconnect]. Разница в том, что стоящие в очереди события сохраняют доставшуюся им метку, поэтому Connect после них ничего не меняет для чтения: [_screen = 1, connect, rename kitchen, disconnect, connect].

Записи поколений остаются состоянием приложения, связанным с очередью. События всё равно доходят до обработчика, который должен проверять каждую отбрасываемую команду. Схема с Expando также требует разных объектов событий. Если открывшийся экран просит батарею тем же каноническим const ReadBattery(), этот add перезаписывает метку стоящего в очереди чтения, и устаревшее чтение всё-таки выполняется: [_screen = 1, connect, battery, rename kitchen, disconnect, connect, battery]. add не возвращает результат, указывающий на пропуск чтения.

Solo

Контроллер может удалять подходящие Job прямо из своей очереди:

enum DeviceKey { connect, readBattery, rename, disconnect }

final class DeviceController extends Solo<DeviceState> {
  final Ble _ble;

  DeviceController(this._ble) : super(const Offline());

  Job<void> connect() => enqueue<Offline, void>(
    key: DeviceKey.connect,
    cancellable: false,
    (ctx) async {
      await ctx.join(_ble.connect);
      ctx.emit(const Connected());
    },
  );

  Job<void> readBattery() =>
      enqueue<Connected, void>(key: DeviceKey.readBattery, (ctx) async {
        final battery = await ctx.join(_ble.battery);
        ctx.emit(ctx.state.copyWith(battery: battery));
      });

  Job<void> rename(String name) => enqueue<Connected, void>(
    key: DeviceKey.rename,
    cancellable: false,
    (ctx) => ctx.join(() => _ble.rename(name)),
  );

  Job<void> disconnect() {
    queue.removeWhere((job) => job.key == DeviceKey.readBattery);
    return enqueue<Connected, void>(
      key: DeviceKey.disconnect,
      cancellable: false,
      (ctx) async {
        await ctx.join(_ble.disconnect);
        ctx.emit(const Offline());
      },
    );
  }
}

Вызовы устройства те же, но удаление происходит при вызове disconnect. Удалённая Job чтения завершается с Cancelled(manual); вызывающий код может наблюдать этот результат. Переименование остаётся в очереди, а отключение выполняется после него. removeWhere работает только с очередью: если чтение уже началось, его она не удаляет, и устройство получает [connect, battery, rename kitchen, disconnect].

У переименования стоит cancellable: false, и это вторая половина требования. Удаление в disconnect называет чтение, поэтому переименование переживает его просто потому, что не названо, а про код, который позже сметёт всю очередь через cancelAll() или queue.clear(), та строка не говорит ничего. Переименование, о котором попросил пользователь, обязано дойти до устройства и тогда, поэтому гарантия принадлежит задаче, а не строке, удаляющей кого-то другого. Такую задачу общая чистка пропускает, а не удаляет, и уносит её только force: true.

cancellable: false стоит и у connect с disconnect, ради того, что бывает после их старта. ctx.join даёт вызову устройства закончиться и после отмены, а тело, отменённое за это время, пропускает следующий за ним emit. Чистка во время disconnect оставила бы устройство отключённым, а состояние Connected, а чистка во время connect наоборот. Работающей Job с cancellable: false чистка отказывает, поэтому состояние вслед за устройством доходит до Offline и Connected. Отменяемой остаётся одна команда, чтение: на устройстве оно ничего не меняет, и выбросить его стоит только показаний батареи.

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

Они же несут состояние, которое нужно каждой команде. Экран закрывается, Disconnect уже в очереди, и в этот момент таймер обновления добавляет ReadBattery. В версии с метками выше новое событие получает метку текущего поколения, того самого, которое Disconnect только что увеличил, и проверка метки его пропускает. Очередь доходит до него уже после отключения, и обработчик зовёт _ble.battery у устройства, которое больше не подключено. Устройство отказывает; bloc сообщает об ошибке в onError и пробрасывает её дальше, а так как обработчика никто не ждёт, она приходит в зону неперехваченной: Bad state: battery: not connected.

Здесь enqueue<Connected, void> проверяется на старте Job, поэтому то же чтение заканчивается на Cancelled(rules: is not Connected), и вызова не происходит. В bloc тот же запрет пишется руками: ещё один if в switch, на каждую команду, которой он нужен.

Что даёт solo: queue.removeWhere убирает ждущее чтение в момент вызова disconnect, cancellable: false заставляет чистку пропустить переименование, а чтение, вставшее в очередь за отключением, кончается Cancelled(rules: is not Connected) и до устройства не доходит.

Ожидание конкретного запроса

Оплату могут вызвать и кнопка Pay, и запрос платформы. Обработчик платформы должен вернуть результат своего заказа. Одновременные запросы одного заказа должны использовать одну оплату; разные заказы должны выполняться последовательно.

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

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

Bloc.add возвращает void, и этот случай затронут в обсуждении ожидания событий: результату придётся ехать на самом событии, в completer’е, которого ждёт вызывающий код. droppable() читается как политика объединения: пока списание выполняется, второй запрос на него не должен начинать ещё одно.

// Первая попытка: отброшенные события оставляют вызывающих ждать.
class Pay extends CheckoutEvent {
  final Order order;
  final Completer<Receipt> result;

  Pay(this.order) : result = Completer<Receipt>();
}

class DroppableCheckoutBloc extends Bloc<CheckoutEvent, CheckoutState> {
  final Api _api;

  DroppableCheckoutBloc(this._api) : super(Idle()) {
    on<Pay>((e, emit) async {
      emit(Paying(e.order.id));
      final receipt = await _api.pay(e.order);
      e.result.complete(receipt);
      emit(Paid(receipt));
    }, transformer: droppable());
  }

  Future<Receipt> pay(Order order) {
    final event = Pay(order);
    add(event);
    return event.result.future;
  }
}

Три запроса, дважды заказ A и затем заказ B, дают одно обращение к API, и ответ получает один вызывающий из трёх: [Receipt(for A), never answered, never answered]. Двое других продолжают ждать, когда всё остальное уже закончилось, и ждали бы столько, сколько живёт процесс: droppable() отбросил их события, а completer отброшенного события не завершает никто.

Второй из этих двоих показывает, что политика делает на самом деле: заказ B оплачивается отдельно, и его тоже отбросили. droppable() отбрасывает то, что пришло, пока работает обработчик, а не то, что дублирует его.

Bloc

Completer едет на событии, как и выше. Объединению не хватает карты выполняющихся оплат. Через неё тот, кто вызывает оплату заказа, который уже выполняется, получает completer той самой оплаты, а не отброшенное событие со своим:

class CheckoutBloc extends Bloc<CheckoutEvent, CheckoutState> {
  final Api _api;
  final _inFlight = <String, Completer<Receipt>>{};

  CheckoutBloc(this._api) : super(Idle()) {
    on<Pay>((e, emit) async {
      try {
        emit(Paying(e.order.id));
        final receipt = await _api.pay(e.order);
        e.result.complete(receipt);
        emit(Paid(receipt));
      } on Object catch (error, stackTrace) {
        if (!e.result.isCompleted) {
          e.result.completeError(error, stackTrace);
        }
        emit(PaymentFailed(e.order.id, error));
      } finally {
        _inFlight.remove(e.order.id);
      }
    }, transformer: sequential());
  }

  Future<Receipt> pay(Order order) {
    final running = _inFlight[order.id];
    if (running != null) return running.future;
    final event = Pay(order);
    _inFlight[order.id] = event.result;
    try {
      add(event);
    } on Object {
      _inFlight.remove(order.id);
      rethrow;
    }
    return event.result.future;
  }
}

await bloc.pay(order) теперь возвращает квитанцию. Три вызова по двум заказам дают два обращения к API, а оба вызова для одного заказа получают его квитанцию. В контроллере остаётся Paid(Receipt(for B)), то есть оплата, завершившаяся последней, а не запись об обеих.

Completer должен завершаться на каждом пути. catch вокруг add удаляет запись из карты, когда закрытый Bloc отклоняет событие. isCompleted предотвращает повторное завершение, если emit бросает ошибку после передачи квитанции. Завершение результата до публикации Paid не даёт ошибке наблюдателя заменить уже полученную квитанцию, хотя состояние тогда показывает PaymentFailed для заказа, который списан. Трансформер остаётся sequential(): объединяет оплаты карта, а до очереди после неё доходят разные заказы, которым нужен порядок, а не взаимное отбрасывание.

Cubit

Метод Cubit может вернуть квитанцию напрямую, но одновременные вызовы автоматически не объединяются и не упорядочиваются. Эта версия добавляет карту выполняющихся оплат и цепочку future:

class CheckoutCubit extends Cubit<CheckoutState> {
  final Api _api;
  final _inFlight = <String, Future<Receipt>>{};
  Future<void> _tail = Future.value();

  CheckoutCubit(this._api) : super(Idle());

  Future<Receipt> pay(Order order) {
    final running = _inFlight[order.id];
    if (running != null) {
      return running;
    }
    final turn = Completer<void>();
    final before = _tail;
    _tail = turn.future;
    final result = before.then((_) => _pay(order)).whenComplete(turn.complete);
    _inFlight[order.id] = result;
    return result;
  }

  Future<Receipt> _pay(Order order) async {
    try {
      emit(Paying(order.id));
      final receipt = await _api.pay(order);
      emit(Paid(receipt));
      return receipt;
    } finally {
      _inFlight.remove(order.id);
    }
  }
}

Она тоже даёт два обращения к API для трёх запросов по двум заказам. Обработка закрытия остаётся неполной: Cubit не ждёт эту собственную цепочку. При закрытии во время оплаты списание проходит, но последующий emit(Paid(...)) бросает Bad state: Cannot emit new states after calling close, и вызывающий код получает эту ошибку вместо квитанции. Рабочая версия должна согласовать закрытие с цепочкой оплат и доставкой результата.

Solo

Используйте идентификатор заказа в ключе и верните Job напрямую:

final class CheckoutController extends Solo<CheckoutState> {
  final Api _api;

  CheckoutController(this._api) : super(const Idle());

  Job<Receipt> pay(Order order) => enqueue<CheckoutState, Receipt>(
    key: ('pay', order.id),
    policy: Policy.droppable,
    cancellable: false,
    (ctx) async {
      ctx.emit(Paying(order.id));
      final receipt = await _api.pay(order);
      ctx.emit(Paid(receipt));
      return receipt;
    },
  );

  bool withdraw(Order order) =>
      queue.removeWhere((job) => job.key == ('pay', order.id), force: true) > 0;
}

Обработчик платформы может разобрать исход этой Job:

Future<Map<String, Object?>> handlePayRequest(
  CheckoutController checkout,
  Order order,
) async {
  switch (await checkout.pay(order).done) {
    case Done(:final value):
      return {'paid': true, 'receipt': value.id};
    case Cancelled(:final reason):
      return {'paid': false, 'cancelled': reason.name};
    case Failed(:final error):
      return {'paid': false, 'error': '$error'};
  }
}

Ключ ('pay', order.id) у всех запросов одного заказа один и тот же, и Policy.droppable возвращает уже ожидающую или работающую Job с этим ключом, а не ставит новую. Оба вызова получают общую квитанцию, а другой заказ создаёт другую Job. Три запроса снова дают два обращения к API.

Списание ждёт обычный await. ctx.abandonable, ctx.join и ctx.uncancellable нужны там, где Job можно отменить. Различаются они тем, что каждый делает с отменой, пришедшей во время вызова: abandonable отпускает вызов, join остаётся с ним до ответа, uncancellable придерживает отмену до конца шага. Эту Job во время работы отменить нельзя. keepWhile у неё не задан, а принимает она базовый CheckoutState, поэтому отбить её не может ни одно правило состояния. Важна здесь именно эта ширина, потому что правило состояния отменяет даже неотменяемую Job: более узкий тип допустил бы отмену между отправкой списания и записью его результата. Всякая другая отмена отклоняема, и cancellable: false её отклоняет: Job.cancel с экрана, закрытие контроллера.

Тело публикует Paid до завершения с квитанцией, а close() ждёт этого завершения. Ошибки API по-прежнему дают Failed; состояние ошибки можно задать через enqueue(ifFailed: ...). Отозвать списание всё равно не смог бы ни один метод ожидания: это API не предоставляет механизма отмены.

Флаг также отклоняет ручную отмену, пока оплата стоит в очереди. Оплату, которую пользователь может отозвать до начала списания, убирает из очереди withdraw. force: true снимает ожидающую Job при любом флаге, а работающую Job очередь не трогает никогда, так что уже начатое списание остаётся, и withdraw возвращает false. close() и queue.clear(force: true) по-прежнему могут отбросить ожидающую оплату без списания, а вызов после закрытия тоже не запускается. Эти случаи, и withdraw среди них, объясняют ветку Cancelled обработчика платформы.

Если вместо флага обернуть всё тело в ctx.uncancellable, так не выйдет. Секция не отклоняет отмену, а придерживает. Отмена, пришедшая во время списания, срабатывает, когда секция закрывается, уже после публикации Paid, и Job кончается Cancelled. Обработчик платформы тогда отвечает paid: false на оплату, которая прошла.

Общая Job в памяти объединяет только одновременные вызовы этого контроллера. Повтор оплаты после перезапуска процесса или сетевого сбоя также требует идемпотентного API оплаты; ни один пример этого не предоставляет.

Что даёт solo: с ключом на заказ и Policy.droppable вызовы одного заказа делят его Job и квитанцию, а другой заказ получает свою Job. cancellable: false отклоняет отмену с экрана, а обработчик платформы читает исход своего запроса.

Реакция на независимое внешнее изменение состояния

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

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

Прямой emit у Bloc помечен @visibleForTesting и документирован для внутреннего использования, поэтому слушатель вместо него добавляет событие SessionRevoked. Одной регистрации на оба события требует раздел «Порядок обновлений общего состояния», и её же напишет читатель, только что усвоивший тот урок:

// Первая попытка: отзыв сессии ждёт за сборкой отчёта.
class FunnelReportBloc extends Bloc<ReportEvent, ReportState> {
  final Reports _reports;

  FunnelReportBloc(this._reports, Auth auth) : super(SignedIn()) {
    auth.onRevoked = (reason) => add(SessionRevoked(reason));
    on<ReportEvent>((e, emit) async {
      switch (e) {
        case SessionRevoked(:final reason):
          emit(SignedOut(reason));
        case BuildReport(:final range):
          if (state is! SignedIn) return;
          final report = await _reports.build(range);
          if (state is! SignedIn) return;
          emit(Ready(report));
      }
    }, transformer: sequential());
  }
}

Опубликованные состояния: [Ready(Report(30 days)), SignedOut(signed out elsewhere)]. Сессия была уже отозвана, когда публиковался отчёт: событие отзыва ждало своей очереди за сборкой, которую оно делает недопустимой. Смотреть здесь стоит на проверку после await. Сработать она тут не может вовсе: единственное событие, которое изменило бы состояние, стоит в очереди за этим же обработчиком. Поэтому от её удаления в поведении контроллера не изменится ничего. Экран, следящий за ним, показывает отчёт сессии, которой больше нет.

Требование раздела «Порядок обновлений общего состояния» и нужная здесь быстрота тянут в разные стороны, и одной регистрацией не сделать и то и другое.

Bloc

Выделите отзыву отдельную регистрацию, чтобы он не ждал за сборкой:

class ReportBloc extends Bloc<ReportEvent, ReportState> {
  final Reports _reports;
  final Auth _auth;

  ReportBloc(this._reports, this._auth) : super(SignedIn()) {
    _auth.onRevoked = (reason) => add(SessionRevoked(reason));
    on<SessionRevoked>((e, emit) => emit(SignedOut(e.reason)));
    on<BuildReport>((e, emit) async {
      if (state is! SignedIn) return;
      final report = await _reports.build(e.range);
      if (state is! SignedIn) return;
      emit(Ready(report));
    }, transformer: sequential());
  }

  @override
  Future<void> close() {
    _auth.onRevoked = null;
    return super.close();
  }
}

Состояние сразу становится SignedOut(signed out elsewhere), и отчёт не публикуется. Та же проверка теперь срабатывает: отзыв больше не ждёт за сборкой, а обработчик сборки и её emitter он при этом не отменяет. Вторая регистрация делает эту проверку нужной, а не заменяет её.

Cubit может отразить уведомление напрямую из метода наследника, но его асинхронным операциям всё ещё нужны такие же проверки допустимости.

Solo

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

final class ReportController extends Solo<ReportState> with SoloStream {
  final Reports _reports;
  final Auth _auth;

  ReportController(this._reports, this._auth) : super(const SignedIn()) {
    _auth.onRevoked = (reason) => externalSetState(SignedOut(reason));
  }

  Job<void> build(Range range) =>
      enqueue<SignedIn, void>(key: 'report', (ctx) async {
        final report = await ctx.join(() => _reports.build(range));
        ctx.emit(Ready(report));
      });

  @override
  void onClose() {
    _auth.onRevoked = null;
  }
}

Если SignedOut опубликовать обычной Job, она встанет в очередь за сборкой и будет ждать работу, которую сама обесценивает. externalSetState сразу обновляет состояние и проверяет работающие тела по их правилам. Здесь enqueue<SignedIn, void> разрешает сборку только при SignedIn, поэтому отзыв отменяет её с Cancelled(rules: is not SignedIn). with SoloStream добавляет stream состояний, из которого здесь читается исход; externalSetState он не нужен.

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

Обе версии останавливают слушатель авторизации, но в разные моменты. ReportBloc снимает его в close(), до закрытия bloc: отзыв, добавленный в закрытый bloc, бросил бы ошибку. ReportController снимает его в onClose, который приходит, когда все Job закончились, а isFinished ещё ложен: SoloCloseMode.drain слышит отзыв, пока крутится его очередь, а после конца до externalSetState ничего не доходит. Оставленный на месте, слушатель держал бы закрытый контроллер достижимым из auth, и отзыв бросил бы ошибку из externalSetState.

При успехе Job может закончиться публикацией Ready, хотя это состояние за пределами SignedIn. Собственный emit исключён из проверки правил; последующая контрольная точка состояния отменила бы Job. Так итоговый переход возможен, а рабочий тип обязан охватывать всё продолжение работы.

Уже начатая сборка завершается в обоих примерах. join ждёт её, прежде чем разрешить старт другой корневой Job. До работы, которую можно остановить, дополнительно дотягивается ctx.onCancel, как в разделе «Перезапуск одной операции в общей очереди». Итоговые обработчики состояния ifFailed и ifCancelled, если они заданы, запрещаются несовместимым внешним обновлением, поэтому не перезаписывают SignedOut при уборке.

Что даёт solo: externalSetState публикует SignedOut сразу, а не ставит его в очередь за сборкой, и правило enqueue<SignedIn, void> отменяет сборку с Cancelled(rules: is not SignedIn).

Завершение начатой записи перед перезапуском

Обновление прошивки последовательно записывает части через BLE. Замена должна остановить старый цикл и дождаться его текущей записи до отправки любой новой части. API BLE в этом примере не может прервать уже принятую запись.

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

restartable() и есть та политика, которую требование называет: новая загрузка заменяет работающую. Цикл написан так, будто замена его останавливает:

// Первая попытка: заменённый цикл продолжает писать.
class UnguardedFirmwareBloc extends Bloc<FirmwareEvent, FirmwareState> {
  final Ble _ble;

  UnguardedFirmwareBloc(this._ble) : super(Idle()) {
    on<Flash>((e, emit) async {
      var written = 0;
      for (final chunk in e.chunks) {
        await _ble.write(chunk);
        emit(Flashing(++written, e.chunks.length));
      }
    }, transformer: restartable());
  }
}

Первая загрузка несёт части с 0 по 5; вторая, с частями с 100 по 105, заменяет её, пока пишется часть 1. Устройство получает [0, 1, 100, 2, 101, 3, 102, 4, 103, 5, 104, 105]: оба цикла продолжили писать, и две загрузки чередуются в эфире. restartable() отменяет emitter заменённого обработчика, и отменённый emitter игнорирует записи, но тело Dart, ожидающее _ble.write, этим не прерывается. Обсуждение restartable поясняет, почему отмена не прерывает ожидаемые future.

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

Раз тело не прерывается, оно должно заметить само: emit.isDone истинно после того, как emitter отменили.

// Вторая попытка: в эфире всё ещё двое пишущих.
class FirmwareBloc extends Bloc<FirmwareEvent, FirmwareState> {
  final Ble _ble;

  FirmwareBloc(this._ble) : super(Idle()) {
    on<Flash>((e, emit) async {
      var written = 0;
      for (final chunk in e.chunks) {
        await _ble.write(chunk);
        if (emit.isDone) return;
        emit(Flashing(++written, e.chunks.length));
      }
    }, transformer: restartable());
  }
}

Теперь старый цикл останавливается: тот же перезапуск записывает [0, 1, 100, 101, …]. Но устройство по-прежнему видит двух пишущих. Новый обработчик начинается, пока старая запись ещё ожидается, и трасса выглядит так: [write 0 start, write 0 end, write 1 start, write 100 start, write 1 end, write 100 end, …]. Проверка распоряжается тем, что публикуется, а не тем, что происходит в эфире.

Bloc

Общая блокировка упорядочивает вызовы устройства:

class Lock {
  Future<void> _tail = Future.value();

  Future<T> protect<T>(Future<T> Function() action) {
    final turn = Completer<void>();
    final before = _tail;
    _tail = turn.future;
    return before.then((_) => action()).whenComplete(turn.complete);
  }
}

class LockedFirmwareBloc extends Bloc<FirmwareEvent, FirmwareState> {
  final Ble _ble;
  final _wire = Lock();

  LockedFirmwareBloc(this._ble) : super(Idle()) {
    on<Flash>((e, emit) async {
      var written = 0;
      for (final chunk in e.chunks) {
        await _wire.protect(() async {
          if (emit.isDone) return;
          await _ble.write(chunk);
        });
        if (emit.isDone) return;
        emit(Flashing(++written, e.chunks.length));
      }
    }, transformer: restartable());
  }
}

Теперь трасса ставит конец каждой записи до начала следующей: [write 0 start, write 0 end, write 1 start, write 1 end, write 100 start, write 100 end, …].

Проверяйте emit.isDone и внутри блокировки, и после записи. Событие может быть заменено во время ожидания блокировки; проверка только перед входом позволила бы его устаревшей записи начаться позднее. Блокировка упорядочивает доступ к устройству, а проверка emitter останавливает устаревшие обработчики. Все операции с этим устройством должны соблюдать то же правило блокировки.

Отдельное событие, публикующее Broken, не меняет isDone этого emitter. В таком сценарии всё ещё нужны дополнительные проверки состояния, чтобы остановить прошивку и сохранить Broken, как в разделе «Реакция на независимое внешнее изменение состояния».

Solo

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

final class FirmwareController extends Solo<FirmwareState> {
  final Ble _ble;

  FirmwareController(this._ble) : super(const Idle());

  Job<void> flash(List<Chunk> chunks) => enqueue<NotBroken, void>(
    key: 'flash',
    policy: Policy.restart,
    (ctx) async {
      var written = 0;
      for (final chunk in chunks) {
        await ctx.join(() => _ble.write(chunk));
        ctx.emit(Flashing(++written, chunks.length));
      }
    },
  );
}

Policy.restart запрашивает отмену и ставит замену в очередь. join ждёт текущую запись и, когда та возвращается, бросает Cancelled, до следующей итерации; заменённая загрузка заканчивается на Cancelled(replaced). Замена начинается после завершения этой Job. Устройство получает части в том же порядке, что у bloc с блокировкой, [0, 1, 100, 101, …], и ни одна запись не начинается раньше, чем кончилась предыдущая.

Внешнее состояние Broken также отменяет эту Job, поскольку больше не соответствует NotBroken. Когда состояние становится Broken во время третьей записи, загрузка останавливается после [0, 1, 2] с Cancelled(rules: is not NotBroken). Одна контрольная точка в join охватывает и замену, и потерю допустимости состояния.

Если загрузка запускает детей через ctx.run, родитель ждёт и этих детей, и освобождение их ресурсов. Работа, которой намеренно разрешено пережить Job, может использовать ctx.unattended: её ошибки передаются хукам Job, но она не удерживает очередь.

Что даёт solo: ctx.join вокруг каждой записи служит единственной контрольной точкой: заменённая загрузка кончается Cancelled(replaced) после своей текущей записи, записи не перекрываются, а состояние Broken останавливает загрузку тем же путём.

Освобождение ресурса, полученного после отмены

Аудиоредактор открывает нативный PCM-буфер для отрисовки волны. Пользователь выбирает другой клип, пока декодирование ещё ожидается. Старую волну нужно отбросить, но её буфер всё равно необходимо освободить. Эти вызовы декодера могут безопасно выполняться одновременно, поскольку буферы независимы. Декодеру с последовательным доступом нужно ожидание из раздела «Завершение начатой записи перед перезапуском».

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

Проверка emit.isDone перед использованием результата предотвращает устаревшее обновление волны, и после раздела «Завершение начатой записи перед перезапуском» эта проверка ставится рефлекторно:

// Первая попытка: устаревший буфер так и не освобождается.
class PreviewBloc extends Bloc<OpenPreview, PreviewState> {
  final Decoder _decoder;

  PreviewBloc(this._decoder) : super(const NoPreview()) {
    on<OpenPreview>((e, emit) async {
      final buffer = await _decoder.open(e.clip);
      if (emit.isDone) return;
      emit(Preview(buffer.waveform()));
      await buffer.release();
    }, transformer: restartable());
  }
}

Пользователь открывает клип 1, а затем клип 2, пока первый ещё декодируется, и декодер отвечает на второй запрос первым. Состояние заканчивается на Preview(2), и это верно, а устаревший буфер освобождается ноль раз: ранний возврат оставляет его сборщику мусора, который не умеет освобождать нативный буфер. По состоянию об этом не сказано ничего.

Bloc

Перенесите проверку внутрь try/finally, который владеет полученным буфером:

class GuardedPreviewBloc extends Bloc<OpenPreview, PreviewState> {
  final Decoder _decoder;

  GuardedPreviewBloc(this._decoder) : super(const NoPreview()) {
    on<OpenPreview>((e, emit) async {
      final buffer = await _decoder.open(e.clip);
      try {
        if (emit.isDone) return;
        emit(Preview(buffer.waveform()));
      } finally {
        await buffer.release();
      }
    }, transformer: restartable());
  }
}

Оба варианта заканчиваются на Preview(2). Первая попытка освобождает один буфер из двух, текущий; защищённый вариант освобождает оба, каждый по разу. finally выполняется и на отменённом пути, и только поэтому устаревший буфер вообще освобождается.

Обработчик продолжает ожидание после отмены emitter и освобождает буфер при его получении. Это рабочий способ управления ресурсом; каждый выход после получения должен оставаться внутри блока try. Дополнительные ресурсы или передача владения требуют соответствующих решений об уборке. Cubit может использовать тот же подход с собственной проверкой устаревшего запроса, поскольку у него нет emitter обработчика и emit.isDone.

Solo

Зарегистрируйте освобождение на вызове, который получает буфер:

final class PreviewController extends Solo<PreviewState> {
  final Decoder _decoder;

  PreviewController(this._decoder) : super(const NoPreview());

  Job<void> open(Clip clip) => enqueue<PreviewState, void>(
    key: 'preview',
    policy: Policy.restart,
    (ctx) async {
      final buffer = await ctx.abandonable(
        () => _decoder.open(clip),
        dispose: (buffer) => buffer.release(),
      );
      ctx.emit(Preview(buffer.waveform()));
    },
  );
}

ctx.abandonable может закончить отменённую Job до конца декодирования. Поздний буфер всё равно передаётся в dispose; буфер, полученный при активной Job, освобождается во время её уборки. Поэтому новая Job может начаться, не ожидая устаревшего декодирования, а каждый буфер будет освобождён.

Здесь нужен dispose, поскольку буфер временный, в том числе при успехе. discard предназначен для ресурса, передаваемого вызывающему коду как успешный результат; он не освободил бы этот временный буфер после успешного обновления волны.

Защищённый bloc и этот контроллер оба проходят [open 1, open 2, ready 2, sample 2, release 2, ready 1, release 1] и заканчиваются на Preview(2).

Позднее освобождение может произойти после закрытия контроллера. В solo close() заканчивается до появления устаревшего буфера; он освобождается, когда декодирование наконец его возвращает. Используйте join, когда и операция, и освобождение её ресурса должны закончиться до продвижения очереди или закрытия контроллера.

Что даёт solo: dispose у ctx.abandonable освобождает каждый буфер, поздний тоже, а новая Job начинается, не ожидая устаревшего декодирования.