3.x: Add subscribe with disposable container

[akarnokd]

Purpose

With the existing lambda-subscribe methods, the returned Disposable is often added to a CompositeDisposable but generally is not removed right when the source terminates but when the parent context calls clear() on the composite. This can lead to unintended retention of resources captured by the lambdas as well as the composite growing and growing with now dead consumers.

These new overloads will interact with the container by adding themselves and removing themselves during the consumer's lifecycle.

Resolves #7295

Description

Wraps the given onXXX callbacks into a Disposable Subscriber/Observer/etc.,
adds it to the given DisposableContainer (such as CompositeDisposable) and ensures, that if the upstream
terminates or this particular Disposable is disposed, the Subscriber/Observer/etc. is removed
from the given composite.

The Subscriber/Observer/etc. will be removed after the callback for the terminal event has been invoked.

Example

CompositeDisposable composite = new CompositeDisposable();

Disposable d = Flowable.just(1).subscribe(
    System.out::println, 
     Throwable::printStackTrace, 
     () -> System.out.println("Done"),
     composite
);

assertEquals(0, composite.size());

// --------------------------

Disposable d2 = Flowable.never().subscribe(
    System.out::println, 
    Throwable::printStackTrace, 
    () -> System.out.println("Done"),
    composite
);

assertEquals(1, composite.size());

d2.dispose();

assertEquals(0, composite.size());

Credits

This code was uplifted and adapted from the RxJavaExtensions project of mine: https://github.com/akarnokd/RxJavaExtensions#subscribeautodispose

[codecov]

Codecov Report

Merging #7298 (1f0d3ae) into 3.x (a9e0a8a) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##                3.x    #7298    +/-   ##
==========================================
  Coverage     99.53%   99.54%            
- Complexity     6744     6783    +39     
==========================================
  Files           747      751     +4     
  Lines         47347    47489   +142     
  Branches       6376     6378     +2     
==========================================
+ Hits          47129    47275   +146     
+ Misses           98       95     -3     
+ Partials        120      119     -1     

Impacted Files	Coverage Δ
...in/java/io/reactivex/rxjava3/core/Completable.java	100.00% <100.00%> (ø)
.../main/java/io/reactivex/rxjava3/core/Flowable.java	100.00% <100.00%> (ø)
src/main/java/io/reactivex/rxjava3/core/Maybe.java	100.00% <100.00%> (ø)
...ain/java/io/reactivex/rxjava3/core/Observable.java	100.00% <100.00%> (ø)
...rc/main/java/io/reactivex/rxjava3/core/Single.java	100.00% <100.00%> (ø)
...ernal/observers/AbstractDisposableAutoRelease.java	100.00% <100.00%> (ø)
.../observers/DisposableAutoReleaseMultiObserver.java	100.00% <100.00%> (ø)
...ernal/observers/DisposableAutoReleaseObserver.java	100.00% <100.00%> (ø)
...l/subscribers/DisposableAutoReleaseSubscriber.java	100.00% <100.00%> (ø)
...l/operators/observable/ObservableFlatMapMaybe.java	91.54% <0.00%> (-3.53%)	⬇️
... and 18 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a9e0a8a...1f0d3ae. Read the comment docs.