如何测试已绑定监听器的StreamController？

问题场景 #

我定义了如下Dart类Foo<T>，在构造函数中已经为StreamController的流绑定了内部监听器_aListner：

class Foo<T> {
  late final StreamController<T> _streamController;

  Foo(){
    _streamController = StreamController<T>();
    // 注：原代码此处存在错误，正确写法应为直接传入_aListner，而非()=>_aListner
    _streamController.stream.listen((T data)=> _aListner);
  }
  
  void _aListner(T data){
    // 处理数据
    print(data);
  }
  
  void addData(T data){
    _streamController.add(data);
  }
  
  Stream<T> get stream => _streamController.stream;
}

我需要验证数据是否按顺序添加到流中，希望使用StreamMatcher的emitsInOrder或emits方法做测试，但测试时始终抛出**"流已被监听"**的错误。测试代码示例：

test('stream getter test.', ()async {
  final Foo<String> foo = Foo<String>();  
  const String data1 = 'data1';
  const String data2 = 'data2';
  const String data3 = 'data3';
  const String data4 = 'data4';

  foo.addData(data1);
  foo.addData(data2);
  foo.addData(data3);
  foo.addData(data4);

  print(foo.stream.length); // 此处也会抛出Bad state: Stream has already been listened to.
});

已知将_streamController改为广播流可解决问题，但类设计仅需单个监听器，不想修改此结构；尝试用mockito模拟类，但不确定是否支持构造函数模拟。核心需求是使用StreamMatcher的相关方法完成测试。

解决方案 #

1. 注入自定义StreamController（推荐） #

修改Foo类的构造函数，允许外部注入StreamController，这样测试时可以用自己的控制器捕获流数据，同时保留原类的单订阅流逻辑：

class Foo<T> {
  late final StreamController<T> _streamController;

  // 允许注入StreamController，默认创建新实例
  Foo([StreamController<T>? controller]){
    _streamController = controller ?? StreamController<T>();
    // 修复原监听器绑定错误
    _streamController.stream.listen(_aListner);
  }
  
  void _aListner(T data){
    print(data);
  }
  
  void addData(T data){
    _streamController.add(data);
  }
  
  Stream<T> get stream => _streamController.stream;
}

测试时，创建自己的StreamController并传入Foo，直接监听这个测试用的控制器来验证数据顺序：

test('验证数据按顺序流入', () async {
  final testController = StreamController<String>();
  final Foo<String> foo = Foo<String>(testController);

  // 使用StreamMatcher的emitsInOrder验证顺序
  expect(testController.stream, emitsInOrder(['data1', 'data2', 'data3', 'data4']));

  foo.addData('data1');
  foo.addData('data2');
  foo.addData('data3');
  foo.addData('data4');

  await testController.close();
});

2. 记录监听器的执行痕迹 #

如果不想修改原类的构造函数，可以在Foo类中添加一个公开的列表，记录_aListner收到的数据，通过断言这个列表的顺序来验证：

class Foo<T> {
  late final StreamController<T> _streamController;
  // 公开用于测试的记录列表
  final List<T> receivedData = [];

  Foo(){
    _streamController = StreamController<T>();
    _streamController.stream.listen(_aListner);
  }
  
  void _aListner(T data){
    print(data);
    receivedData.add(data); // 记录收到的数据
  }
  
  void addData(T data){
    _streamController.add(data);
  }
  
  Stream<T> get stream => _streamController.stream;
}

测试代码直接断言列表内容：

test('验证数据顺序', () async {
  final Foo<String> foo = Foo<String>();
  const expectedData = ['data1', 'data2', 'data3', 'data4'];

  for (final data in expectedData) {
    foo.addData(data);
  }
  // 等待流处理完成
  await foo._streamController.close();

  expect(foo.receivedData, equals(expectedData));
});

3. 修复原类的监听器绑定错误 #

注意原代码中监听器的绑定存在错误：_streamController.stream.listen((T data)=> _aListner); 这种写法只是返回_aListner函数，并不会执行它。正确的绑定方式是直接传入_aListner：_streamController.stream.listen(_aListner);，否则内部监听器根本不会处理数据，这是必须先修复的问题。

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