Dart：从另一个流的监听器中yield流事件

Question

我有一个生成特定事件stream的函数。现在我有一个来自存储服务的流，它有自己的事件。我正在寻找一种方法，在storage stream中有变化时产生我的事件。

这段代码片段无法完成此任务。

Stream<BlocState> mapEventToState(
    BlocEvent event,
  ) async* {
  if (event is UploadData) {
    yield UploadDataProgress(progress: 0.0);
    final Storage storage = Storage();
    final Stream<StorageEvent> upload = storage.upload(event.data);

    upload.listen((StorageEvent storageEvent) async* {
      print('***** Listener: ${storageEvent.type} - ${storageEvent.progress}');

      if (storageEvent.type == StorageEventType.error) {
        yield UploadDataError(errorMessage: storageEvent.error);
      }

      if (storageEvent.type == StorageEventType.success) {
        yield UploadDataSuccess();
      }

      if (storageEvent.type == StorageEventType.progress) {
        yield UploadDataProgress(progress: storageEvent.progress);
      }
    });
  }
}

输出： 调试打印可以工作，但事件不会发送到监听器。

***** Listener: StorageEventType.progress - 0.01924033836457124
***** Listener: StorageEventType.progress - 0.044581091468101464
***** Listener: StorageEventType.progress - 0.6986233206170177
***** Listener: StorageEventType.progress - 1.0

- nilobarp

---

可能是Dart/Flutter中回调函数内使用"yield"的重复问题。 - mirkancal

Answer 1

您的yield语句是从匿名函数(StorageEvent storageEvent) async* {中产生，而不是从mapEventToState中产生的。

只需将listen()替换为await for即可。

Stream<BlocState> mapEventToState(
    BlocEvent event,
  ) async* {
  if (event is UploadData) {
    yield UploadDataProgress(progress: 0.0);
    final Storage storage = Storage();
    final Stream<StorageEvent> upload = storage.upload(event.data);

    await for (StorageEvent storageEvent in upload) {
      print('***** Listener: ${storageEvent.type} - ${storageEvent.progress}');

      if (storageEvent.type == StorageEventType.error) {
        yield UploadDataError(errorMessage: storageEvent.error);
      }

      if (storageEvent.type == StorageEventType.success) {
        yield UploadDataSuccess();
      }

      if (storageEvent.type == StorageEventType.progress) {
        yield UploadDataProgress(progress: storageEvent.progress);
      }
    }
  }
}

- Gazihan Alankus

---

如果我们使用Stream<double>代替BlocState（比如用于上传进度），是否可以通过await for来捕获“完成”（listen(onDone: (here))）？ - mirkancal

是的 https://dart.dev/tutorials/language/streams#receiving-stream-events - Gazihan Alankus

Answer 2

就我所知，我曾经遇到过类似的问题，当时我正在订阅Firebase快照，并尝试基于文档数据产生事件。

如果我使用等待异步的范式，将无法获得订阅句柄。结果，在我想要关闭BLoC时停止读取流就变得困难了。

我找到了一个绕路的解决方案。

Stream<BlocState> mapEventToState(BlocEvent event) async* {
  if (event is FetchEvent) {
    yield LoadingState();
    _subscription = SomeStream(event.someKey).listen((snapshot) {
      add(OnSnapshotEvent(snapshot));
    });
  }

  if (event is OnSnapshotEvent) {
    if (SomeCondition(event.snapshot)) {
      yield SomeResultState();
    } else {
      yield SomeOtherState();
    }
  }
}

这样，我就有一个订阅的控制句柄，所以我总是可以很好地进行清理。

- Mere Vicharr