Rxjava2/RxAndroid2, dispose (looking for an equivalent for rx1 unsubsribe)

Question

This is my first time developing in reactive paradigm world, and i started using rxjava2/rxandroid2, as based on videos I've watched and articles I've read, it seems like its better to start with 2 as 1 has so many changes that differs the library in a big scale, but now I'm having some trouble looking for something that acts like the

 unsubscribe()

method of the former rxjava/rxandroid library

my goal is just quite simple

perform an API call(network operation)
listen and react on what the observable will emit (happy path)
do not listen or react when app goes to PAUSE state
or, unsubscribe on observable as soon as android goes to the pause life-cycle

, based on the resources around there is

 dispose()

method of rx2, what I understand with this is that it disposes any current resources(in my case, base on what i understand, invoking this will make the observable detach itself to any observer).

but that doesn't seem to be what I'm expecting, please have a look at the ff codes:

public class MainActivity extends AppCompatActivity {

    final Disposable disposable = new Disposable() {

        @Override
        public void dispose() {
            Log.e("Disposed", "_ dispose called.");
        }

        @Override
        public boolean isDisposed() {
            return true;
        }
    };

    @Override
    protected void onCreate(Bundle savedInstanceState) {

        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Observer<Object> observer = new Observer<Object>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.e("OnSubscribe", "On Subscribed Called");
            }

            @Override
            public void onNext(Object value) {
                Log.e("onNext", "Actual Value (On Next Called).");
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onComplete() {
                Log.e("OnComplete", "On Complete Called.");
            }
        };

        EventsApiService.getInstance().testApi().testCall()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnDispose(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.e("Disposed?", "__ Dispose");
                    }
                })
                .subscribe(observer);

        observer.onSubscribe(disposable);
    }

    @Override
    public void onPause() {

        super.onPause();
        disposable.dispose();
    }
}

I'm having this output:

03-23 09:08:05.979 3938-3938/edu.rx.study E/Disposed: _ dispose called.
03-23 09:08:13.544 3938-3938/edu.rx.study E/onNext: Actual Value (On Next Called).
03-23 09:08:13.544 3938-3938/edu.rx.study E/OnComplete: On Complete Called.

I was expecting that onNext won't be called anymore or maybe both onNext and onComplete, but that doesn't seem to be working, am i missing something here? or theres something i totally don't understand, my thinking with my code is,

"what if onNext is performing something towards a widget(UI)(Observer) and the app goes on pause state?", I don't want that UI(Observer) to react on that particular UI anymore.

Many people are right, and I admit, switching to reactive programming is quite hard especially rxjava2/rxandroid2 has a very steep learning curve.

Any help will be greatly appreciated.

yosriz yosriz · Accepted Answer · 2017-03-23T11:52:28

You're handling incorrectly the Observer and the Disposable, the Disposable object should be handing to you by the Observable, you can't just create it by yourself, and call explicitly Observer.onSubscribe() with it, as it's not connected to the Observable and does not terminates it. (you can also notice that Observer.onSubscribe is called twice, one by the Observable and one by you)

What you should do, is simply use the onSubscribe(Disposable d) method at your Observer to save the Disposable, which will be called automatically by the Observable and will hand you the correct Disposable object, that you can successfully terminate the network operation with it.

Another option is, to not use at all the subscribe(Observer o) method, but other overloads that takes your onNext/onError/onCompleted as parameters, and returns Disposable object, which you can dispose (unsusbcribe) with it, for terminating the network call.

Rxjava2/RxAndroid2, dispose (looking for an equivalent for rx1 unsubsribe)

1 Answers