功能请求: 为由 `Publisher` 生成的资源添加 `Observable.usingWhen()`

在 Reactor 中,有 Mono.usingWhen() ,它的工作原理与 Mono.using() 类似,但支持使用 Publisher 生成和清理的资源。例如,以下是 r2dbc-pool 中的一个用例:

Mono<Object> result = Mono.usingWhen(
    connectionFactory.create(), // Publisher<Connection>
    connection -> executeQuery(connection),
    Connection::close // Publisher<Void>
);

在 RxJava 中,我们有 Single.using() ,其工作原理与 Mono.using() 相同,因此不支持反应性资源生成/清理。我曾想过采用以下方式:

Single<Object> result = Single
    .fromPublisher(connectionFactory.create())
    .flatMap(connection -> executeQuery(connection)
        .flatMap(result -> Completable
            .fromPublisher(connection.close())
            .toSingleDefault(result)));

然而,这并没有处理错误、释放或终止连接清理等情况。.doOnXyz() 方法也无法满足这些需求,因此我猜想这可能需要自定义的 Observable 实现,除非我遗漏了什么显而易见的东西。我还检查了 RxJavaExtensions 中是否有类似于我要实现的功能,但我无法找到任何相关内容,也未在 Stack Overflow 上找到相关信息。