Как связать асинхронные операции с использованием Java RX Observable?

Question

Я хочу сделать запрос HTTP повторно и действовать на результат. Я начинаю с public Observable<NewsItem> fetchItems(NewsFeed feed). Один запрос получает несколько новостей, но я решил сгладить его.

Идея состояла в том, чтобы использовать Observable.interval() сделать запрос несколько раз, а затем объединить полученные Observables в один.

 Observable 
      .interval(timePerItem, TimeUnit.MILLISECONDS) 
      .map(i -> feed) 
      .map(feed -> fetchItems(feed)) 
      .subscribe(result -> System.out.println(result)); 

Но результат Observable<Observable<NewsItem>> не Observable<NewsItem>. Как их вывести?

Я нашел оператора marge() (RX-Java doc: Marge). Но, похоже, это не подходит для использования.

В предыдущей версии я использовал CompletableFuture<List<NewsItem>> fetchNewsItems(), но я не смог поместить его в цепь Observable.

Answer 1

Не уверен, что я понимаю проблему, но разве вы не просто ищете flatMap?

Observable 
    .interval(timePerItem, TimeUnit.MILLISECONDS) 
    .flatMap(i -> fetchItems(feed)) 
    .subscribe(result -> System.out.println(result));