RxJava flatmap Verkettungsanfragen

Question

Ich benutze Retrofit mit RxJAva für eine App, die Rss Feeds erhält, aber der RSS enthält nicht alle Informationen, also verwende ich jsoup, um jeden Artikel Link zu analysieren, um das Bild und die Artikelbeschreibung zu erhalten. Jetzt habe ich es auf diese Weise bin mit:

public Observable<Rss> getDumpData() { 
    return newsAppService.getDumpData() 
      .flatMap(rss -> Observable.from(rss.channel.items) 
      .observeOn(Schedulers.io()) 
      .flatMap(Checked.f1(item -> Observable.just(Jsoup.connect(item.link).get()) 
      .observeOn(Schedulers.io()) 
      .map(document -> document.select("div[itemprop=image] > img").first()) 
        .doOnNext(element -> item.image = element.attr("src")) 
      ))) 
      .defaultIfEmpty(rss) 
      .ignoreElements() 
      .observeOn(Schedulers.io()) 
      .subscribeOn(AndroidSchedulers.mainThread()); 
} 

und ich auf dieser Linie eine Störung erhalte: defaultIfEmpty(rss) es nicht rss des flatmap erkennen. und wenn ich die defaultIfEmpty(rss) in Flatmap-Klammern verschiebe, habe ich einen anderen Fehler, der besagt, dass der Rückgabetyp in Element geändert werden muss. ist ihre Lösung?

Answer 1

zunächst alles, was Sie brauchen, um mit observeOn aller Gleichzeitigkeit loszuwerden und subscribeOn verwenden.

.observeOn(Schedulers.io()) 

Bitte beachten Sie bei AndroidScheduler mit observeOn wenn Sie möchten, Daten von einem anderen Thread synchronisieren zurück zur Event-Schleife. Normalerweise würden Sie observeOn verwenden, bevor Sie eine Observable abonnieren, um zurück zur ui-Schleife zu synchronisieren und ui-Informationen zu ändern.

.observeOn(AndroidSchedulers.mainThread()) 

Zweitens ist es nicht zu empfehlen, Objekte in der Pipeline zu mutieren. Sie sollten ein neues Objekt sehr oft zurückgeben.

.doOnNext(element -> item.image = element.attr("src")) 

Ich habe versucht, Ihre Lösung unter Berücksichtigung der ersten beiden Punkte zu refaktorieren. Ich benutze RxJava2-RC5

Der flatMap-Operator hat viele Überblendungen. Eine davon bietet eine Funktion, um den eingehenden Wert und den erstellten Wert zusammenzufassen.

Observable<Rss> rssItemObservable = newsService.getDumpData() 
       .flatMap(rss -> getRssItemInformation(rss).subscribeOn(Schedulers.io()), 
         (r, rItemList) -> { 
          Rss rInterim = new Rss(); 
          rInterim.items = rItemList; 
          return rInterim; 
         }); 

Hilfe-Methode zum Abrufen von Informationen für jedes Element in Rss. Bitte denken Sie daran, die Überladung mit maxConcurrency zu verwenden, da sie standardmäßig jeden Stream auf einmal abonniert. Daher würde flatMap viele http-Anfragen erstellen.

private Observable<List<RssItem>> getRssItemInformation(Rss rss) { 
     return Observable.fromIterable(rss.items) 
       .flatMap(rssItem -> getImageUrl(rssItem).subscribeOn(Schedulers.io()), (rItem, img) -> { 
        RssItem item = new RssItem(); 
        printCurrentThread("merge1"); 
        item.image = img; 
        item.link = rItem.link; 
        return item; 
       }).toList().toObservable(); 
} 

Hilfe-Methode zum Abrufen der Bild-URL. Zurückkehrendes Observable ist nicht autorisiert über Gleichzeitigkeit. Wenn ein Fehler auftritt, wird eine leere Zeichenfolge als Standardwert zurückgegeben.

private Observable<String> getImageUrl(String link) { 
      return Observable.fromCallable(() -> Jsoup.connect(link).get()) 
       .map(document -> document.select("div[itemprop=image] > img").first()) 
       .map(element -> element.attr("src")) 
       .onErrorResumeNext(throwable -> { 
        return Observable.just(""); 
       }); 
} 

Sie können das vollständige Beispiel auf Github betrachten.gist: https://gist.github.com/anonymous/a8e36205fc2430517c66c802f6eef38e

Answer 2

Sie können den internen Parameter eines Parameters RxJava (Lambda-Parameter flatMap) nicht mit einem anderen Operatorparameter (defaultIfEmpty) mischen.

Zunächst einmal schaffen eine Hilfsfunktion Hauptreaktionsstrom sauberer zu halten:

private Observable<List<Item>> getDetails(List<Item> items) { 
    return Observable.from(items) 
       .observeOn(Schedulers.io()) 
       .flatMap(Checked.f1(item -> 
        Observable.zip(
         Observable.just(item), 
          Observable.just(Jsoup.connect(item.link).get()) 
          .observeOn(Schedulers.io()) 
          .map(document -> document.select("div[itemprop=image] > img").first()), 
          (itemInner, element) -> { 
           itemInner.image = element.attr("src"); 
           return itemInner; 
          } 
        ) 
       )) 
       .toList(); 
} 

Dann Hauptfunktion umformatieren:

newsAppService.getDumpData() 
    .flatMap(rss -> 
     Observable.zip(
      Observable.<Rss>just(rss), 
      getDetails(rss.channel.items), 
      (rssInner, items) -> { 
       rssInner.channel.items = items; 
       return rss; 
      }).onErrorResumeNext((throwable -> Observable.just(rss)) 
     ) 
    ) 
    .observeOn(Schedulers.io()) 
    .subscribeOn(AndroidSchedulers.mainThread()); 

Hoffe, dass ich richtig Ihr Ziel bekam. Es kann nicht funktionieren, da ich es nicht testen kann, aber ich hoffe, Sie bekommen die Idee. Der Grund habe ich .zip Funktionen es, dass Sie nicht den Verweis auf die derzeit item oder analysiert verlieren können rss