RxAndroid 可观察循环答案

【问题标题】：RxAndroid observable loopingRxAndroid 可观察循环
【发布时间】：2018-01-29 07:22:34
【问题描述】：

所以我正在开发一个由其他人开发的旧项目，并且碰巧在该项目中使用了 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 中调用下一个方法。有谁知道我在这里做错了什么？

【问题讨论】：

是的，您使用的是AsyncObservable。请改用defer。
@akarnokd 所以不用just 使用defer，然后在Func0 中我必须返回一个新的Observable 所以我会返回Observable.just(NetworkUtils.getCachedStream())？
JohnWowUs 提供的答案也正确吗？

标签： android retrofit rx-java rx-android

【解决方案1】：

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

【讨论】：

在return state;之前调用可以吗？
是的，请先返回

【解决方案2】：

使用Observable.fromCallable 比使用Observable.create 好得多，这正是为了避免您在代码中违反Observable 合约——即不调用onCompleted。我推荐这样的东西：

@Override
public Observable<List<SocialNews>> getStream(@Path("stream") String stream, @Query("networks") String networks, @Query("page") Integer page, @Query("limit") Integer limit) {
Observable.fromCallable(() -> NetworkUtils.hasActiveInternetConnection(context))
          .flatMap(hasConnection -> {
            if (hasConnection) {
                return APIClient.getService(context).getStream(stream, networks, page, limit);
            } else {
                return Observable.just(NetworkUtils.getCachedStream())
            }
          })
}

此外，使用concat 和first 进行这种回退类型的数据检索有一个非常好的模式。有关详细信息，请参阅 Dan Lew 的 blog post。

【讨论】：