Sending data over the air and after completion end the operation using RxAndroidBLE

I am trying to send the 128 bytes of the block to BLE Controller using the RxAndroidBle library. the flow to send data from mobile to BLE controller is as follows

1. Connect with BLE Controller
2. Start OTA (sending 1)
3. Send CRC (of the data block)
4. Send data block
5. wait for 2 seconds
6. repeat step 3
7. END OTA (sending 2)

Here is snapshot of a code

.flatMap(rxBleConnection -> prepareWriting())
.flatMapIterable(otaMetaData -> otaMetaData)
.zipWith(Observable.interval(2, TimeUnit.SECONDS), (item, interval) -> item)
.doOnNext(metaData -> {
    otaMetaData = metaData;
})
.map(otaMetaData -> {
    return mRxBleConnection.writeCharacteristic(OTA_CHECKSUM, otaMetaData.getCrcBlock()).toObservable();
})

.doOnNext(otaMetaData -> {
    Log.e(TAG, "Writing CRC " + Arrays.toString(BLEUtils.toHex(otaMetaData.getCrcBlock())));


})
.map(bytes -> {
    return mRxBleConnection.writeCharacteristic(OTA_DATA, otaMetaData.getDataBlock()).toObservable();
})
.doOnNext(otaMetaData -> {
    Log.e(TAG, "Writing Data " + Arrays.toString(BLEUtils.toHex(otaMetaData.getDataBlock())));

})
.flatMap(bytes -> mRxBleConnection.writeCharacteristic(OTA_CONTROL,OTA_DATA_END).toObservable())

The problem is while sending the END OTA because as the flatMapIterable returns 20 items, .flatMap(bytes -> mRxBleConnection.writeCharacteristic(OTA_CONTROL,OTA_DATA_END) is getting called 20 times.

So, I am not sure how I can send the OTA_DATA_END command when all the 20 items get processed. Moreover, any suggestion to improve the existing code is welcome.

You can use flatMapIterable() with toList(). Try to add toList() operator before OTA_DATA_END command like:

.toList() // wait for all commands to complete
.flatMap(bytes -> mRxBleConnection.writeCharacteristic(OTA_CONTROL,OTA_DATA_END).toObservable())

EDIT

Better to separate steps like

.flatMap(rxBleConnection -> prepareWriting())
.flatMap(otaMetaData -> Observable.fromIterable(otaMetaData)
        .zipWith(Observable.interval(2, TimeUnit.SECONDS), (metaData, interval) -> metaData)
        .flatMap(metaData -> {
            return mRxBleConnection.writeCharacteristic(OTA_CHECKSUM, metaData.getCrcBlock())
                    .toObservable();
        }, (metaData, bytes) -> metaData) /* result selector */
        .flatMap(metaData -> {
            return mRxBleConnection.writeCharacteristic(OTA_DATA, metaData.getDataBlock())
                    .toObservable();
        }, (metaData, bytes) -> metaData)
        .toList()
        .toObservable()
)
.flatMap(otaMetaData -> {
    return mRxBleConnection.writeCharacteristic(OTA_CONTROL, OTA_DATA_END)
            .toObservable();
})