Nsryjdtyk

I have the following use case of a simple BLE device setup process using RxAndroidBle:

1. Connect to a BLE device.


2. Start listening to notification characteristic and set up a parser to parse each incoming notification. Parser will then use a PublishSubject to publish parsed data.


3. Perform a write to write characteristic (negotiate secure connection).


4. Wait for parser PublishSubject to deliver the parsed response from device - public key (which arrived through the notification characteristic as a response to our write).


5. Perform another write to the write characteristic (set connection as secure).


6. Deliver a Completable saying if the process has completed successfully or not.

Right now my solution (not working) looks like this:

deviceService.connectToDevice(macAddress)

    .andThen(Completable.defer { deviceService.setupCharacteristicNotification() })

    .andThen(Completable.defer { deviceService.postNegotiateSecurity() })

    .andThen(Completable.defer {

        parser.notificationResultSubject

            .flatMapCompletable { result ->

                when (result) {

                    DevicePublicKeyReceived -> Completable.complete()

                    else -> Completable.error(Exception("Unexpected notification parse result: ${result::class}"))

                }

            }

    })

    .andThen(Completable.defer { deviceService.postSetSecurity() })

And the DeviceService class:

class DeviceService {



    /**

     * Observable keeping shared RxBleConnection for reuse by different calls

     */

    private var connectionObservable: Observable<RxBleConnection>? = null



    fun connectToDevice(macAddress: String): Completable {

        return Completable.fromAction {

            connectionObservable = 

                rxBleClient.getBleDevice(macAddress)

                .establishConnection(false) 

                .compose(ReplayingShare.instance())

        }

    }



   fun setupCharacteristicNotification(): Completable =

        connectionObservable?.let {

            it

                .switchMap { connection ->

                    connection.setupNotification(UUID_NOTIFICATION_CHARACTERISTIC)

                        .map { notificationObservable -> notificationObservable.doOnNext { bytes -> parser.parse(bytes) }.ignoreElements() }

                        .map { channel ->

                            Observable.merge(

                                Observable.never<RxBleConnection>().startWith(connection),

                                channel.toObservable()

                            )

                        }

                        .ignoreElements()

                        .toObservable<RxBleConnection>()

                }

                .doOnError { Timber.e(it, "setup characteristic") }

                .take(1).ignoreElements()

        } ?: Completable.error(CONNECTION_NOT_INITIALIZED)



   fun postNegotiateSecurity(): Completable {

        val postLength = negotiateSecurity.postNegotiateSecurityLength()

        val postPGK = negotiateSecurity.postNegotiateSecurityPGKData()



        return connectionObservable?.let {

            it.take(1)

                .flatMapCompletable { connection ->

                    postLength

                        .flatMapSingle { connection.write(it.bytes.toByteArray()) }

                        .doOnError { Timber.e(it, "post length") }

                        .flatMap {

                            postPGK

                                .flatMapSingle { connection.write(it.bytes.toByteArray()) }

                                .doOnError { Timber.e(it, "post PGK") }

                        }

                        .take(1).ignoreElements()

                }

        } ?: Completable.error(CONNECTION_NOT_INITIALIZED)

    }



    fun postSetSecurity(): Completable =

        connectionObservable?.let {

            it.take(1)

                .flatMapCompletable { connection ->

                    negotiateSecurity.postSetSecurity()

                        .flatMapSingle { connection.write(it.bytes.toByteArray()) }

                        .take(1).ignoreElements()

                }

        } ?: Completable.error(CONNECTION_NOT_INITIALIZED)

   }



private fun RxBleConnection.write(bytes: ByteArray): Single<ByteArray> =

    writeCharacteristic(UUID_WRITE_CHARACTERISTIC, bytes)

The problem is that it gets stuck in deviceService.postNegotiateSecurity() and never gets past. I don't get any data in the parser as well, so I assume I'm incorrectly subscribing to the notification characteristic.

negotiateSecurity.postNegotiateSecurityLength() and negotiateSecurity.postNegotiateSecurityPGKData() are methods which prepare data to be sent and deliver it as Observable<SendFragment>. Because of data frame size limit, one frame might be encoded as several fragments, which are then emitted by these Observables.