RxJava2/RxAndroidBle: subscribe to Observable from side effects

Question

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.

Answer 1

Recap:

• postNegotiateSecurity() is never completed
• negotiateSecurity.postNegotiateSecurityLength() may emit one or more times
• negotiateSecurity.postNegotiateSecurityPGKData() may emit one or more times

Analysis (omitted logs for readability):

it.take(1)
    .flatMapCompletable { connection ->
        postLength
            .flatMapSingle { connection.write(it.bytes.toByteArray()) }
            .flatMap {
                postPGK // may emit more than one value
                    .flatMapSingle { connection.write(it.bytes.toByteArray()) }
            }
            .take(1) // first emission from the above `flatMap` will finish the upstream
            .ignoreElements()
    }

Every emission from postLength will start a characteristic write. Every succeeded write will start subscription to postPGK. If postLength will emit more than once — more subscriptions to postPGK will be made.

Every subscription to postPGK most likely will result in multiple emissions. Every emission will then be flatMapped to a characteristic write. Every write succeeded write will emit a value.

After the first emission from the above mentioned characteristic write the upstream will be disposed (because of .take(1) operator).

If the postNegotiateSecurity() is actually started it will also finish or error (given that both postLength and postPGK will emit at least one value) since there is no additional logic here.

Conclusion

postNegotiateSecurity() will most probably complete (but not in an intended manner) as the first packet from postPGK will finish it. I would assume that the peripheral expects full data before it will notify anything therefore it is waiting for the PGK to be fully transmitted which will not happen as shown above.

Logs from the application with RxBleLog.setLogLevel(RxBleLog.VERBOSE) set on could help with understanding of what actually happened.