android - how to call multiple setupIndication using RxAndroidBle

Question

I am trying to read the value after firing the rxBleConnection.setupIndication(UUID) I have many characteristics UUIDs and I want to line up using RxJava in a way that I can get consolidated values just the way we get using Single.Zip or Observable.zip

For example, using RxAndroidBle we can read multiple characteristics, would it be possible to do the same for setupIndication. As setupIndication is returning Observable<Observable<byte[]>> it is not possible to zip i guess.

here is the library that I am using

What I expect

Disposable disposable = device.establishConnection(false)
                .flatMap(rxBleConnection -> Observable.zip(
                        rxBleConnection.setupIndication(UUID1),
                        rxBleConnection.setupIndication(UUID2),
                        rxBleConnection.setupIndication(UUID3),
                        rxBleConnection.setupIndication(UUID4),
                        BLEReading::new
                ))
                .subscribe(
                        model -> {
                            // Process your model.

                            Log.e(TAG , "FINAL DATA ");
                        },
                        throwable -> {
                            // Handle an error here.
                        }
                ); 

Currently, I have to perform the setupIndication for all 5 charactericts.

connectDisposible = device.establishConnection(false)
                .flatMap(rxBleConnection->rxBleConnection.setupIndication(UUID1))
                .flatMap(notificationObservable -> notificationObservable)
                .subscribe(
                        bytes -> {
                            Log.e(TAG,"Notification bytes"+Arrays.toString(BLEUtils.toHex(bytes)));

                        },
                        throwable -> {
                            Log.e(TAG,"Notification Error "+throwable.getMessage());

                        }
                );

Edit

connectDisposible = device.establishConnection(false)
                .flatMap(rxBleConnection -> Observable.zip(
                        rxBleConnection.setupIndication(UUID1).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID2).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID3).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID4).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID5).flatMap(it -> it),

                        BLEReading::new
                ))
                .subscribe(
                        model -> {
                            //control never reaches here

                            Log.e(TAG , "FINAL DATA "+model);
                        },
                        throwable -> {
                            // Handle an error here.
                            Log.e(TAG , "error"+throwable.getMessage());

                        }
                );

however, in logcat i can the indications are successfully set.

 setCharacteristicNotification() - uuid: 705f68f7-83c9-6562-b2c5 enable: true
 setCharacteristicNotification() - uuid: 314fae3a-d0cf-51c4-4a67 enable: true
 setCharacteristicNotification() - uuid: 8599c5ba-f827-2d16-ce14 enable: true
 setCharacteristicNotification() - uuid: 6fbba050-e87b-6ea8-6e5d enable: true

Answer 1

The simplest way to have all indications lined up is to flatMap the indication observables into individual indications. Just have in mind that each emission to the subscribe block will happen when all of the indication observables will emit.

Using NotificationSetupMode.QUICK_SETUP to not miss emissions that happen in response to setting Client Characteristic Configuration (CCC) Descriptor.

Disposable disposable = device.establishConnection(false)
                .flatMap(rxBleConnection -> Observable.zip(
                        rxBleConnection.setupIndication(UUID1, NotificationSetupMode.QUICK_SETUP).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID2, NotificationSetupMode.QUICK_SETUP).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID3, NotificationSetupMode.QUICK_SETUP).flatMap(it -> it),
                        rxBleConnection.setupIndication(UUID4, NotificationSetupMode.QUICK_SETUP).flatMap(it -> it),
                        BLEReading::new
                ))
                .subscribe(
                        model -> {
                            // Process your model.

                            Log.e(TAG , "FINAL DATA ");
                        },
                        throwable -> {
                            // Handle an error here.
                        }
                );