複数のRxJavaオブザーバブルで副作用を管理する

Question

私は、連携して動作する2つの特徴を持つBLE GATTサービスを使用しています。 1つは書き込み専用の特性で、文字列値をクエリとして送信し、もう1つは通知に対する応答を受け取る通知のみの特性です。

通知サービスは少し遅く、通知が読み取られる前に次のクエリに移動しないことが重要です。そうでない場合、応答は失われます。

この目的のために、私はRxAndroidBle Observablesを使用していました。書き込みと通知の特性に別々のチャンネルがあります。 3番目のObservableがクエリを提供します。しかし、書き込みはあまりにも速く進んでいます。

ConnectableObservable notifyObservable = 
    createNotifyObservable(NOTIFY_UUID).publish(); 

queryObserverable 
    .doOnSubscribe(notifyObservable::connect) 
    .doOnNext(query -> Log.d(TAG, "Processing query: " + query)) 
    .flatMap(query -> createWriteObservable(WRITE_UUID, query))) 
    .doOnNext(request -> Log.d(TAG, "Write initiated.")) 
    .flatMap(request -> notifyObservable) 
    .doOnNext(response -> Log.d(TAG, "Query response: " + response)); 

だからアプリケーションを実行するには、ここで私がログに見るものだ（含まれているタイムスタンプ、プロセスIDおよびスレッドID）：

06-22 14:30:01.991 14085-15360 Processing query: Query1 
06-22 14:30:02.011 14085-15360 Processing query: Query2 
06-22 14:30:02.011 14085-15360 Processing query: Query3 
06-22 14:30:07.261 14085-15443 Write initiated. 
06-22 14:30:07.301 14085-15445 Write initiated 
06-22 14:30:07.321 14085-15447 Query response: Response3 
06-22 14:30:07.321 14085-15449 Write initiated. 
06-22 14:30:07.321 14085-15447 Query response: Response3 
06-22 14:30:07.351 14085-15453 Query response: Response3 

次のことを確実にするRxJavaと方法はあります応答が受信された後にのみ書き込みが行われますか？

EDIT： flatMapコールの両方に推奨されるmaxConnection引数を指定すると、呼び出しは適切な順序で発生しますが、Observableの最初のクエリに対してのみ発生します。あなたのflatMapにmaxConcurrent = 1引数を渡す

06-22 14:39:09.841 22245-23079 Processing query: Query1 
06-22 14:39:15.131 22245-23166 Write initiated. 
06-22 14:39:15.201 22245-23169 Query response: Response1 

Answer 1

てみコール：ここでは、この場合のログです

... 
.flatMap(query -> createWriteObservable(WRITE_UUID, query), 1) 
... 
.flatMap(request -> createNotifyObservable(NOTIFY_UUID), 1) 
... 

Answer 2

その後、あなたはあなたのクエリを詰めることができ、各クエリに対して正確に一つの応答があった場合機能：

Observable<String> query(String query) { 
    return notifyObservable.flatMap(notifyBytesObservable -> // to be sure that when we will start writing we're already listening 
    Observable.combineLatest(// with combineLatest both Observables will be subscribed at the same time 
     notifyBytesObservable.first(), // to unsubscribe from notification after the first one 
     createWriteObservable(uuid, query), 
     { notificationBytes, writtenBytes -> notificationBytes } 
    ) 
) 
    .map(notificationBytes -> String(notificationBytes)) 
} 

そして今、あなたはこのような呼び出しを行うことができる必要があります：

queryObserverable 
    .flatMap(queryString -> query(queryString), 1) 

こちらがお役に立てば幸いです。

最高のごあいさつ