Flutter bloc with streams

I'm trying to use flutter bloc pattern in my application. It uses quick_blue library to scan BLE devices. QuickBlue provides a streams with scan results. My goal is to start scan with one event(emit new state each time device is found) and stop scan with another event.

Problem: I am confused how to cancel an active scan event. Since scanResultStream comes from 3rd party library, I can't control it. So, emit.onEach stays active until another StartScanEvent fires (thanks to restartable()).

I've looked in flutter bloc with stream example, but there stream is created inside application, so user can full control over it. I'd appreciate any advice.

Here's my bloc code:

class BLEBloc extends Bloc<BLEEvent, BLEState> {

final List<BlueScanResult> scanResult = [];
final Stream scanStream = QuickBlue.scanResultStream;

BLEBloc() : super(BLEInitState()) {
  on<StartScanEvent>((event, emit) async {
    QuickBlue.startScan();
    scanResult.clear();

    // want to cancel this 
    await emit.onEach(
      scanStream,
      onData: (data) {
        scanResult.add(data);
        emit(DeviceFoundState(scanResult));
      },
    );

    print("finished"); // finishes only on next StartScanEvent
  }, transformer: restartable());

  on<StopScanEvent>(
    (event, emit) {
      QuickBlue.stopScan();
      emit(ScanFinishedState());
    },
  );
}
}

I think I've found solution.

Stream subscription should fire different event when listening to the stream. And this new event should emit state.

Here is the example code:

class BBloc extends Bloc<MyEvent, MyState> {
  var counterStream = Stream<int>.periodic(const Duration(seconds: 1), (x) => x)
      .asBroadcastStream();
  StreamSubscription? _subscription;
  BBloc() : super(MyInitalState()) {
    on<PressedEvent>((event, emit) {
      _handlePressedEvent(event, emit);
    });
    on<SubEvent>(
      (event, emit) {
        emit(StateA());
      },
    );
  }
  void _handlePressedEvent(PressedEvent event, Emitter<MyState> emit) {
    // Cancel previous subscription if it exists
    _subscription?.cancel();

    // Start a new subscription
    _subscription = counterStream.listen((data) {
      print(data);
      add(SubEvent());
    });
  }
}