RxJava 平面图链接请求 [英] RxJava flatmap chaining requests

问题描述

我正在将 Retrofit 与 RxJAva 用于获取 Rss Feeds 的应用程序，但 rss 不包含所有信息，因此我使用 jsoup 来解析每个项目链接，以检索图像和文章的描述.现在我这样使用它:

i am using Retrofit with RxJAva for an app that gets Rss Feeds, but the rss doesn't contain all the informations so i use jsoup to parse every item link, to retrieve the image and the article's description. now i am using it this way:

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());
}

我在这一行收到一个错误:defaultIfEmpty(rss)它无法识别平面地图的 rss.当我在平面图括号中移动 defaultIfEmpty(rss) 时，我有另一个错误，说必须将返回类型更改为 Element.他们有什么解决办法吗?

and i am getting an error on this line: defaultIfEmpty(rss) it doesn't recognize rss of the flatmap. and when i move the defaultIfEmpty(rss) in flatmap brackets i have another error saying that the return type must be changed to Element. is their any solution ?

推荐答案

首先你需要摆脱所有的并发性，使用 observeOn 并使用 subscribeOn.

first of all you need to get rid of all the concurrency with observeOn and use subscribeOn.

.observeOn(Schedulers.io())

如果要将数据从另一个线程同步回事件循环，请考虑将 observeOn 与 AndroidScheduler 一起使用.通常，您会在订阅 observable 之前使用 observeOn 以同步回 ui-loop 并更改 ui-information.

Please consider using observeOn with AndroidScheduler if want to sync back data from another thread back to the event-loop. Normally you would use observeOn before subscribing to a observable in order to sync back to ui-loop and change ui-information.

.observeOn(AndroidSchedulers.mainThread())

其次，不建议在管道中改变对象.您应该非常及时地返回一个新对象.

Secondly it is not recommended to mutate objects in the pipeline. You should return a new object very time.

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

考虑到前两点，我尝试重构您的解决方案.我正在使用 RxJava2-RC5

I tried to refactor your solution under consideration of the first two points. I am using RxJava2-RC5

flatMap 操作符有很多重载.其中之一提供了将传入值和创建值压缩在一起的功能.

The flatMap operator has many overloades. One of them provides a function to zip together the incoming value and the created value.

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

用于检索 Rss 中每个项目的信息的帮助方法.请考虑将重载与 maxConcurrency 一起使用，因为默认情况下它会立即订阅每个流.因此 flatMap 会创建很多 http 请求.

Helping-method for retrieving information for each item in Rss. Please consider using the overload with maxConcurrency, because on default it will subscribe to every stream at once. Therefore flatMap would create many http-requests.

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();
}

用于检索图像 url 的帮助方法.返回 observable 并不是对并发性的看法.如果发生错误，将返回一个空字符串作为默认值.

Helping-method for retrieving the image url. Returning observable is not opinionated about concurrency. If an error occurs, an empty string will be returned as default value.

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("");
                });
}

您可以在 github.gist 上查看完整示例:https://gist.github.com/匿名/a8e36205fc2430517c66c802f6eef38e

You may look at the full example at github.gist: https://gist.github.com/anonymous/a8e36205fc2430517c66c802f6eef38e

这篇关于RxJava 平面图链接请求的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！