How to concatenate two observable operations in a linear fashion (do first this thing and after that one finishes do the second thing)?

Question

Polidea has released a new handy library called RxAndroidBle which is very useful for dealing with a lot of issues when you're using vanilla Bluetooth APIs.

Before explaining more the idea is to have a POJO model that has all the recent values that the device send (or I query) to me (in this case is represented by a Map object):

If I want to be notified about multiple characteristic notifications I can do this:

final UUID serviceUuid = // your service UUID
final Map<UUID, byte[]> genericModel = new HashMap<>();
final Observable<RxBleConnection> connectionObservable = // your connectionObservable

connectionObservable
        .flatMap(connection -> 
                connection.discoverServices()
                        .flatMap(services -> services.getService(serviceUuid).map(BluetoothGattService::getCharacteristics)) // get characteristics you're interested in
                        .flatMap(Observable::from) // deal with every characteristic separately
                        .flatMap(characteristic -> connection
                                        .setupNotification(characteristic) // setup notification for each
                                        .flatMap(observable -> observable), // to get the raw bytes from notification
                                Pair::new) // merge characteristic with byte[] to keep track from which characteristic the bytes came
        )
        .subscribe(
                pair -> genericModel.put(pair.first.getUuid(), pair.second),
                throwable -> { /* handle errors */}
        );

And now when I'm connected to the device the notifications are updating the POJO object (the Map in this particular example).

If I want to read values I can do the following:

 connectionObservable.flatMap(connection ->
    connection.discoverServices()
        .flatMap(services -> services.getService(SERVICE_ID).map(BluetoothGattService::getCharacteristics))
        .flatMap(Observable::from)
        .filter(characteristic -> BleUtils.hasReadProperty(characteristic.getProperties()))
        .flatMap(connection::readCharacteristic, Pair::new)
)
    .subscribe(
            pair -> genericModel.put(pair.first.getUuid(), pair.second),
            throwable -> { /* handle errors */}
    );

My main question is:

I would like to upon the first connection: read an amount of characteristics that has the reading property and only after that subscribe to those notifications that have the notification property. This is concatenate the operations. How I can achieve this?

The thing is when I do the first part of the reading, the observable read those characteristics properties but won't emit a doOnComplete method as is waiting for more so I can't kickstart or compose the next operation which is subscribing and listening to changes.

I know for sure the amount of characteristic that has the read property but I would like to do it in a generic fashion (i.e it doesn't matter if I have 7 or 15 characteristics that I want read, I only want to read them all, write the pojo values, and after that start listening the notifications).

Maybe the option is to compose an observable that counts the successful reading and after that it starts listening notifications.

How is the best reactive approach to achieve that?

To put you in situation this is the original thread that spawned this question.

Answer 1

To achieve what you want you can take several ways. One of them is:

    final UUID serviceUuid = // your service UUID
    final Map<UUID, byte[]> genericModel = new HashMap<>();
    final Observable<RxBleConnection> connectionObservable = // your connectionObservable

    connectionObservable
            .flatMap( // get the characteristics from the service you're interested in
                    connection -> connection
                            .discoverServices()
                            .flatMap(services -> services
                                    .getService(serviceUuid)
                                    .map(BluetoothGattService::getCharacteristics)
                            ),
                    Pair::new
            )
            .flatMap(connectionAndCharacteristics -> {
                final RxBleConnection connection = connectionAndCharacteristics.first;
                final List<BluetoothGattCharacteristic> characteristics = connectionAndCharacteristics.second;
                return readInitialValues(connection, characteristics)
                        .concatWith(setupNotifications(connection, characteristics));
            })
            .subscribe(
                    pair -> genericModel.put(pair.first.getUuid(), pair.second),
                    throwable -> { /* handle errors */}
            );

Where:

private Observable<Pair<BluetoothGattCharacteristic, byte[]>> readInitialValues(RxBleConnection connection,
                                                                                List<BluetoothGattCharacteristic> characteristics) {
    return Observable.from(characteristics) // deal with every characteristic separately
            .filter(characteristic -> (characteristic.getProperties() & BluetoothGattCharacteristic.PROPERTY_READ) != 0) // filter characteristics that have read property
            .flatMap(connection::readCharacteristic, // read characteristic
                    Pair::new); // merge characteristic with byte[] to keep track from which characteristic the bytes came
}

private Observable<Pair<BluetoothGattCharacteristic, byte[]>> setupNotifications(RxBleConnection connection,
                                                                                List<BluetoothGattCharacteristic> characteristics) {
    return Observable.from(characteristics) // deal with every characteristic separately
            .filter(characteristic -> (characteristic.getProperties() & BluetoothGattCharacteristic.PROPERTY_NOTIFY) != 0) // filter characteristics that have notify property
            .flatMap(characteristic -> connection
                            .setupNotification(characteristic) // setup notification for each
                            .flatMap(observable -> observable), // to get the raw bytes from notification
                    Pair::new); // merge characteristic with byte[] to keep track from which characteristic the bytes came
}

Alternatively to concatWith() you could inverse the order and use startWith().

Best Regards