RxJava - 完了する前に他のObservable/Completableをトリガーした後に完了

Question

RxJavaを完全に新しくしているわけではありませんが、単純なタスクのように思えます。

反応性のあるAPIを公開するデータソースがあります。データを取得して返すだけで、何も送信しない場合は自動的に接続を閉じます。非同期的に異なるスケジューラで実行されている

public Observable<Object> execute(String query) { 

    Single<RxConnection> rxConnection = getRxDB().getConnection(); 

    return rxConnection.flatMapObservable(conn -> { 
     Observable<Object> rxResult = conn.query(query); 

     return rxResult.doOnCompleted(() -> { 
      conn.close(); // THIS DOES NOT WORK. I would like to close the connection and to wait without blocking. 
     }); 

    }); 

} 

conn.query（）とはconn.close（）：

は、ここに私のコードです。 conn.close（）がサブスクライバを持たないCompletableを返すため、このコードは機能しません。さらに、doOnCompletedメソッド自体を手動で購読すると、rxResult Observableは接続が閉じられるのを待たずに完了します。

私は「を実行（文字列クエリ）」メソッドは、その観測を返すことたい： - ）（conn.queryによってフェッチすべての項目を放ち を呼び出す - 放出する項目がない場合は、それがトリガーconn.close（） - conn.close（）Completableの後でのみ完了します。

ありがとうございました。

Answer 1

観察可能なものはサブスクリプションを取得し、セットアップ中は取得しません。

編集：未完了のサブスクリプションが発生したときに接続を閉じるdoOnUnsubscribe/doOnTerminateも必要です。

Answer 2

リソースの閉鎖はObservable.usingで管理されています

Observable<T> obs = 
    Observable.using(
     resourceFactory, 
     observableFactory, 
     disposeAction) 

この作成方法は、resourceFactoryによって作成されたリソースが停止（終了またはエラー）の上に配置されたり解除されることを保証します。これは、接続、クエリおよびリリースが場合にのみ起こることを保証します

private Observable<Object> executeDeferred(String query) { 
    // your original execute() method here, including doOnComplete 
} 


public Observable<Object> execute(String query) { 
    return Observable.defer(() -> executeDeferred(query)); 
} 

：、あなたが観察が購読ときのみ、クエリを実行したい

Observable<Object> rxResult = conn.query(query) 
.concatWith(conn.close().toObservable()) 
.onErrorResumeNext(e -> 
    conn.close().toObservable().concatWith(Observable.error(e))); 

Answer 3

ので：いかが