RxJava 2 onDispose not called for Observable.share()

[gkamperis]

Hi,

using RxJava 2.1.10.

according to the documentation for .share() I expect the onDispose to be called on the shared observable once all subscribers dispose.

Is my expectation wrong?

Thanks in advance.

http://reactivex.io/documentation/operators/refcount.html

Sample:

io.reactivex.Observable<Long> unshared = io.reactivex.Observable.interval(0, 100, TimeUnit.MILLISECONDS);
unshared.doOnDispose(() -> Log.d("RX TEST", "DISPOSED UNSHARED!"));
unshared.doFinally(() -> Log.d("RX TEST", "FINALLY UNSHARED!"));
unshared.doOnComplete(() -> Log.d("RX TEST", "COMPLETE UNSHARED!"));

io.reactivex.Observable<Long> shared = unshared.share();
shared.doOnDispose(() -> Log.d("RX TEST", "DISPOSED SHARED!"));
shared.doFinally(() -> Log.d("RX TEST", "FINALLY SHARED!"));
shared.doOnComplete(() -> Log.d("RX TEST", "COMPLETE SHARED!"));

Disposable sub1 = shared
		.doOnDispose(() -> Log.d("RX TEST", "doOnDispose Subscription 1"))
		.subscribe(i -> Log.d("RX TEST", "Shared Subscription #1: " + i));
Disposable sub2 = shared
		.doOnDispose(() -> Log.d("RX TEST", "doOnDispose Subscription 2"))
		.subscribe(i -> Log.d("RX TEST", "Shared Subscription #2: " + i));

_handler.postDelayed(new Runnable() {
	@Override
	public void run() {
		sub1.dispose();
		Log.d("RX TEST", "disposing Subscription 1");
		sub2.dispose();
		Log.d("RX TEST", "disposing Subscription 2");
	}
}, 500);

Log:
RX TEST: Shared Subscription #1: 0
RX TEST: Shared Subscription #2: 0
RX TEST: Shared Subscription #1: 1
RX TEST: Shared Subscription #2: 1
RX TEST: Shared Subscription #1: 2
RX TEST: Shared Subscription #2: 2
RX TEST: Shared Subscription #1: 3
RX TEST: Shared Subscription #2: 3
RX TEST: Shared Subscription #1: 4
RX TEST: Shared Subscription #2: 4
RX TEST: Shared Subscription #1: 5
RX TEST: Shared Subscription #2: 5
RX TEST: doOnDispose Subscription 1
RX TEST: disposing Subscription 1
RX TEST: doOnDispose Subscription 2
RX TEST: disposing Subscription 2

[akarnokd]

You are ignoring the return value of doOnDispose and thus you are not chaining anything onto shared or unshared. RxJava is a fluent library where you keep "dotting into" the result of one operation to apply another one. For example:

io.reactivex.Observable<Long> unshared = io.reactivex.Observable.interval(
     0, 100, TimeUnit.MILLISECONDS)
.doOnDispose(() -> Log.d("RX TEST", "DISPOSED UNSHARED!"))
.doFinally(() -> Log.d("RX TEST", "FINALLY UNSHARED!"))
.doOnComplete(() -> Log.d("RX TEST", "COMPLETE UNSHARED!"));

io.reactivex.Observable<Long> shared = unshared.share()
.doOnDispose(() -> Log.d("RX TEST", "DISPOSED SHARED!"))
.doFinally(() -> Log.d("RX TEST", "FINALLY SHARED!"))
.doOnComplete(() -> Log.d("RX TEST", "COMPLETE SHARED!"));