当 Obs B 依赖于 Obs A 时合并两个 observables

Question

我有两个存储库类，存储库 A 作为 Observable 的竞赛列表>。

我的 Competition 班级有一个 countryId。

我还有一个存储库 B，它返回一个国家列表作为 Observable> 或一个按 Id 的国家作为 Observable

我想检索我的比赛列表，并通过返回包含以下内容的 CountryCompetition 类以某种方式将其与各自的国家/地区合并：

class CountryCompetition {
    public Country country;
    public Competition competition;
}

这意味着将我的competitionService.getCompetitions() 与countryService.getCountry(competition.id) 结合起来，但我不确定如何实现这一点； merge 或 zip 采取 observable，我还没有每个国家的 ID。

 mCompetitionService.getCompetitions(wrapSearch(constraint))
                .flatMap(new Func1<List<Competition>, Observable<Competition>>() {
                    @Override
                    public Observable<Competition> call(List<Competition> competitions) {
                        return Observable.from(competitions);
                    }
                })

                .map(new Func1<Competition, CountryCompetition>() {
                    @Override
                    public CountryCompetition call(Competition competition) {
                        CountryCompetition c = new CountryCompetition();
                        c.setCompetition(competition);
//Here i would like to set the Country as well, but mCountryService.getCountryById(competition.getCountryId()) returns another observable.
                        return c;
                    }
                })
                .toList()
                .subscribeOn(mSchedulerProvider.io())
                .observeOn(mSchedulerProvider.ui())
                .subscribe(subscriber);

Answer 1

如果我正确理解您的问题，您有两个列表，并且您希望基本上合并它们。您可能会发现我的解决方案很有用：

模型类（为简单起见，不包括构造函数）：

class Country {
    String name;
    String id;
}

class Competition {
    String name;
    String countryId;
}

class CountryCompetition {
    public Country country;
    public Competition competition;
}

遵循源 Observable 定义的虚拟数据：

public Observable<List<Competition>> getCompetitions() {
    ArrayList<Competition> competitions = new ArrayList<>();
    competitions.add(new Competition("First", "id_0"));
    competitions.add(new Competition("Second", "id_1"));
    competitions.add(new Competition("Third", "id_1"));
    competitions.add(new Competition("Fourth", "id_2"));
    competitions.add(new Competition("Fifth", "id_3"));

    return Observable.just(competitions);
}

public Observable<List<Country>> getCountries() {
    ArrayList<Country> competitions = new ArrayList<>();
    competitions.add(new Country("Germany", "id_0"));
    competitions.add(new Country("Czech Republic", "id_1"));
    competitions.add(new Country("Slovakia", "id_2"));
    competitions.add(new Country("Poland", "id_3"));

    return Observable.just(competitions);
}

最后合并逻辑。我希望你熟悉 lambda：

public void fun() {
    Observable.zip( // (1)
            getCompetitions(),
            getCountries(),
            Pair::create)
            .flatMap(pair -> getCompetitionsWithCountries(pair.first, pair.second)) // (2)
            .subscribeOn(Schedulers.immediate())
            .observeOn(Schedulers.immediate())
            .subscribe(countryCompetitions -> {
                for (CountryCompetition countryCompetition : countryCompetitions) {
                    System.out.print(countryCompetition.toString()); // (6)
                }
            });
}

public Observable<List<CountryCompetition>> getCompetitionsWithCountries(List<Competition> competitions, List<Country> countries) {
    return Observable.from(competitions) // (3)
            .map(competition -> {
                Country country = searchForCountry(countries, competition.countryId); // (4)
                return new CountryCompetition(country, competition);
            })
            .toList(); // (5)
}

public Country searchForCountry(List<Country> countries, String countryId) {
    for (Country country : countries) {
        if (country.id.equals(countryId)) {
            return country;
        }
    }

    throw new RuntimeException("Country not found");
}

有趣部分的解释。 ：

1. zip() 运算符接受两个 observables 并产生 Pair 的结果。在这种情况下，两个列表对。
2. flatMap() 运算符接受这对并返回 Observable<List<CountryCompetition>> 类型的新 observable
3. from() 运算符获取 Competitions 的列表，并将每个 Competition 项目单独发送到 Rx 链。
4. 现在我们有了单一的比赛和国家列表，所以我们根据countryId 搜索Country 对象。现在我们同时拥有Competition 和Country 对象，我们可以将map() 它们转换为新的CountryCompetition 对象。
5. 将所有发出的CountryCompetition 项目打包回列表。
6. 打印结果。对象countryCompetitions 的类型是List<CountryCompetition>

Answer 2

不是很复杂：

  public Observable<CountryCompetition> getCountyComps(){
      return getCompetitions()
                 .flatMap(competition ->
                      Observable.zip(
                      Observable.just(competition),
                      getCountry(competition.id)),
                      (comp, country) -> new CountryCompetition(comp, country))
    }