RxJava-在完成之前触发另一个可观察/可完成，并仅在完成之后触发

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将在不等待连接关闭的情况下完成。

我希望“execute（String query）”方法返回一个可观察值：-发出conn.query（）调用获取的所有项-当没有更多项要发出时，它触发conn.close（）-仅在conn.close（）Completable之后完成；

谢谢

Answer 1

因此，您希望仅在可观察对象被订阅时执行查询：

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获得订阅时发生，而不是在安装过程中发生。

编辑：您还需要一个doOnUn订阅/doOnTerminate，以在过早退订发生时关闭连接。

Answer 2

使用可观察的<代码>管理资源关闭。使用：

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

此创建方法确保由resourceFactory创建的资源在终止（完成或错误）或取消订阅时被释放。

Answer 3

怎么样：

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