Flutter Bloc监听Stream时的超时处理异常问题

在Flutter应用中使用Bloc库处理事件时，监听Stream并希望处理超时情况，原代码如下：

await emit.onEach(
  bluetoothConnection.input!,
  onData: (data) {
    answer = answer + String.fromCharCodes(data);
    if (answer.contains('hello:ok')) {
      isAnswered = true;
      emit(
        ConnectionSuccessState(
          device: event.device.copyWith(
            isConnected: true,
            bluetoothConnection: bluetoothConnection,
          ),
        ),
      );
    }
  },
).timeout(const Duration(seconds: 1), onTimeout: () {
  debugPrint('onTimeout');
});

当超时发生时，能看到调试日志：

I/flutter ( 5875): onTimeout

但随后会抛出未处理异常：

E/flutter ( 5875): [ERROR:flutter/runtime/dart_vm_initializer.cc(41)] Unhandled Exception: 'package:bloc/src/emitter.dart': Failed assertion: line 150 pos 7: '_disposables.isEmpty': 
E/flutter ( 5875): 
E/flutter ( 5875): An event handler completed but left pending subscriptions behind.
E/flutter ( 5875): This is most likely due to an unawaited emit.forEach or emit.onEach. 
E/flutter ( 5875): Please make sure to await all asynchronous operations within event handlers.
E/flutter ( 5875): 
E/flutter ( 5875):   **BAD**
E/flutter ( 5875):   on<Event>((event, emit) {
E/flutter ( 5875):     emit.forEach(...);
E/flutter ( 5875):   });  
E/flutter ( 5875):   
E/flutter ( 5875):   **GOOD**
E/flutter ( 5875):   on<Event>((event, emit) async {
E/flutter ( 5875):     await emit.forEach(...);
E/flutter ( 5875):   });

正确处理方式 #

问题根源在于timeout操作默认会抛出TimeoutException，即便提供了onTimeout回调，该异常仍会触发，导致emit.onEach内部的Stream订阅未被正确清理，从而触发Bloc的断言错误。

方案一：在Stream层面处理超时并关闭流

将timeout直接附加到原始Stream上，在onTimeout中关闭流并发送超时状态，确保整个Stream正常结束：

await emit.onEach(
  bluetoothConnection.input!.timeout(
    const Duration(seconds: 1),
    onTimeout: (sink) {
      debugPrint('onTimeout');
      // 发送自定义的超时状态
      emit(ConnectionTimeoutState(device: event.device));
      sink.close(); // 关闭流，终止订阅
    },
  ),
  onData: (data) {
    answer = answer + String.fromCharCodes(data);
    if (answer.contains('hello:ok')) {
      isAnswered = true;
      emit(
        ConnectionSuccessState(
          device: event.device.copyWith(
            isConnected: true,
            bluetoothConnection: bluetoothConnection,
          ),
        ),
      );
    }
  },
);

方案二：捕获超时异常

通过try-catch捕获timeout抛出的TimeoutException，确保emit.onEach的异步操作正常完成：

try {
  await emit.onEach(
    bluetoothConnection.input!,
    onData: (data) {
      answer = answer + String.fromCharCodes(data);
      if (answer.contains('hello:ok')) {
        isAnswered = true;
        emit(
          ConnectionSuccessState(
            device: event.device.copyWith(
              isConnected: true,
              bluetoothConnection: bluetoothConnection,
            ),
          ),
        );
      }
    },
  ).timeout(const Duration(seconds: 1), onTimeout: () {
    debugPrint('onTimeout');
    emit(ConnectionTimeoutState(device: event.device));
  });
} on TimeoutException catch (_) {
  // 捕获超时异常，避免未处理异常抛出
}

核心原理 #

Bloc要求事件处理函数中的所有异步操作必须被await，且不能留下未完成的Stream订阅。当超时触发时，若未处理TimeoutException，emit.onEach的Future会被异常终止，内部订阅未被清理，就会触发_disposables.isEmpty的断言错误。

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