RxAndroid, Stop timer/observable.interval in activity onDestroy()

Question

I am using RxAndroid for a timer task stuff in my application. Doing it with Observable.interval(). Everything works perfectly but I need to stop the timer in onDestroy() of the activity. I am really not getting any idea to do it. Can anyone please help me out? This is my code I am using:

Observable.interval(0, 10, TimeUnit.SECONDS)
            .flatMap(n -> mObservableNewChat)
            .doOnError(i -> showNoInternetConnection())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<NewChatModel>() {
                @Override
                public void onCompleted() {
                    Log.e("ActivityChat: ", "onCompleted");
                }

                @Override
                public void onError(Throwable e) {
                    Log.e("ActivityChat: ", "onError- " + e.getMessage());
                }

                @Override
                public void onNext(NewChatModel model) {
                    Log.e("ActivityChat: ", "onNext");
                    apiCallingNewChat();
                }
            });

And dependencies I am using are:

compile 'io.reactivex:rxjava:1.1.6'
compile 'io.reactivex:rxandroid:1.2.1'
// GSON
compile 'com.google.code.gson:gson:2.4'
// Serializing JSON
compile 'com.squareup.retrofit2:adapter-rxjava:2.0.2'
compile 'com.squareup.retrofit2:converter-gson:2.0.0'

Answer 1

What you can do here is have all the subscriptions references used in your Activity in a CompositeDisposable (RxJava 2) or CompositeSubsciption (RxJava) object.

Add your subscriptions to it as:

Subscription subscription = Observable.interval(0, 10, TimeUnit.SECONDS)
            .flatMap(n -> mObservableNewChat)
            .doOnError(i -> showNoInternetConnection())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<NewChatModel>() {
                @Override
                public void onCompleted() {
                    Log.e("ActivityChat: ", "onCompleted");
                }

                @Override
                public void onError(Throwable e) {
                    Log.e("ActivityChat: ", "onError- " + e.getMessage());
                }

                @Override
                public void onNext(NewChatModel model) {
                    Log.e("ActivityChat: ", "onNext");
                    apiCallingNewChat();
                }
            });

disposables.add(subscription);

And on onDestroy you can:

@Override
protected void onDestroy() {
    disposables.dispose();
    super.onDestroy();
}

You can create your CompositDisposable in your onCreate callback.

Answer 2

Use RxJava2

inside your gradle put this library

/*RxJava2 lib*/
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
    compile "io.reactivex.rxjava2:rxjava:2.1.8"
    compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'

Now in your activity/fragment

CompositeDisposable cd;

Observable.interval(0, 10, TimeUnit.SECONDS).subscribe(new Observer<Long>() 
       {

        @Override
        public void onSubscribe(Disposable d) {
           /*add disposable object here*/
            cd.add(d);
        }

        @Override
        public void onNext(Long aLong) {

        //your logic here 
        }

        @Override
        public void onError(Throwable e) {
        }

        @Override
        public void onComplete() {
        }
    });

And then call dispose() of composite disposable

 @Override
   protected void onDestroy() {
    cd.dispose();
    super.onDestroy();
    }

Answer 3

You just need to call Dispose of your Observable in your onDestroy method

Ps: it's preferable to call the dispose in OnStop method

 Disposable disposable;

disposable = Observable.interval(0, 10, TimeUnit.SECONDS)...subscribe(...);

override fun onStop() {
        super.onStop()
        disposable.dispose()
}

Answer 4

I have figured it out someway but don't know if it is the correct way to do it. I am doing it using RxJava only. Here is the code:

private Subscriber<NewChatModel> mIntervalApiCallSubscriber; 
mIntervalApiCallSubscriber = new Subscriber<NewChatModel>() {
        @Override
        public void onCompleted() {
            Log.e("ActivityChat: ", "onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            Log.e("ActivityChat: ", "onError- " + e.getMessage());
        }

        @Override
        public void onNext(NewChatModel newChatModel) {
            Log.e("ActivityChat: Interval ", "onNext");
            apiCallingNewChat();
        }
    };

    mSubscriptionNewChat = Observable.interval(0, 5, TimeUnit.SECONDS)
            .flatMap(n -> mObservableNewChat)
            .doOnError(i -> showNoInternetConnection())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(mIntervalApiCallSubscriber);

And in Activity onDestroy(),

@Override
protected void onDestroy() {
    super.onDestroy();
    if (this.mIntervalApiCallSubscriber != null)
        this.mIntervalApiCallSubscriber.unsubscribe();
}