Использование RxAndroidBle, как я могу подписаться на ответы от написания до характеристики?

Question

Устройство BLE, которое я подключаю, испускает байты по одной из характеристик GATT в ответ на запись в характеристику. Предполагается, что клиенты должны включать уведомления об этой характеристике и интерпретировать байты изменения для характеристики. (Поведение, которое я контролирую, включает службу сканирования ближайших беспроводных сетей, а затем прослушивание вывода службы).

Я использую RxAndroidBle и следую за examples. У меня есть активное соединение Observable. Характеристика, которую я хочу наблюдать, имеет UUID, называемый AP_SCAN_DATA. Предполагается испустить 0xFE в ответ на получение письменного 0xFF.

Как я могу позвонить setupNotification и установить Наблюдатель на него, чтобы поймать испущенный 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(); 
       } 
      }); 

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 */ } 
      ); 

FYI - вы должны обрабатывать ошибки в каждом .subscribe(), если вы не 100% уверен, что Observable не выделяет ошибок.

Answer 2

Для тех читателей, которые не используют версию Java, поддерживающую lambdas, вот моя реализация ответа @ 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); 
       } 
      });