Dart Stream的firstWhere匹配后未即时返回问题及解决方案咨询

你遇到的两个场景的问题根源分为两类：

1. 示例代码的延迟问题：sleep是Dart的同步阻塞API，会直接暂停整个Isolate的事件循环，不会让出执行权。当你执行yield 1之后，生成器函数会继续向下执行同步sleep逻辑，此时哪怕1已经被写入流队列，firstWhere的匹配、完成回调也因为事件循环被阻塞无法被调度，必须等10秒sleep结束、同步逻辑执行完毕后，事件循环才有机会处理之前的流事件，给你造成了要等到yield 2才返回的错觉。
2. 业务代码的挂起问题：

• 你每次调用lines()都会创建一个全新的Stream，默认WebSocket流是单订阅模式，且新Stream只会接收调用lines()之后新推送的WebSocket消息，如果目标CodeMessage是在调用lines()之前就已经到达，新的流监听不到历史消息，就会一直挂起等待符合条件的新消息。
• 如果你反复调用lines()监听同一个WebSocket单订阅流，后续的监听还会直接抛出异常。

示例代码修复 #

把同步阻塞的sleep替换为异步等待Future.delayed，主动让出事件循环执行权，匹配到对应值后firstWhere会立刻完成并取消流监听，不会等待后续的延迟逻辑：

Stream<int> countStream() async* {
  for (int i = 1; i <= 100; i++) {
    yield i;      
    await Future.delayed(Duration(seconds: 10)); // 替换为异步等待
  }
}

业务代码修复 #

方案1：复用Stream实例，转广播流支持多消费者 #

缓存lines()返回的Stream实例，转成广播流支持多端同时监听，避免每次调用都创建新的流：

Stream<Message>? _lineStream;
Stream<Message> lines() {
  if (_lineStream == null) {
    _lineStream = () async* {
      var partial = '';
      await for (String chunk in ws!) {
        var lines = chunk.split('\n');
        lines[0] = partial + lines[0];
        partial = lines.removeLast();
        for (final line in lines) {
          var msg = Message.parse(line);
          if (msg != null) yield msg;
        }
      }
    }().asBroadcastStream(); // 转广播流支持多订阅
  }
  return _lineStream!;
}

方案2：手动实现匹配逻辑，避免隐形资源泄漏 #

如果需要更灵活的控制（比如超时、匹配后立刻取消监听），可以手动实现订阅逻辑，不用封装好的firstWhere：

Future<Message> waitForCodeMessage({Duration? timeout}) async {
  final completer = Completer<Message>();
  StreamSubscription? sub;
  sub = lines().listen((msg) {
    if (msg is CodeMessage && !completer.isCompleted) {
      completer.complete(msg);
      sub?.cancel();
    }
  }, onError: (e, stack) {
    if (!completer.isCompleted) {
      completer.completeError(e, stack);
      sub?.cancel();
    }
  }, onDone: () {
    if (!completer.isCompleted) {
      completer.completeError(StateError('WebSocket流已关闭，未收到目标CodeMessage'));
    }
  });
  if (timeout != null) {
    completer.future.timeout(timeout, onTimeout: () {
      if (!completer.isCompleted) {
        completer.completeError(TimeoutException('等待CodeMessage超时', timeout));
        sub?.cancel();
      }
      throw ''; // 不会实际抛出，由completer控制返回
    });
  }
  return completer.future;
}

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