I have a problem with subscribe multiple different characteristic types in my application. At first I connect to the device and declare Observable values

let peripheralObservable <- let with information about connected device
let dataStream1Observable = peripheralObservable
        .flatMap { (peripheral) -> Observable<Characteristic> in
            return peripheral.setNotificationAndMonitorUpdates(for: "some ID")
        }.map { (characteristic) -> Data? in
            return characteristic.value
        }
    let isConnectedObservable = peripheralObservable
        .flatMap { (peripheral: Peripheral) -> Observable<Bool> in
            return peripheral.rx_isConnected
        }

Then I subscribed first value (isConnectedObservable), everything works good. Next I call dispose on this subscribtion and start new one on dataStream1Observable, then I wait a moment and I get error "timeout error". What I do wrong? It is a error in my application or in this library. I really need help.

Hello, I'm fighting an issue currently involving "orphan" connections, where the app has disconnected however iOS is maintaining the physical link indefinitely until the peripheral power is disconnected. This has the effect of not only making the peripheral unable to be connected to again by the app, but also blocking all other devices from seeing it when scanning.

While trying to fix the underlying issue I updated my scan code to check if one of our devices was already connected before beginning a scan. The issue I'm having is retrieveConnectedPeripherals is returning no devices even when in this "orphan" state. The Observable<[Peripheral]> array is always [].

Looking at the CoreBluetooth method: centralManager.retrieveConnectedPeripherals(withServices: serviceUUIDs).map(RxCBPeripheral.init))

The documentation says: "The list of connected peripherals can include those that are connected by other apps and that will need to be connected locally using the connect(_:options:) method before they can be used."

Yet I don't see a way to pass those options in, which seems important if the method is supposed to return a peripherial. It's possible this is not related to why it's not working, but I thought I would ask.

Thanks in advance for any help you can offer!

Hi, Ive been trying to write (chunked) and read large values, found some stuff in the issues but still no luck here is what Im doing now, it never reaches to step2 of handshake Thanks in advance, love the project btw.

override func viewDidLoad() {
    super.viewDidLoad()

    manager.rx_state
        .filter { $0 == .poweredOn }
        .take(1)
        .flatMap { _ in self.manager.scanForPeripherals(withServices: nil) }
        .filter { $0.peripheral.name == "TT 3015" }.take(1)
        .flatMap { $0.peripheral.connect() }
        .flatMap { MTKHandshakeProcessor(peripheral: $0).process(libCommand: self.libCommand) }
        .subscribe(onNext: { response in
            print("NEXT -> \(response.data.utf8String ??? "")")
        }).addDisposableTo(disposeBag)
}

enum MTKHandshake {
    case step1
    case step2
    
    func payload(libCommand: LibCommandGateway) -> Data {
        switch self {
        case .step1: return libCommand.dataCommand(2, command: "REQV")
        case .step2: return libCommand.dataCommand(2, command: "0 8 1.1 IOS")
        }
    }
}


struct MTKHandshakeProcessor {
    let peripheral: Peripheral
    
    init(peripheral: Peripheral) {
        self.peripheral = peripheral
    }
    
    func process(libCommand: LibCommandGateway) -> Observable<CommandResult> {
        let processor = MTKResponseProcessor()
        let handler = { (data: Data) in
            processor.findCommandResponse(data: data, libCommand: libCommand)
        }
        
        return peripheral
            .writeAndRead(data: MTKHandshake.step1.payload(libCommand: libCommand), untilFound: handler)
            .flatMap { _ in self.peripheral.writeAndRead(data: MTKHandshake.step2.payload(libCommand: libCommand), untilFound: handler) }
    }
    
}

struct CommandResult {
    let data: Data
}

extension Peripheral {

    func writeChunked(writer: Characteristic, data: Data) -> Observable<Characteristic> {
        let chunks = data.chunked(byMaxSize: maximumWriteValueLength(for: .withResponse))
        return Observable.from(chunks).flatMap { writer.writeValue($0, type: .withResponse) }
    }
    
    func writeAndRead(data dataToSend: Data, untilFound: @escaping ((Data) -> BinaryCommand?)) -> Observable<CommandResult> {
        
        let pendingData = NSMutableData()
        
        // This observable will take care of accumulating data received from notifications
        let result = characteristic(with: MTKDeviceCharacteristic.read)
            .flatMap { $0.monitorValueUpdate() }
            .takeWhile { characteristic in
                return untilFound(pendingData as Data) != nil
            }
            .reduce(pendingData, accumulator: { (data, characteristic) -> NSMutableData in
                if let packetData = characteristic.value {
                    data.append(packetData)
                }
                return data
            })
        
        
        let writes = characteristic(with: MTKDeviceCharacteristic.write)
            .flatMap { self.writeChunked(writer: $0, data: dataToSend) }
        
        return Observable.zip(result, writes, resultSelector: { (result: NSMutableData, query: Characteristic) -> CommandResult in
            return CommandResult(data: result as Data)
        })
    }
}

extension Data {
    func chunked(byMaxSize: Int) -> [Data] {
        return (self as NSData).dataChunks(ofSize: UInt(byMaxSize))
    }
}

Hi,

Thanks for this awesome work.

I'm having some trouble to find a way to correctly restore my device connection if any disconnect or bluetooth state trigger an error.

It's probably due to my skill level in Rx but I would love your insights.

I'm trying to create a Device object which will wrap the logic of waiting for correct BLE state, scan for peripheral, connect to one with specific name, detect services & characteristics, listen to some notification and then republish the notification using some PublishSubject to be consume by my controller (or what ever else).

The issues is that I would like to reconnect all of this if any error occurs (like device disconnection, ble state changes, ...).

This code have some trivial issues, like subscribing twice to the characteristics, but I can figure out how to not do it, if I use a simple share on characteristics and the retry on my notification it won't work.

I'm having trouble wrapping my head around this...

class Device {
    let measure = PublishSubject<Data>()
    let alert = PublishSubject<Data>()
    
    private let disposeBag = DisposeBag()
    
    init(name: String) {
        let manager = BluetoothManager(queue: .main)
        
        let queue = DispatchQueue(label: "fr.xdev.rxbluetoothkit.bluetooth")
        let scheduler = SerialDispatchQueueScheduler(queue: queue, internalSerialQueueName: "")
        
        let characteristics = manager.rx_state
            .filter { $0 == .poweredOn }
            .take(1)
            .flatMap { _ in manager.scanForPeripherals(withServices: [.common(.main)]) } // extension on CBUUID
            .filter { $0.peripheral.name == name }
            .take(1)
            .flatMap { $0.peripheral.connect() }
            .flatMap { $0.discoverServices([.common(.main)]) } // extension on CBUUID
            .flatMap { Observable.from($0) }
            .flatMap { $0.discoverCharacteristics([.common(.measure), .common(.protection)]) } // extension on CBUUID
            .flatMap { Observable.from($0) }
        
        characteristics
            .filter { $0.uuid.type == .measure } // extension on CBUUID
            .flatMap { $0.setNotificationAndMonitorUpdates() }
            .map { $0.value }
            .unwrap() // ignore nil value
            .retryWithDelay(5) // !!!: retry and resubscribe to the full chain on error
            .subscribe(onNext: { [unowned self] data in
                self.measure.onNext(data!)
            })
            .disposed(by: disposeBag)
        
        characteristics
            .filter { $0.uuid.type == .alert } // extension on CBUUID
            .flatMap { $0.setNotificationAndMonitorUpdates() }
            .map { $0.value }
            .unwrap() // ignore nil value
            .retryWithDelay(5) // !!!: retry and resubscribe to the full chain on error
            .subscribe(onNext: { [unowned self] data in
                self.alert.onNext(data!)
            })
            .disposed(by: disposeBag)
    }
}

Here is an extended version of this sample project: https://gist.github.com/huguesbr/41290240fad1a4de290dd3e81305ed96

Any advices of how to organize this?

Thanks

Hi everyone, I'm trying to make periodical scan using RxBluetoothKit without success. Can anyone give some directions. I can't find something similar in the docs or sample app. What I want to do is scan for 3 seconds, pause 1 second and start scanning again for 3 seconds and so on.

Thanks.

I used to use 4 rx_isConnected for a good time. But with a 5 observeConnection, this will automatically call the disconnected method observeDisconnect after my active connection. I don't know what it was. I'm confused. @minixT

The new major release just broke my neck, I'm currently trying to update my app to support it. I used listenOnRestoredState to silent the annoying CoreBluetooth logs, but now this API is gone. My app isn't running yet so I cannot tell for sure if the warning will re-appear.

Hi,

Here is my workflow :

1. scanForPeripherals : got DeviceA
2. stopScan
3. manager.connect(DeviceA)
4. working with device...
5. manager.cancelPeripheralConnection(DeviceA)
6. scanForPeripherals : got nothing on iOS11, got DeviceA on iOS10

Here is some snippet of my code, maybe i'm doing something wrong ?

func scan(_ delegate: BluetoothClientScanDelegate) {
        scanningDisposable = manager.rx_state
            .filter { $0 == .poweredOn }
            .take(1)
            .flatMap { _ in self.manager.scanForPeripherals(withServices: [BluetoothService.device.uuid], options:nil) }
            .subscribeOn(MainScheduler.instance)
            .subscribe(onNext: { device in
                delegate.onDeviceFound(device)
            }, onError: { error in
                delegate.onScanError(error)
            })
    }

func stopScan() {
        scanningDisposable?.dispose()
    }

func connectToDevice(_ device: Peripheral, delegate: BluetoothClientDeviceDelegate) {
        deviceDelegate = delegate
        currentDevice = device

        Logger.get().info("Connecting to device...")

        let _ = manager.connect(device).subscribe(
            onNext: { _ in
                Logger.get().info("Connected to device")
                self.monitorDisconnection()
                self.setupModeData()
            },
            onError: {
                Logger.get().error($0)
                self.deviceDelegate?.onDeviceFailed($0)
            }
        )
    }

func disconnectFromDevice() {
        deviceDelegate = nil
        disconnectionDisposable?.dispose()
        notificationDisposable?.dispose()
        guard let device = currentDevice else {
            return
        }
        currentDevice = nil
        _ = manager.cancelPeripheralConnection(device).subscribe()
    }

I'm facing a weird issue (version 5.1.0 of this lib): When using the CentralManager without any init option, the observeState() observable is well working.

But as soon as I add a restore identifier key, the observeState() observable just sends one (and only one) unsupported state, then no more events even after enabling/disabling Bluetooth (either from control center or Settings). As soon as I remove the restoreIdentifier options, it works back again nicely...

Is there anything wrong I did, or an issue in the lib? Thanks for the help!

Here is the observe state part that prints everything happening there:

public func observeBluetoothState() -> Observable<BluetoothState> {
        return manager!.observeState()
            .do(
                onNext: { newstate in print("RXBT observeState onNext, new state is \(newstate)") },
                onError: { error in print("RXBT observeState onError, error is \(error)") },
                onCompleted: { print("RXBT observeState onCompleted") },
                onSubscribe: { print("RXBT observeState onSubscribe") },
                onSubscribed: { print("RXBT observeState onSubscribed") },
                onDispose: { print("RXBT observeState onDispose") }
            )
    }

CentralManager initialization without options:

var manager = CentralManager(  queue: .main )

/* result output: 
RXBT observeState onSubscribe
RXBT observeState onSubscribed
RXBT observeState onNext, new state is poweredOn
RXBT observeState onNext, new state is poweredOff
RXBT observeState onNext, new sRXBT observeState onNext, new state is poweredOff
RXBT observeState onNext, new state is poweredOn
RXBT observeState onNext, new state is poweredOff
RXBT observeState onNext, new state is poweredOn
*/

CentralManager initialization with restore options:

var manager = CentralManager(  queue: .main,
                                   options: [CBCentralManagerOptionRestoreIdentifierKey:"get.myId.centralManagerRestoreKey" as AnyObject],
                                   onWillRestoreCentralManagerState: {
                                            centralManagerRestoredState in
                                            print("RXBT centralManagerRestoredState is \(centralManagerRestoredState)")
                                            // do any stuff here later... 
                                        }
        )

/* result output: 
RXBT observeState onSubscribe
RXBT observeState onSubscribed
RXBT observeState onNext, new state is unsupported
*/

Hello! Congrats on the release!

I'm in the process of updating to the new API changes now and can't seem to find a replacement for .cancelConnection() on a peripheral.

Basically this class gets handed a peripheral, calls cancel connection, subscribes, then onNext notifies the app that the disconnect occurred.

What's the 5.0 way to do this?

Thanks!

Subsequent calls to discoverServices method in Peripheral with different services UUIDs causing it to return an empty sequence instead of success. This happens because after the first call the peripheral already has some discovered services and in the next call the comparison between sets of services identifiers fails if you ask to discover another service

Hello everyone, thank you for the great job with this library 🎉 I used this lib for my app but I facing the following problem: After I received data from 1 charateristics and continued reading another charateristics it got no response. Code:

Can you check and help me to resolve this issue? Thank!

Describe the bug I have two services in one peripheral. Service A
55AA0148-0001-55AA-1461-227D5E1F4C35 with characteristic tx: 55AA0001-0001-55AA-1461-227D5E1F4C35

Service B 55AA0248-0001-55AA-1461-227D5E1F4C35 with similar characteristic tx: 55AA0001-0001-55AA-1461-227D5E1F4C35

When I try to observe both tx characteristics, I get only one notification To Reproduce

typealias ReadCharacteristic = Observable<Data>

public class DefaultCharacteristicsService {
    public let observeConnection: Observable<Bool>
    private let bag = DisposeBag()
    private let peripheral: Peripheral
    private var peripheralObserver: Observable<Peripheral> = .empty()
    private var peripheralSubscription: SerialDisposable
    
    public init(peripheral: Peripheral) {
        self.peripheral = peripheral
        observeConnection = peripheral.observeConnection()
        peripheralSubscription = SerialDisposable()
        peripheralSubscription.disposed(by: bag)
    }
    
    public func start() -> Observable<Bool> {
        peripheralObserver = peripheral.establishConnection().share(replay: 1, scope: .forever)
        peripheralSubscription.disposable = peripheralObserver.subscribe()
        return peripheral.observeConnection()
    }
    
    public func stop() {
        peripheral.manager.manager.cancelPeripheralConnection(peripheral.peripheral)
        peripheralSubscription.disposable.dispose()
    }

    public func observeCharacteristic(service: GATTService) -> ReadCharacteristic {
        characteristic(service: service)
            .flatMap {
                $0.characteristic.properties.contains(.notify) ||
                    $0.characteristic.properties.contains(.indicate) ? $0.observeValueUpdateAndSetNotification() : $0.readValue().asObservable()
            }
            .compactMap { $0.characteristic.value }
            .debug("observeCharacteristic \(service) characteristic")
    }

    private func characteristic(service: GATTService) -> Observable<Characteristic> {
        peripheralObserver
            .filter { $0.isConnected }
            .flatMap {
                $0.discoverServices([service.serviceCBUUID])
            }
            .asObservable()
            .flatMap { Observable.from($0) }
            .flatMap {
               $0.discoverCharacteristics([service.characteristicCBUUID])
            }
            .asObservable()
            .flatMap { Observable.from($0) }
    }
}

func test(peripheral: Peripheral) {
        let service = DefaultCharacteristicsService(peripheral: peripheral)
        service.start()
        
        service.observeCharacteristic(service: .init(characteristicCBUUID: CBUUID(string: "55AA0001-0001-55AA-1461-227D5E1F4C35"),
                                                     serviceCBUUID: CBUUID(string: "55AA0148-0001-55AA-1461-227D5E1F4C35"))).subscribe(onNext: {
            print($0.debugDescription)
        }, onError: {
            print($0.localizedDescription)
        }).disposed(by: bag)
        
        service.observeCharacteristic(service: .init(characteristicCBUUID: CBUUID(string: "55AA0001-0001-55AA-1461-227D5E1F4C35"),
                                                     serviceCBUUID: CBUUID(string: "55AA0248-0001-55AA-1461-227D5E1F4C35"))).subscribe(onNext: {
            print($0.debugDescription)
        }, onError: {
            print($0.localizedDescription)
        }).disposed(by: bag)
}

It print events in first subscription

Expected behavior It print events in different subscriptions

Probably problem is here (CharacteristicNotificationManager): https://ibb.co/MCkFKtz

Environment:

• All Devices
• Library version 6.0

Thank you for your amazing job!

Describe the bug A clear and concise description of what the bug is.

To Reproduce Steps to reproduce the behavior:

1. Go to '...'
2. Click on '....'
3. Scroll down to '....'
4. See error

Expected behavior A clear and concise description of what you expected to happen.

Environment:

• Device: [e.g. iPhone6, Macbook Pro 2018]
• OS: [e.g. iOS8.1, macOS 10.11]
• Library version [e.g. 5.1.2]
• Swift version [e.g. 4.1]

When terminated the app the connection is still restored but when the device lose connection it can't reconnect again

class BluetoothManager { //MARK: - Singelton static let shared = BluetoothManager()
//MARK: - Propertise var centralManager: CentralManager!

//MARK: - Configure Central Manager
func configureCentralManager() {
    //        let options = [CBCentralManagerOptionRestoreIdentifierKey: "NOUR", CBCentralManagerOptionShowPowerAlertKey: true] as [String: AnyObject]
    
    centralManager = CentralManager(
        options: [CBCentralManagerOptionRestoreIdentifierKey: NSString("NOUR")],
        onWillRestoreCentralManagerState: { (restoreState) in
            print("RESTORED")
            print(restoreState.centralManager )
        })
}

//MARK: - Start Scanning For Peripherals
func startScanningForPeripherals() {
    _ = centralManager.observeState()
        .startWith( centralManager.state )
        .filter { $0 == .poweredOn }
        .flatMap { _ in self.centralManager.scanForPeripherals(withServices: [self.beaconServiceCBUUID]) }
        .subscribe(onNext: { [weak self] scannedPeripheral in
            guard let self = self else { return }
            self.configureScannedPeripherals(scannedPeripheral: scannedPeripheral)
        }, onError: { [weak self] error in
            guard let self = self else { return }
           print("\(error)")
            
        })
}

//MARK: - Connect To Peripheral Devices
func connectToPeripheralDevices(peripheral: Peripheral) {
    peripheral.establishConnection()
        .flatMap { $0.discoverServices(nil) }.asObservable()
        .flatMap { Observable.from($0) }
        .flatMap { $0.discoverCharacteristics(nil)}.asObservable()
        .flatMap { Observable.from($0) }
        .filter({ $0.properties.contains(.read) })
        .flatMap { $0.readValue() }
        .subscribe(onNext: {
            let data = $0.value
        }, onError: { error in
            print(error)
        }).disposed(by: disposeBag)
 }

}

// In App Delegate func application(_ application: UIApplication, didFinishLaunchingWithOptions launchOptions: [UIApplication.LaunchOptionsKey: Any]?) -> Bool { BluetoothManager.shared.configureCentralManager() }