Do I have to unsubscribe from completed observable?

Question

If an observable completes, do I still have to unsubscribe / dispose (in RxJava2) the observable to remove the Observer (prevent memory leaks) or is this handled internally by RxJava once a onComplete or onError event occurs?

what about other types like Single, Completable, Flowable etc.

Answer 1

Yes you are correct.

After a stream is terminated ( onComplete / onError has been called ), subscriber unsubscribes automatically. You should be able to test these behaviors using isUnsubscribed() method on the Subscription object.

Answer 2

While you do not need to manually unsubscribe from a terminated stream, you can still create a memory leak using RxJava2 if you are not careful.

Consider the following code:

repository.getData()     .subscribeOn(Schedulers.io())     .observeOn(AndroidSchedulers.mainThread())     .subscribe(data -> myTextView.setText(data.toString())); 

The lambda parameter in the subscribe is "syntatic sugar" over an anonymous inner class:

subscribe(new Consumer<Data>() {     @Override     public void accept(final Data data) {         myTextView.setText(data.toString());     } }); 

On the JVM, an anonymous inner class maintains a reference to the outer class.

Assume that for the above naive code, the outer class is an Activity (this would also follow for a Fragment, Service, BroadcastReceiver or any class whose lifecycle is controlled by the Android OS).

The Activity subscribes to the Observer but then is destroyed by the Android OS in conditions of low-memory (you can mimic this effect by turning on Developer Options/Don't Keep Activities). If the work on Schedulers.io() is still running when the Activity is destroyed, a reference will still be maintained to the Activity through the anonymous inner class. This means a memory leak that prevents the Activity from being finalized by the garbage collector. If the Activity has a number of Views or, say, a Bitmap object then the memory leak can be quite substantial.

There are a number of solutions here but one of them is to maintain a CompositeDisposable object and to clear this in the onDestroy() lifecycle method of the Android Activity:

public class MyActivity extends Activity {     DataRepository dataRepository;    CompositeDisposable disposables;     @Override    public void onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        disposables = new CompositeDisposable();    }     public void onButtonClick(View v) {        repository.getData()                        .subscribeOn(Schedulers.io())           .observeOn(AndroidSchedulers.mainThread())           .doOnSubscribe(disposable -> disposables.add(disposable))           .subscribe(data -> myTextView.setText(data.toString()));    }     @Override    public void onDestroy() {        disposables.clear();        super.onDestroy();    } } 

You can refer to a good example of how to use RxJava in an Android app in the official Google Android Architecture Blueprints.