Как я могу правильно выйти из потока, который использует Futures / async / await?

Question

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

Я использую шаблон BLoC, поэтому моя функция _mapEventToState выглядит следующим образом:

  Stream<WizardState> _mapJoiningCongregationToState(
      int identifier, int password) async* {
    _subscription?.cancel();
    _subscription= (_provider.doThings(
            id: identifier, password: password))
        .listen((progress) => {
              dispatch(Event(
                  progressMessage: progress.progressText))
            }, onError: (error){
              print(error);
            }, onDone: (){
              print('done joiining');
            });
  }

Тогда в провайдере / сервисе ... это первая попытка.

final StreamController<Progress> _progressStream = StreamController<JoinCongregationProgress>();

 @override
  Stream<JoinCongregationProgress> doThings(
      {int id, int password}) async* {

        await Future.delayed(Duration(seconds:2));
_progressStream.add(JoinCongregationProgress(progressText: "kake1..."));
await Future.delayed(Duration(seconds:2));
_progressStream.add(JoinCongregationProgress(progressText: "kake5!!!..."));
yield* _progressStream.stream;
}

Оператор yield возвращается, но только через после обе ожидаемые функции завершены. Это имеет для меня полный смысл, очевидно, я бы не ожидал, что код завершится не по порядку и каким-то образом запустит yield *, прежде чем ждать завершения 'await's.

Для того, чтобы "подписаться" на продвижение этого сервиса, мне нужно вернуть поток обратно вызывающей стороне, написать обновления в пользовательском интерфейсе и т. Д. На мой взгляд, это так же просто, как перенести выход * до первого жду. Вот так.

 final StreamController<Progress> _progressStream = StreamController<JoinCongregationProgress>();

     @override
      Stream<JoinCongregationProgress> doThings(
          {int id, int password}) async* {
            yield* _progressStream.stream;
            await Future.delayed(Duration(seconds:2));
    _progressStream.add(JoinCongregationProgress(progressText: "kake1..."));
    await Future.delayed(Duration(seconds:2));
    _progressStream.add(JoinCongregationProgress(progressText: "kake5!!!..."));

    }

Но затем установка точек останова для последующих вызовов _progressStream.add показывает, что они никогда не будут вызываться. Я застрял на этом, есть идеи, что это может быть? Я знаю, что это как-то связано с тем, как я смешивал фьючерсы и потоки.

Answer 1

yield* ожидает завершения потока, который он возвращает.

В этом случае вы хотите немедленно вернуть поток, а затем асинхронно передать некоторые данные в этот поток.

Что-нибудь еще добавляет события в контроллер потока? Если нет, то вы должны просто сделать:

@override
Stream<JoinCongregationProgress> doThings({int id, int password}) async* {
    await Future.delayed(Duration(seconds:2));
    yield JoinCongregationProgress(progressText: "kake1...");
    await Future.delayed(Duration(seconds:2));
    yield JoinCongregationProgress(progressText: "kake5!!!...");
}

Контроллер потока не требуется.

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

final StreamController<Progress> _progressStream = StreamController<JoinCongregationProgress>();

@override
Stream<JoinCongregationProgress> doThings({int id, int password}) {
  () async {
    await Future.delayed(Duration(seconds:2));
    _progressStream.add(JoinCongregationProgress(progressText: "kake1..."));
    await Future.delayed(Duration(seconds:2));
    _progressStream.add(JoinCongregationProgress(progressText: "kake5!!!..."));
  }(); // Spin off async background task to update stream controller.
  return _progressStream.stream;
}