How to disable a notification with rxandroidble?

I'm currently trying to use rxandroidble in order to replace the native BLE API of Android of one of our app.

How to disable a notification? I'm able to enable it with the sample code, this one:

device.establishConnection(context, false)
.flatMap(rxBleConnection -> rxBleConnection.setupNotification(characteristicUuid)) 
.doOnNext(notificationObservable -> { // OK }) 
.flatMap(notificationObservable -> notificationObservable)     
.subscribe(bytes -> { // OK });

But in my product I have a use case where I have to disable / enable the notification(s) on demand.

Plus, I tried to directly unsubscribe / reconnect instead of disable / enable the notification but the unsubscribe command is never executed apparently, my hypothesis is because I have a high throughput (my device notifies at 300 - 400Hz), is it plausible?

(I know that BLE is not the most appropriate technology for high throughput but it's for R&D purpose here :) )

Thanks for your help!

Enabling notifications happens whenever the Observable from RxBleConnection.setupNotification() will be subscribed. To disable the notification one must unsubscribe from the above subscription.

There are several ways in which it can be coded. One of them is:

    final RxBleDevice rxBleDevice = // your RxBleDevice
    final Observable<RxBleConnection> sharedConnectionObservable = rxBleDevice.establishConnection(this, false).share();

    final Observable<Boolean> firstNotificationStateObservable = // an observable that will emit true when notification should be enabled and false when disabled
    final UUID firstNotificationUuid = // first of the needed UUIDs to enable / disable
    final Subscription subscription = firstNotificationStateObservable
            .distinctUntilChanged() // to be sure that we won't get more than one enable commands
            .filter(enabled -> enabled) // whenever it will emit true
            .flatMap(enabled -> sharedConnectionObservable // we take the shared connection
                    .flatMap(rxBleConnection -> rxBleConnection.setupNotification(firstNotificationUuid)) // enable the notification
                    .flatMap(notificationObservable -> notificationObservable) // and take the bytes
                    .takeUntil(firstNotificationStateObservable.filter(enabled1 -> !enabled1)) // and we are subscribing to this Observable until we want to disable - note that only the observable from sharedConnectionObservable will be unsubscribed
            )
            .subscribe(
                    notificationBytes -> { /* handle the bytes */ },
                    throwable -> { /* handle exception */ }
            );

Note that in the above example the connection will be closed whenever the last subscription to sharedConnectionObservable will end.

To enable / disable different characteristics you can copy and paste the above code with different Observable<Boolean> as enable / disable inputs and different UUID's.