带有替换和取消的 Flutter 多个流

如何解决带有替换和取消的 Flutter 多个流

我在 rxDart 0.26.0 中使用 redux_epics 并试图实现以下目标：

制作多组聊天应用程序。需要在创建的每个聊天组上开始流式聊天（firestore 集合）。当群组被删除或退出时也取消流。

Stream<dynamic> streamGroupChat(
    Stream<dynamic> actions,EpicStore<AppState> store) {
  return actions
      .where((action) =>
          action is LoadGroupSucceededAction && action.group != null)
      .flatMap((action) {
    return _streamChat(action.group!.id!)
        .takeuntil(actions
            .where((action) => action is ClearChatsByGroupAction)
            .where((newAction) => newAction.groupId == action.group!.id!))
        .map((e) {
      return LoadChatsSucceededAction(e);
    });
  });
}

我面临的问题是： 当我使用 flatMap 时，我将所有聊天流加载到商店...但 takeuntil（取消）不起作用。 如果我将 flatMap 更改为 switchMap，takeuntil 开始工作，但只有最后一个 LoadGroupSucceededAction 被流式传输（覆盖以前的）。 我认为 mergeMap 是关键，但在 RxDart 中找不到。

如何为新组添加 (flatMap) 多个流，为重新加载的组替换 (switchMap) 现有流，以及取消 (takeuntil) 流？

谢谢。