RxJava - Combine two Observables

Question

I need to merge Images from another Retrofit request to its specific series. What sounds like an easy task has brought me deep into reactive hell with no sign of hope.

Specifically, I also create the Service via an Observable since I need to obtain an authentication token to create the service.

I based my attempts on this answer: https://stackoverflow.com/a/28418503/2192545.

public void loadSeries(String searchFor) {

    mainView.showLoadingIndicator();
    if (subscription != null) subscription.unsubscribe();


    TVApplication application = TVApplication .get(mainView.getContext());

    application.getTVService()
            .flatMap(service -> service.searchSeries(searchFor)
                    .flatMap(series -> Arrays.asList(series).stream()
                            .<List<Series>>flatMap(aSeries -> {
                                Observable<Series> obsSeries = Observable.just(aSeries);
                                Observable<ImageListWrapper> obsImages = service.queryImages(aSeries.getId(), "poster");
                                return Observable.zip(obsSeries, obsImages, (Series s, ImageListWrapper i) -> combineSeriesData(s, i));
                            })));
}

private Series combineSeriesData(Series s, ImageListWrapper i) {
    s.setPosters(i.getImages());
    return s;
}

I'm in way over my head. I just get "Cannot infer functional interface type" on the Func2 part of Observable.zip in the IDE, the build log says:

Error:(63, 58) error: incompatible types: bad return type in lambda 

expression
no instance(s) of type variable(s) T1,T2,R exist so that Observable<R> conforms to Stream<? extends List<Series>>
where T1,T2,R are type-variables:
T1 extends Object declared in method <T1,T2,R>zip(Observable<? extends T1>,Observable<? extends T2>,Func2<? super T1,? super T2,? extends R>)
T2 extends Object declared in method <T1,T2,R>zip(Observable<? extends T1>,Observable<? extends T2>,Func2<? super T1,? super T2,? extends R>)
R extends Object declared in method <T1,T2,R>zip(Observable<? extends T1>,Observable<? extends T2>,Func2<? super T1,? super T2,? extends R>)

score 2 · Accepted Answer · answered Sep 05 '16 at 18:41

Ok, first things first; don't create the Service as an Observable, it will make your code harder. There are ways around that.

Secondly, you don't need to use zip:

application
  .getTVService()
  .flatMap(service ->
     service
     .searchSeries(searchFor)
     .flatMapIterable(series -> series)
     .flatMap(aSeries -> 
        service
        .queryImages(aSeries.getId(), "poster")
        .map(i ->  combineSeriesData(aSeries, i))
     )
  );

RxJava - Combine two Observables

1 Answers1