如何在 RxAndroidBLE 中启用/禁用通知/指示

Question

我正在创建一个RxJava2 链，我想在其中启用和禁用通知。我设置的流程如下。

1. 建立连接。
2. 将通知设置为READ_STATUS UUID。
3. 如果返回的字节是zero，则执行写字节01到WRITE_STATUS UUID，在WRITE_STATUS之后，启用READ_STATUS UUID的通知以验证它有字节值1。
4. 否则，如果返回的字节为1，则只需启用其他指标（UUID1、UUID2、UUD3）并读取该值。

我在第 2 步和第 3 步遇到问题，我通过启用通知来读取 READ_STATUS UUID 的值。为了重新读取该值，我可能需要禁用通知，然后再次启用它。要禁用通知，我必须处理那个特定的 setupNotification 。

代码如下

  connectDisposable=

                    device.establishConnection(false)

                   .flatMap(rxBleConnection -> {

                       rxBleConnection.discoverServices();
                       mRxBleConnection = rxBleConnection;
                       return Observable.just(rxBleConnection);
                   })
                   .flatMap(rxBleConnection ->mRxBleConnection.setupNotification(READ_STATUS,NotificationSetupMode.QUICK_SETUP).flatMap(it->it))
                   .takeUntil(bytes -> { 
                        
                        if(getByteValue(bytes)==0)
                            return false;// dispose above to disable the notification
                        else
                            return true; // no need to disable the notification and continue writing
                    })
                            

                     .flatMap(bytes -> {

                        return Observable.zip(

                                mRxBleConnection.writeCharacteristic(WRITE_STATUS, new byte[]{1}).toObservable(),
                                // setupNotification again to check whether read status has 1 or not 
                                mRxBleConnection.setupNotification(READ_STATUS, NotificationSetupMode.QUICK_SETUP).flatMap(it->it),
                                Pair::new

                        );
                    })
                     .flatMap(bytes ->{

                        byte [] val= bytes.first;
                        if(getByteValue(val) == 1){

                            return Observable.zip(
                                    mRxBleConnection.setupIndication(HISTORY, NotificationSetupMode.QUICK_SETUP).doOnNext(observable -> Log.e(TAG,"Here 1 ")).flatMap(it -> it),
                                    mRxBleConnection.setupIndication(PARAMCHECK, NotificationSetupMode.QUICK_SETUP).doOnNext(observable -> Log.e(TAG,"Here 2 ")).flatMap(it -> it),
                                    mRxBleConnection.setupIndication(FAULTINFO, NotificationSetupMode.QUICK_SETUP).doOnNext(observable -> Log.e(TAG,"Here 3 ")).flatMap(it -> it),
                                  
                                    Data::Readings);
                        }
                       
                        return Observable.empty();
                    }).subscribe(data -> {
                    
                    });

这段代码的问题是我的takeUntil最后触发了它没有处理之前的setupNotificaion操作，以便我以后可以重新阅读。

我尝试了this 线程中提到的解决方案，但不幸的是我没有分享RxBleConnection

Answer 1

这段代码的问题是我的 takeUntil 最后触发了它没有处理之前的 setupNotificaion 操作，以便我以后可以重新阅读它。

问题是你的条件是倒置的。来自.takeUntil()Javadoc：

 * @return an Observable that first emits items emitted by the source Observable, checks the specified
 *         condition after each item, and then completes when the condition is satisfied.

您使用过：

                   .takeUntil(bytes -> { 
                        
                        if(getByteValue(bytes)==0)
                            return false;// dispose above to disable the notification
                        else
                            return true; // no need to disable the notification and continue writing
                    })

上游应该被处理时应该满足的地方（返回 true）：

                   .takeUntil(bytes -> { 
                        
                        if(getByteValue(bytes)==0)
                            return true;// dispose above to disable the notification
                        else
                            return false; // no need to disable the notification and continue writing
                    })

Answer 2

要取消订阅或处置setupNotification 或setupIndication，可以使用以下代码。我相信可能有不同的方法，但到目前为止我可以找到这个

private Observable<Pair<byte[],byte[]>> getValueFromIndication(RxBleConnection rxBleConnection){

         final PublishSubject<Boolean> unsubscribeOperation= PublishSubject.create();


        return Observable.zip(

                rxBleConnection.setupIndication(TSDictionary.FAULT_RETRY_COUNT_SEQUENCE,NotificationSetupMode.QUICK_SETUP).flatMap(it->it).takeUntil(unsubscribeOperation),
                rxBleConnection.setupIndication(TSDictionary.FAULT_RETRY_INFORMATION,NotificationSetupMode.QUICK_SETUP).flatMap(it->it).takeUntil(unsubscribeOperation),

                (bytes, bytes2) -> {

                    unsubscribeOperation.onNext(true);

                    return Pair.create(bytes,bytes2);
                }
        );
}

在上面的代码中，我正在压缩两个指示操作，一旦我从中获得值，我将使用 PublishSubject 和 takeUntil 从更改链中取消订阅。