android - RxAndroid 可观察循环

Question

所以我正在做一个由其他人开发的旧项目，碰巧项目中使用了 RxAndroid。
我不得不在一些应用程序的请求中添加缓存处理，而 OkHttp 缓存并没有像我需要的那样做得很好。所以我想出了实现我自己的服务的解决方案，该服务检查互联网连接然后发出请求，或者根据连接从缓存中读取它。我没有使用 RxAndroid 的经验，但我读了一些书，然后想到了这个:

@Override
public Observable<List<SocialNews>> getStream(@Path("stream") String stream, @Query("networks") String networks, @Query("page") Integer page, @Query("limit") Integer limit) {
    return Observable.create(new AsyncOnSubscribe<Object, List<SocialNews>>() {
        @Override
        protected Object generateState() {
            return null;
        }

        @Override
        protected Object next(Object state, long requested, Observer<Observable<? extends List<SocialNews>>> observer) {
            if (NetworkUtils.hasActiveInternetConnection(context)) {
                observer.onNext(APIClient.getService(context).getStream(stream, networks, page, limit));
            } else {
                observer.onNext(Observable.just(NetworkUtils.getCachedStream()));
            }
            return state;
        }
    });
}

我这样订阅:

 final Observable<List<SocialNews>> observable = APIClient.getCacheService(getActivity()).getStream(stream, networks, page, SportFiveAPI.DEFAULT_ITEM_LIMIT);
    LifecycleObservable.bindFragmentLifecycle(lifecycle(),
            AppObservable.bindFragment(this, observable)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread()))
            .subscribeOn(Schedulers.io())
            .subscribe(
                    socialNews -> {
                        // do stuff
                    },
                    error -> {
                        // do stuff
                    }
            );

但不知何故，它陷入了循环，并不断调用我的 AsyncOnSubscribe 中的 next 方法。有人知道我在这里做错了什么吗？

Answer 1

您没有在您的观察者上调用 onCompleted。