Feature Request: Add `Observable.usingWhen()` for resources generated by a `Publisher`

Version: 3.1.6

In Reactor, there is Mono.usingWhen(), which works similarly to Mono.using(), but supports resources that are generated and cleaned up with a Publisher.

For example, here's a use case in r2dbc-pool:

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

In RxJava, we have Single.using(), which works the same way as Mono.using(), so there's no support for reactive resource generation/cleanup. I was thinking of doing something like this:

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

However, this does not handle cases like connection cleanup on error, dispose or terminate. The .doOnXyz() methods would not suffice, so I'm guessing this would require a custom Observable implementation, unless I'm missing something obvious. I've also checked RxJavaExtensions for anything similar to what I'm trying to achieve, but I was not able to find anything, nor could I find anything relevant on Stack Overflow.

I currently have the option of just using Mono.usingWhen() and then convert it to a Single, but it would be nice to have this natively available in RxJava. Would you be willing to add support for usingWhen()?

Beginnen Sie damit, das bestehende Verhalten von Single.using() zu überprüfen und es mit der Semantik von Reactor's Mono.usingWhen() zu vergleichen. Definieren Sie das Verhalten von Observable.usingWhen() für die Bereinigung bei Erfolg, Fehler, Beendigung und Disposition und fügen Sie Abdeckung für diese Fälle hinzu. Als abgeschlossen gilt die native Unterstützung für von Publisher erstellte Ressourcen und deren Bereinigung.