1

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

我正在使用 RxAndroidBle 并遵循示例。我有一个活动连接 Observable。我要观察的特征有一个名为AP_SCAN_DATA. 它应该0xFE响应接收到书面的0xFF.

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

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

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();
                }
            });
4

2 回答 2

3

已经有一个主题，您可以在其中找到一些见解 -> RxAndroidBle 保持持久连接 + 写入/通知处理

这就是您如何在仅使用单个.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不会发出错误。

于 2016-12-14T10:40:14.763 回答
2

对于那些不使用支持 lambdas 的 Java 版本的读者，这是我对@s_noopy 答案的实现。

connectionObservable
    .flatMap(new Func1<RxBleConnection, Observable<Observable<byte[]>>>() {
            @Override
            public Observable<Observable<byte[]>> call(RxBleConnection connection) {
                return connection.setupNotification(AP_SCAN_DATA);
            }             
        }, new Func2<RxBleConnection, Observable<byte[]>, Observable<byte[]>>() {
            @Override
            public Observable<byte[]> call(RxBleConnection connection, Observable<byte[]> apScanDataNotificationObservable) {
                return Observable.combineLatest(
                    connection.writeCharacteristic(AP_SCAN_DATA, CharacteristicValue.RESET.asBytes()),
                    apScanDataNotificationObservable.first(),
                    new Func2<byte[], byte[], byte[]>() {
                        @Override
                        public byte[] call(byte[] writtenBytes, byte[] responseBytes) {
                                    return responseBytes;
                                }
                            }
                        );
                    }
                }
            ).flatMap(new Func1<Observable<byte[]>, Observable<byte[]>>() {
                @Override
                public Observable<byte[]> call(Observable<byte[]> observable) {
                    return observable;
                }
            })
            .first()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<byte[]>() {
                @Override
                public void call(byte[] bytes) {
                    Log.i(TAG, "notification response...." + HexString.bytesToHex(bytes));
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    logError(throwable);
                }
            });
于 2016-12-14T20:03:27.683 回答