顺序执行 rx.Observables

Question

我正在使用 Fernando Ceja 的简洁架构开发一个 Android 应用。我的交互者或用例之一负责获取用户的提要数据。为了获取数据，首先我必须从数据库表中检索用户的团队，然后我必须从服务器端获取 Feed 列表。

这就是我从数据库层获取团队的方式：

    mTeamCache.getAllTeams().subscribe(new DefaultSubscriber<List<SimpleTeam>>() {
        @Override
        public void onNext(List<SimpleTeam> simpleTeams) {
            super.onNext(simpleTeams);
            mTeams = simpleTeams;
        }
    });

TeamCache 基本上只是另一个交互器，负责获取我在数据库中拥有的所有团队。

以下是我从服务器端获取 Feed 数据的方式：

    mFeedRepository.getFeed(0, 50).subscribe(new ServerSubscriber<List<ApiFeedResponse>>() {
        @Override
        protected void onServerSideError(Throwable errorResponse) {
            callback.onFeedFetchFailed(...);
        }

        @Override
        protected void onSuccess(List<ApiFeedResponse> responseBody) {
            //Do stuff with mTeams
            callback.onFeedFetched(...);
        }
    });

我的 GetFeedInteractor 类有一个名为 execute 的方法，我通过回调传递我稍后在 UI 中使用它来处理响应。所有这一切的问题是，目前我正在链接这样的响应：

@Override
public void execute(final Callback callback, String userSipId) {
    mTeamCache.getAllTeams().subscribe(new DefaultSubscriber<List<SimpleTeam>>() {
        @Override
        public void onNext(List<SimpleTeam> simpleTeams) {
            super.onNext(simpleTeams);
            mTeams = simpleTeams;
            getFeedFromRepository(callback);
        }
    });
}

public void getFeedFromRepository(final Callback callback) {
    mFeedRepository.getFeedRx(0, 50).subscribe(new ServerSubscriber<List<ApiFeedResponse>>() {
        @Override
        protected void onServerSideError(Throwable errorResponse) {
            callback.onFeedFetchFailed("failed");
        }

        @Override
        protected void onSuccess(List<ApiFeedResponse> responseBody) {
            //Do stuff with mTeams
            List<BaseFeedItem> responseList = new ArrayList();

            for (ApiFeedResponse apiFeedResponse : responseBody) {
                responseList.add(FeedDataMapper.transform(apiFeedResponse));
            }

            callback.onFeedFetched(responseList);
        }
    });
}

如您所见，一旦我从缓存交互器获取团队集合，我就会调用从同一个订阅者获取提要的方法。我不喜欢这个。我希望能够做一些更好的事情，比如使用 Observable.concat(getTeamsFromCache(), getFeedFromRepository());在订阅者中将调用链接到另一个 rx.Observable 并不是一件好事。我想我的问题是，如何链接两个使用不同订阅者的 rx.Observables？

更新：

ServerSubscriber 是我实现订阅改造服务的订阅者。它只是检查错误代码和一些东西。这里是：

https://gist.github.com/4gus71n/65dc94de4ca01fb221a079b68c0570b5

默认订阅者是一个空的默认订阅者。这里是：

https://gist.github.com/4gus71n/df501928fc5d24c2c6ed7740a6520330

TeamCache#getAllTeams() 返回 rx.Observable> FeedRepository#getFeed(int page, int offset) 返回 rx.Observable>

更新 2：

这是获取用户提要的交互器现在的样子：

@Override
    public void execute(final Callback callback, int offset, int pageSize) {
        User user = mGetLoggedUser.get();
        String userSipid = mUserSipid.get();

        mFeedRepository.getFeed(offset, pageSize) //Get items from the server-side
                .onErrorResumeNext(mFeedCache.getFeed(userSipid)) //If something goes wrong take it from cache
                .mergeWith(mPendingPostCache.getAllPendingPostsAsFeedItems(user)) //Merge the response with the pending posts
                .subscribe(new DefaultSubscriber<List<BaseFeedItem>>() {
                    @Override
                    public void onNext(List<BaseFeedItem> baseFeedItems) {
                        callback.onFeedFetched(baseFeedItems);
                    }

                    @Override
                    public void onError(Throwable e) {
                        if (e instanceof ServerSideException) {
                            //Handle the http error
                        } else if (e instanceof DBException) {
                            //Handle the database cache error
                        } else {
                            //Handle generic error
                        }
                    }
                });
    }

Answer 1

我认为你错过了 RxJava 和响应式方法的要点，你不应该有不同的订阅者与 OO 层次结构和回调。
您应该构造分离的Observables，它应该发出它所处理的特定数据，没有Subscriber，然后您可以根据需要链接Observable，最后，您拥有对最终结果做出反应的订阅者预计来自链接的 Observable 流。

类似这样的东西（使用 lambdas 来编写更精简的代码）：

TeamCache mTeamCache = new TeamCache();
FeedRepository mFeedRepository = new FeedRepository();

Observable.zip(teamsObservable, feedObservable, Pair::new)
    .subscribe(resultPair -> {
            //Do stuff with mTeams
            List<BaseFeedItem> responseList = new ArrayList(); 
            for (ApiFeedResponse apiFeedResponse : resultPair.second) {
                responseList.add(FeedDataMapper.transform(apiFeedResponse));
            }
        }, throwable -> {
             //handle errors
           }
     );

我使用 zip 而不是 concat，因为您似乎在这里有 2 个独立的调用，您希望等待两者完成（将它们“压缩”在一起）然后采取行动，但当然，当您已分离Observables 流，您可以根据需要将它们以不同的方式链接在一起。

至于您的 ServerSubscriber 以及所有响应验证逻辑，它也应该是 rxify，因此您可以将其组合到您的服务器 Observable 流中。

类似的东西（为了简化而发出的一些逻辑，因为我不熟悉它......）

Observable<List<SimpleTeam>> teamsObservable = mTeamCache.getAllTeams();
Observable<List<ApiFeedResponse>> feedObservable = mFeedRepository.getFeed(0, 50)
            .flatMap(apiFeedsResponse -> {
                if (apiFeedsResponse.code() != 200) {
                    if (apiFeedsResponse.code() == 304) {
                        List<ApiFeedResponse> body = apiFeedsResponse.body();
                        return Observable.just(body);
                        //onNotModified(o.body());
                    } else {
                        return Observable.error(new ServerSideErrorException(apiFeedsResponse));
                    }
                } else {
                    //onServerSideResponse(o.body());
                    return Observable.just(apiFeedsResponse.body());
                }
            });