android - 使用 RxAndroidBle，我如何订阅对特性的写入响应？

Question

我正在连接的 BLE 设备在其 GATT 特征之一上发出字节以响应对特征的写入。客户端应该启用该特征的通知，并解释该特征的更改字节。 (我正在控制的行为是打开附近无线网络的扫描服务，然后监听服务输出。)

我正在使用 RxAndroidBle 并遵循 examples .我有一个 Activity 连接 Observable。我要观察的特征有一个名为 AP_SCAN_DATA 的 UUID。它应该发出 0xFE 以响应接收到的书面 0xFF。

如何调用 setupNotification 并在其上设置一个 Observer 以捕获发出的 byte[]，然后将值写入特征，以便我可以捕获回应？

到目前为止我的最大努力:

connectionObservable.observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<RxBleConnection>() {
                @Override
                public void onCompleted() { // ignore...
                }

                @Override
                public void onError(Throwable e) { // ignore...
                }

                @Override
                public void onNext(final RxBleConnection connection) {
                    Observable.just(connection)
                              .flatMap(new Func1<RxBleConnection, Observable<Observable<byte[]>>>() {
                                  @Override
                                  public Observable<Observable<byte[]>> call(RxBleConnection connection) {
                                      return connection.setupNotification(AP_SCAN_DATA);
                                  }
                            })
                            .doOnNext(new Action1<Observable<byte[]>>() {
                                @Override
                                public void call(Observable<byte[]> observable) {
                                    Log.i(TAG, "notification has been set up");
                                    // This code logs on DEBUG that a write was made, but no response ever arrives 
                                    connection.writeCharacteristic(AP_SCAN_DATA, CharacteristicValue.RESET.asBytes())
                                            .observeOn(AndroidSchedulers.mainThread())
                                            .subscribe();

                                }
                            })
                            .flatMap(new Func1<Observable<byte[]>, Observable<byte[]>>() {
                                @Override
                                public Observable<byte[]> call(Observable<byte[]> observable) {
                                    return observable;
                                }
                            })
                            .doOnNext(new Action1<byte[]>() {
                                @Override
                                public void call(byte[] bytes) {
                                    Log.i(TAG, "want to read response bytes here, but I don't... " + HexString.bytesToHex(bytes));
                                }
                            })
                            .subscribe();
                }
            });

Answer 1

已经有一个主题，您可以从中找到一些见解 -> RxAndroidBle keeping a persistant connection + Write/Notification handling

这就是您如何在仅使用单个 .subscribe() 的情况下获得相同的结果。

    connectionObservable
            .flatMap( // when the connection is available...
                    rxBleConnection -> rxBleConnection.setupNotification(AP_SCAN_DATA), // ... setup the notification...
                    (rxBleConnection, apScanDataNotificationObservable) -> Observable.combineLatest( // ... when the notification is setup...
                            rxBleConnection.writeCharacteristic(AP_SCAN_DATA, writeValue), // ... write the characteristic...
                            apScanDataNotificationObservable.first(), // ... and observe for the first notification on the AP_SCAN_DATA
                            (writtenBytes, responseBytes) -> responseBytes // ... when both will appear return just the response bytes...
                    )
            )
            .flatMap(observable -> observable) // ... flatMap the result as it is Observable<byte[]>...
            .first() // ... and finish after first response is received to cleanup notifications
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    responseBytes -> { /* consume the response here */ },
                    throwable -> { /* handle exception */ }
            );

仅供引用 - 您应该在每个 .subscribe() 中处理错误，除非您 100% 确定 Observable 不会发出错误。