Retrofit + Rxjava. Problem z funkcją Observable.zip()

Question

Witam. W aplikacji Android próbuję wykonać kilka zapytań za pomocą biblioteki Retrofit , żeby pobrać dane w formacie JSON z endpointa na przeglądarce. Żeby zsynchronizować zapytania używam biblioteki Rxjava. Próbuję zrobić to w ten sposób że tworzę dynamicznie listę obiektów Observable a następnie przesyłam je do funkcji Observable.zip() żeby ze wszystkich danych zrobić jeden obiekt, a następnie za pomocą funkcji subscribe() wykorzystać dane z tego obiektu. Niestety ciągle mam błąd w funkcji Observable.zip() . Szukałem długo po internecie rozwiązania ale zazwyczaj są niekompletne lub przestarzałe. Proszę o pomoc i z góry dziękuję :) . Zamieszczam mój kod:

            //Tworze liste Observable 
            final List<Observable<PojoData>> requests = new ArrayList<>();

            //Tworze dynamicznie zapytania Retrofit i dodaje je do listy Observable
            Iterator iter = nodesId.iterator();
            while (iter.hasNext()) {
                int nodeNumber = (int) iter.next();
                requests.add(service.getData("data",String.valueOf(nodeNumber), String.valueOf(minDay)+","+ String.valueOf(maxDay)));
            }

            //Funkcja zip w ktorej jest blad
            Observable<PojoAllSecondActivity> combined = Observable.zip(requests, new Function<PojoData[], PojoAllSecondActivity>() {
                @Override
                public PojoAllSecondActivity apply(PojoData[] pojoData) throws Exception {
                    return new PojoAllSecondActivity(Arrays.asList(pojoData));
                }
            });

            // obiekt Consumer do pozniejszej funkcji subscribe()
            Consumer<PojoAllSecondActivity> subscriber = new Consumer<PojoAllSecondActivity>() {
                @Override
                public void accept(PojoAllSecondActivity pojoAllSecondActivity) throws Exception {
                    Log.d(TAG, "accept:  " + pojoAllSecondActivity.getPojoDataList().get(0).getDataOutputs().get(0).getCreatedat());
                }
            };

            //subscribe()
            combined.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(subscriber);

Klasa PojoAllSecondActivity:

public class PojoAllSecondActivity {
    public List<PojoData> pojoDataList;

    public PojoAllSecondActivity(List<PojoData> pojoDataList) {
        this.pojoDataList = pojoDataList;
    }

    public List<PojoData> getPojoDataList() {
        return pojoDataList;
    }

    public void setPojoDataList(List<PojoData> pojoDataList) {
        this.pojoDataList = pojoDataList;
    }
}

Klasa PojoData:

public class PojoData {

    @SerializedName("output")
    @Expose
    private List<DataOutput> dataOutputs = null;

    public List<DataOutput> getDataOutputs() {
        return dataOutputs;
    }

    public void setOutput(List<DataOutput> dataOutputs) {
        this.dataOutputs = dataOutputs;
    }

}

 

Answer 1

Możesz to zrobić prościej:

Observable.concat(requests)
    .toList()
 

Co będzie typem Observable<List<PojoData>> więc w onNext dostaniesz już List<PojoData>.

Poza tym zapytania z retrofita są wykonywane raz więc możesz użyć Single zamiast Observable i zrobić tak samo jak powyżej podałem

Comments

Dzięki ! Ostatecznie jednak wczoraj udało mi się zmieniająć PojoData[] na Object[] tak jak poniżej, chociaż nie mam pojęcia czemu dopiero to zadziałało.  Mógłbyś mi jeszcze wytłumaczyć czy w tej sytuacji lepiej użyć Single czy to bez znaczenia skoro działa? Bo jak rozumiem te zapytania HTTP są wykonywane równocześnie asynchronicznie a funkcja zip() czeka aż każde zapytanie się zakończy i zbiera z nich wyniki? :) Jestem początkujący w tym :)

Observable<PojoAllSecondActivity> combined = Observable.zip(requests, new Function<Object[], PojoAllSecondActivity>() {
                @Override
                public PojoAllSecondActivity apply(Object[] objects) throws Exception {
                    ArrayList<PojoData> pojoDataArrayList = new ArrayList<>();
                    for (Object o : objects) {
                        pojoDataArrayList.add((PojoData) o);
                    }
                    return new PojoAllSecondActivity(pojoDataArrayList);
                }
            });

 

---

Do zwykłych zapytań http nie ma potrzeby używania Observable bo masz jedną odpowiedź z jednego zapytania, więc w tym wypadku lepiej użyć Single. Single to takie Observable w którym onNext jest wywołane tylko raz. Czyli zamiast Consumerów dla onNext, onComplete oraz onError masz samo onSuccess i onError, tu bardziej chodzi o czystość kodu i eliminowanie niepotrzebnych elementów.

Jeśli chodzi zaś o zip vs concat -> concat zwraca Flowable i wykonuje po kolei wszystkie Single/Observable które były użyte w tym concat i zwraca ich wartości po kolei w onNext, tyle że używając .toList() to samo agreguje ci wynik z onNext w jedną listę zwróconych wartości i zwraca ją w onSuccess. Zip zaś wykonuje wszystko równolegle i zwraca wszystkie wyniki dopiero gdy wszystkie SingleObservable zostaną zakończone.

Więc dla czystości lepiej użyć powyższego rozwiązania z Single i concat. A jeśli potrzebujesz wykonania równoległego to możesz dodatkowo wywołać parallel() po concat.

Single.concat(requests)
    .parallel()
    .toList()

 

---

Rozumiem. Dzięki jeszcze raz :)