첫 번째 방사에 의한 특성 표시 방출 횟수 결정

Question

현재 Bluetooth 장치에 대한 프로토콜을 구현 중이며 RxAndroidBle 라이브러리 (버전 1.4.3)를 사용 중입니다.

특성에 쓰고 특성 알림을 통해 응답을 듣는 방식으로 장치에서 데이터를 요청해야합니다.

내가의 코드를 사용하고 (쓰기, 듣기) 2 개 작업을 결합하려면 :이 방법은 나를 위해 작동 https://stackoverflow.com/a/41140523/734385

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) 

, 유일한 문제는 (코드 첫 번째 20 바이트 나에게 준다이다 apScanDataNotificationObservable.first()으로 인해).

불행히도, 나는 받고있는 패키지의 크기를 알지 못합니다. 나는 처음 20 바이트의 머리말에서만 정보를 추출 할 수있다. 버퍼 기능이 모두 미리 크기를 알아야하는 것처럼 보입니다.

위의 코드를 Rx 체인의 일부로 사용하여이 작업을 정상적으로 수행 할 수있는 방법이 있습니까?

즉, Rx 체인의 첫 방출을 기준으로 방출 수를 제어 할 수 있습니까?

아니면 완전히 잘못된 접근이 있습니까?

Answer 1

원하는 것을 얻을 수 있습니다.

가장 쉬운 방법은 교환 할 수있을 것 Observable.combineLatest(...)에 :

newResponseEndWatcher()는 수신 된 값이 모두가 예상되는 경우 결정하는 로직을 포함해야

Observable.merge(
     rxBleConnection.writeCharacteristic(AP_SCAN_DATA, writeValue).ignoreElements(), // send the request but ignore the returned value 
     apScanDataNotificationObservable.takeUntil(newResponseEndWatcher()) // take the response notifications until the response end watcher says so 
); 

. 그것은 다음과 같이 수 :

private Func1<byte[], Boolean> newResponseEndWatcher() { 
    return new Func1<byte[], Boolean>() { 

     private static final int NOT_INITIALIZED = -1; 

     private int totalLength = NOT_INITIALIZED; 
     private int receivedLength = NOT_INITIALIZED; 

     @Override 
     public Boolean call(byte[] bytes) { 
      if (isNotInitialized(totalLength)) { // if it is the first received value 
       // parse totalLength from the header 
      } 
      // update receivedLength 
      return receivedLength >= totalLength; 
     } 

     private boolean isNotInitialized(int value) { 
      return value == NOT_INITIALIZED; 
     } 
    }; 
} 

그냥 생각하고있는 결과 newResponseEndWatcher()입니다 Func1은 상태입니다. apScanDataNotificationObservable.takeUntil(newResponseEndWatcher())의 결과 인 관찰 가능을 변수에 저장하는 경우 다음 구독은 조기에 종료 될 수 있습니다.

는이 등록 될 때마다 newResponseEndWatcher()를 호출 한 다음 apScanDataNotificationObservable.takeUntil(newResponseEndWatcher) 새로운 만들 것 하나 Observable.using() 기능을 사용할 수 있습니다이 문제를 완화하기 위해 : 빠른 응답을위한

Observable.using(
     () -> newResponseEndWatcher(), // create a new response end watcher on each subscription 
     responseEndWatcher -> apScanDataNotificationObservable.takeUntil(responseEndWatcher), // create the response observable that will complete properly 
     responseEndWatcher -> { /* ignored, responseEndWatcher will get GCed eventually */ } 
);