En utilisant RxAndroidBle, comment souscrire à des réponses en écrivant à une caractéristique?

Question

Le périphérique BLE auquel je me connecte émet des octets sur l'une de ses caractéristiques GATT en réponse à l'écriture dans la caractéristique. Les clients sont censés activer les notifications sur cette caractéristique et interpréter les octets de changement sur la caractéristique. (Le comportement que je contrôle active un service de numérisation pour les réseaux sans fil avoisinants, puis écoute la sortie du service.)

J'utilise RxAndroidBle et suis le examples. J'ai une connexion active Observable. La caractéristique que je veux observer a un UUID appelé AP_SCAN_DATA. Il est censé émettre 0xFE en réponse à la réception d'un écrit 0xFF.

Comment puis-je appeler setupNotification et configurer un observateur pour capturer byte[] s, puis écrire une valeur à la caractéristique, afin que je puisse attraper la réponse?

Mon meilleur effort à ce jour:

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

Il existe déjà un sujet dans lequel vous trouverez un aperçu ->RxAndroidBle keeping a persistant connection + Write/Notification handling

Voici comment vous pouvez obtenir le même résultat en utilisant un seul .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 */ } 
      ); 

Pour votre information - vous devez gérer les erreurs dans tous les .subscribe() sauf si vous êtes sûr à 100% que le Observable n'émet pas d'erreurs.

Answer 2

Pour les lecteurs n'utilisant pas une version Java qui prend en charge lambdas, voici mon implémentation de la réponse de @ 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); 
       } 
      });