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()?

Inizia esaminando il comportamento esistente di Single.using() e confrontandolo con la semantica di Mono.usingWhen() di Reactor. Definisci il comportamento di Observable.usingWhen() per la pulizia in caso di successo, errore, terminazione e disposizione, quindi aggiungi la copertura per questi casi. Il lavoro è completo quando sono disponibili il supporto nativo per le risorse create da Publisher e la relativa pulizia.