Dart如何基于一个Stream的事件动态返回对应Stream，无需第三方包

原有写法问题分析 #

• 第一种map写法：map操作只是将stream1的每个事件转换为Stream<Data2?>类型，最终得到的是嵌套流Stream<Stream<Data2?>>，和返回值类型不匹配，自然无法正常运行。
• 第二种await for写法问题：
  1. 逻辑上属于串行处理，必须等待当前子流全部发射完毕后，才会处理stream1的下一个事件，如果你需要的是stream1出新事件时立刻切换到新的子流、停止接收旧子流的事件，该写法完全不符合需求。
  2. 如果你代码中提前消费了单订阅类型的stream1（比如你注释的map操作已经监听了一次stream1），第二次用await for监听就会出现异常，单订阅流仅允许被监听一次。

解决方案 #

场景1：新事件触发时切换到最新子流（匹配绝大多数动态切换需求） #

核心思路是通过StreamController手动管理上下游订阅，每次收到stream1的新事件时，先取消上一个子流的订阅，再监听新的子流并转发事件：

Stream<Data2?> getStream12() {
  // 若需要多订阅可改为 StreamController<Data2?>.broadcast()
  final controller = StreamController<Data2?>();
  StreamSubscription<Data1?>? stream1Sub;
  StreamSubscription<Data2?>? stream2Sub;

  controller.onListen = () {
    stream1Sub = getStream1().listen(
      (event) async {
        // 取消旧子流订阅，避免收到旧数据
        await stream2Sub?.cancel();
        stream2Sub = null;

        if (event == null) {
          if (!controller.isClosed) controller.add(null);
          return;
        }

        // 监听新的子流
        stream2Sub = getStream2(event.var).listen(
          (data2) => !controller.isClosed ? controller.add(data2) : null,
          onError: (e, st) => !controller.isClosed ? controller.addError(e, st) : null,
          onDone: () {
            stream2Sub = null;
            // 上游流已结束且没有活跃子流时关闭控制器
            if (stream1Sub == null && !controller.isClosed) controller.close();
          },
        );
      },
      onError: (e, st) => !controller.isClosed ? controller.addError(e, st) : null,
      onDone: () {
        stream1Sub = null;
        if (stream2Sub == null && !controller.isClosed) controller.close();
      },
    );
  };

  // 处理监听取消逻辑，避免内存泄漏
  controller.onCancel = () async {
    await stream1Sub?.cancel();
    await stream2Sub?.cancel();
    stream1Sub = null;
    stream2Sub = null;
    if (!controller.isClosed) await controller.close();
  };

  // 若需要支持暂停/恢复可放开以下注释
  // controller.onPause = () {
  //   stream1Sub?.pause();
  //   stream2Sub?.pause();
  // };
  // controller.onResume = () {
  //   stream1Sub?.resume();
  //   stream2Sub?.resume();
  // };

  return controller.stream;
}

场景2：串行处理所有子流（适合子流必须完整执行的场景） #

如果你确定需要等前一个子流全部执行完毕再处理下一个stream1事件，只需要保证stream1仅被监听一次即可，修正写法如下：

Stream<Data2?> getStream12() async* {
  // 直接在await for中创建并监听stream1，避免提前消费
  await for (var event in getStream1()) {
    if (event == null) {
      yield null;
    } else {
      yield* getStream2(event.var);
    } 
  }
}

内容的提问来源于stack exchange，提问作者Ghost