asynchronous - 我怎样才能正确地从使用 Futures/async/await 的流中产生？

Question

我遇到了一个奇怪的问题，如果我在我的 flutter 应用程序中从我的提供者那里 yield*，函数中的其余代码就不会完成。

我使用的是 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*。

不过，为了“订阅”此服务的进度，我需要将流返回给调用者，在 UI 上写入更新等。在我看来，这就像移动 yield 一样简单*到前先等待。像这样。

 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 调用上设置断点表明这些永远不会被调用。我坚持这个，知道它可能是什么吗？我知道这与我混合 Futures 和 Streams 的方式有关。

Answer 1

yield* 等待它返回的流完成。

在这种情况下，您希望立即返回一个流，然后将一些数据异步提供给该流。

是否有任何else 向流 Controller 添加事件？如果没有，你应该能够做到:

@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!!!...");
}

不需要流 Controller 。

如果其他功能也添加到流 Controller ，那么您确实需要它。然后，您必须将创建的流分成一个更新流 Controller 的异步部分和一个返回流的同步部分。也许:

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;
}