rxdart BehaviorSubject 发射所有历史值而非仅最新值问题

问题复现 #

根据官方定义，BehaviorSubject是rxdart提供的特殊StreamController，核心特性为：捕获添加到控制器的最新条目，将其作为首个事件发送给任意新监听者，正常逻辑下仅向新订阅者发射最新值。实际使用中出现不符合预期的行为：BehaviorSubject向监听者发射了所有历史值，无法定位根因。

测试Bloc代码 #

class TestBloc {

  final _controller = BehaviorSubject.seeded([]);

  Stream get stream$ => _controller.stream;

  TestBloc(BehaviorSubject subject) {
    subject.listen((e) {
      print("listen $e");
      _controller.add([e]);
    });
  }

}

测试用例代码 #

test("test bloc", () async {
  final subject = BehaviorSubject();

  final bloc = TestBloc(subject);

  subject.add(1);
  subject.add(2);

  // 取消下一行注释可使测试通过
  // await expectLater(subject.stream, emits(2)); 

  // 此处断言失败
  await expectLater(bloc.stream$, emits([2]));
});

错误日志 #

Expected: should emit an event that [2]
  Actual: <Instance of 'BehaviorSubject<List<dynamic>>'>
   Which: emitted * []
                  * [1]
                  * [2]
            which emitted an event that at location [0] is [] which shorter than expected

问题根因 #

这不是rxdart的缺陷，是对Dart Stream默认异步调度机制不熟悉导致的使用错误，事件执行顺序如下：

1. 默认构造的BehaviorSubject是异步投递事件的，调用add()时不会立刻触发监听回调，而是将事件加入微任务队列，等当前同步代码块执行完毕后才会依次处理。
2. 测试代码中连续调用subject.add(1)、subject.add(2)时，两个事件只是进入队列，绑定在subject上的监听回调还未执行，此时bloc内部的_controller仍然保留初始化时的seeded值[]。
3. 执行到await expectLater(bloc.stream$, emits([2]))时，首先监听到_controller的当前最新值[]并作为第一个事件抛出，直接不符合[2]的预期。
4. 同步代码执行完毕后微任务队列开始处理：依次触发e=1、e=2的监听回调，向_controller添加[1]、[2]，这两个值后续也会被流发射，就出现了日志中三个值全部输出的现象。

注释掉的代码之所以能让测试通过，是因为await expectLater(subject.stream, emits(2))会等待subject的事件处理完成，此时微任务队列里的回调已经全部执行，_controller的最新值已经更新为[2]，后续再监听bloc流时只会拿到最新值。 #

解决方案 #

根据业务场景二选一即可：

• 方案1：创建BehaviorSubject时开启同步调度，add调用会立刻触发监听回调，后续监听时_controller已经更新到最新值：

  final subject = BehaviorSubject(sync: true);
  

• 方案2：如果不想修改流的同步模式，在断言前等待所有微任务处理完成，确保监听回调执行完毕再校验结果：

  subject.add(1);
  subject.add(2);
  await Future.delayed(Duration.zero); // 等待事件队列清空
  await expectLater(bloc.stream$, emits([2]));
  

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