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