2.x/3.x: Consumers leaked longer than they have to

[jstaahl]

Consumers passed to any of the .subscribe() overloads have their captured variables leaked longer than they have to be.

Taking Single as an example, though this happens for any of the RxJava types:

val disposable = someSingle
    .doStuff()
    .subscribe { _ ->
        println(someLargeObject.status)
    }
compositeDisposable.add(disposable)

If the above stream finishes quickly and there are no longer any other references to someLargeObject, the consumer passed in to the subscribe function which captures the someLargeObject will not be released until the compositeDisposable is cleared.

A way around this is to just change the RxJava stream to not capture anything in the subscribe consumer, but instead capture it earlier and pass it along through the stream:

val disposable = someSingle
    .doStuff()
    .map { it to someLargeObject }  // capture here isntead
    .subscribe { (_, obj) ->
        println(obj.status)
    }
compositeDisposable.add(disposable)

This works because once the final observer receives onSuccess, it clears its reference to its upstream disposable, which includes the map which contains the closure that captured someLargeObject. I understand that generally, good RxJava design includes "keeping everything within the stream", in other words capturing as little as possible by only operating on objects in the stream. However, this can sometimes be difficult to achieve.

A simple fix for this is to have the terminal observer, in this case ConsumerSingleObserver, null out its references to its onSuccess and onError consumers as soon as it transitions to disposed.
This would solve a rather common memory leak vector that many RxJava developers unknowingly fall victim to.

[akarnokd]

This is a known limitation of those methods. The 1st party workaround is to use DisposableObserver and and manually remove/delete it upon termination:

DisposableObserver<T> observer = new DisposableObserver<T>() {

    @Override
    public void onComplete() {
        composite.delete(this);
    }
};
composite.add(observer);

There is a 3rd party extension subscribeAutoDispose that does this too.

[jstaahl]

Thanks. Any reason in particular why changing the implementations of ConsumerSingleObserver, etc to null out their fields is not a viable option?

[akarnokd]

Nulls are generally trouble, so are non final fields. Also any remnants in the composite have to be considered for a resolution. In general, one has to be careful with lambda capture regardless.

With Kotlin extension methods, you can also add your custom lambda-based subscribe method with your nulling behavior.