RxJava zip для объединения вызовов API - PullRequest
0 голосов
/ 16 ноября 2018

Я пытаюсь сделать параллельные вызовы API, используя Observable.zip. Каждый из этих вызовов API возвращает Observable>. Я создаю список наблюдаемых и отправляю этот список в zip вместе с функцией, которая возвращает Observable>, составленную из отдельных Observable>, которые возвращаются этими вызовами API. По некоторым причинам объект [] инициализируется с помощью 20 элементов TreeSet, что нормально, но каждый TreeSet пуст, хотя в журналах я могу обнаружить ответ для вызовов API, содержащих данные.

subscription = repository
            .getFollowers(id, OFFSET)
            .subscribeOn(Schedulers.io())
            .map(FollowerResponse::getCollection)
            .flatMap((Function<List<Follower>, Observable<TreeSet<Track>>>) followers -> {
                List<Observable<TreeSet<Track>>> list = new ArrayList<>();
                observable = repository.getTracks(OFFSET, String.valueOf(followers.get(0).getId()));
                list.add(observable.subscribeOn(Schedulers.io())
                        .onErrorResumeNext(Observable.empty()));
                for (int i = 1; i < followers.size(); i++) {
                    observable = Observable.merge(observable, repository.getTracks(OFFSET, String.valueOf(followers.get(i).getId())));

                    list.add(observable.subscribeOn(Schedulers.io())
                            .onErrorResumeNext(Observable.empty()));
                }

                return Observable.zip(list, new Function<Object[], TreeSet<Track>>() {
                    @Override
                    public TreeSet<Track> apply(Object[] objects) throws Exception {
                        TreeSet<Track> treeSet = new TreeSet<>();
                        Log.e("zip ", "inside zip");
                        for (Object obj : objects) {
                            treeSet.addAll((TreeSet<Track>) obj);
                        }
                        return treeSet;
                    }

                });
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<TreeSet<Track>>() {
                @Override
                public void accept(TreeSet<Track> tracks) throws Exception {
                    Log.e("zip ", "inside subscribe");
                }
            });

1 Ответ

0 голосов
/ 16 ноября 2018

блин, я только что понял, что не удаляю Observable.merge, который я пробовал до Observable.zip! Действительно помогает отправлять сообщения, это как думать вслух LOL

...