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

Start by reviewing the existing Single.using() behavior and comparing it with Reactor's Mono.usingWhen() semantics. Define the Observable.usingWhen() behavior for cleanup on success, error, termination, and disposal, then add coverage for those cases. Done means native support for Publisher-created resources and cleanup.