How do I stop RxSwift ble scanner once it has found a match?

I have a ble scanner that works and looks like this:

func scan(serviceId: String) -> Observable<[BleHandler.BlePeripheral]> {
    knownDevices = []
    return waitForBluetooth()
        .flatMap { _ in self.scanForPeripheral(serviceId: serviceId) }
        .map { _ in self.knownDevices }
}

private func waitForBluetooth() -> Observable<BluetoothState> {
    return self.manager
        .observeState()
        .startWith(self.manager.state)
        .filter { $0 == .poweredOn }
        .take(1)
}

Then in the viewModel class it filters matches from core data:

func scanAndFilter() -> Observable<[LocalDoorCoreDataObject]> {
        let persistingDoors: [LocalDoorCoreDataObject] = coreDataHandler.fetchAll(fetchRequest: NSFetchRequest<LocalDoorCoreDataObject>(entityName: "LocalDoorCoreDataObject"))

    return communicationService
        .scanForDevices(register: false)
        .map{ peripherals in
            print("🐶 THIS WILL GO ON FOR ETERNITY", peripherals.count)
            self.knownDevices = peripherals
            return persistingDoors
                .filter { door in peripherals.contains(where: { $0.identifier.uuidString == door.dPeripheralId }) }
        }
}

And in the view I want to connect when the scan is completed:

private func scanAndConnect(data: LocalDoorCoreDataObject) {
    viewModel.scanRelay().subscribe(
        onNext: {
            print("🐶SCANNED NAME", $0.first?.dName)},
        onCompleted: {
            print("🐶COMPLETED SCAN")
            self.connectToFilteredPeripheral(localDoor: data)
    }).disposed(by: disposeBag)
}

It never reaches onCompleted as it will just scan for eternity even after having found and filtered the core data match. In Apple's framework coreBluetooth I could simply call manager.stopScan() after it has found what I want, but that doesn't seem to be available on the Rx counterpart. How does it work for RxSwift

You can create a new Observable that looks for devices and then completes as soon as it finds the device(s) you're looking for. This would be something like:

func scanAndFilter() -> Observable<[LocalDoorCoreDataObject]> {
        return Observable.deferred { in
            let persistingDoors: [LocalDoorCoreDataObject] = coreDataHandler.fetchAll(fetchRequest: NSFetchRequest<LocalDoorCoreDataObject>(entityName: "LocalDoorCoreDataObject"))

            return communicationService
                .scanForDevices(register: false)
                .filter { /* verify if the device(s) you're looking for is/are in this list */ }
                .take(1)
        }
    }

The filter operator will make sure that only lists that contain the device you're looking for are passed on and the take(1) operator will take the first emitted value and complete immediately.

The deferred call makes sure that the fetch request that is performed in the first line is not executed when you call scanAndFilter() but only when somebody actually subscribes to the resulting Observable.