Я пытаюсь сделать параллельные вызовы 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");
}
});