android - RxAndroid，在 onDestroy() Activity 中停止计时器/observable.interval

Question

我在我的应用程序中使用 RxAndroid 来处理计时器任务。用 Observable.interval() 来做。一切正常，但我需要在 Activity 的 onDestroy() 中停止计时器。我真的不知道怎么做。谁能帮帮我？这是我正在使用的代码:

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();
                }
            });

我使用的依赖项是:

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

您在这里可以做的是在 CompositeDisposable (RxJava 2) 或 CompositeSubsciption (RxJava) 对象中使用您的 Activity 中的所有订阅引用。

将您的订阅添加为:

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);

在 onDestroy 上你可以:

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

您可以在 onCreate 回调中创建您的 CompositDisposable。