RxJava цепочка запросов плоской карты

прежде всего вам нужно избавиться от параллелизма сObservOn и использовать subscribeOn.

.observeOn(Schedulers.io())

Если вы хотите синхронизировать данные из другого потока обратно в цикл обработки событий, рассмотрите возможность использованияObservOn с AndroidScheduler. Обычно перед подпиской на наблюдаемый объект следует использовать visibleOn, чтобы синхронизироваться с циклом пользовательского интерфейса и изменить информацию пользовательского интерфейса.

.observeOn(AndroidSchedulers.mainThread())

Во-вторых, не рекомендуется изменять объекты в конвейере. Вы должны вернуть новый объект очень вовремя.

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

Я попытался реорганизовать ваше решение с учетом первых двух пунктов. Я использую RxJava2-RC5.

У оператора flatMap много перегрузок. Один из них предоставляет функцию для объединения входящего значения и созданного значения.

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

Helping-метод для получения информации для каждого элемента в Rss. Рассмотрите возможность использования перегрузки с maxConcurrency, поскольку по умолчанию он будет подписываться на каждый поток сразу. Поэтому flatMap создаст много http-запросов.

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-адреса изображения. Возврат наблюдаемого не влияет на параллелизм. В случае ошибки в качестве значения по умолчанию будет возвращена пустая строка.

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/anonymous/a8e36205fc2430517c66c802f6eef38e

Нельзя смешивать внутренний параметр одного параметра RxJava (параметр flatMap lambda) с другим параметром оператора (defaultIfEmpty).

Прежде всего, создайте вспомогательную функцию, чтобы поддерживать чистоту основного реактивного потока:

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

Затем переформатируйте основную функцию:

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

Надеюсь, я правильно прицелился. Это может не сработать, так как я не могу проверить это, однако я надеюсь, что вы поняли идею. Причина, по которой я использовал функции .zip, заключается в том, что вы не можете потерять ссылку на анализируемые в данный момент item или rss.