RX слияние нескольких наблюдаемых

Question

В моем фрагменте я делаю запрос api, и все работает нормально. Теперь я хочу проверить ответ, чтобы узнать, больше ли количество доступных на сервере элементов, чем возвращаемый результирующий набор, заданный в таблице. Если это так, я хотел бы сделать дополнительные вызовы api, пока все результаты не будут получены и возвращены подписчику в качестве одного ответа. Какой оператор RX мне нужно использовать для достижения этого и как приостановить возврат ответа ниже до тех пор, пока последующие вызовы api не будут завершены?

getAction<T>(path: string, params?: {}): Observable<T> { 
 
     return this._http.get("url") 
 
      .map(res => { 
 
       let response = res.json(); 
 
       // If more pages available make additional api calls & return as single result 
 
       return response; 
 
      }); 
 
    }

Answer 1

Посмотрите на expand.

Для извлечения нескольких страниц данных рекурсивно, вы могли бы сделать что-то вроде:

class MyExample { 
    search(offset) { 
    return this.http.get(`/search?offset=${offset}`); 
    } 

    searchAll() { 
    return this.search(0) 
       .expand(results => { 
       if (loadNextPage(results)) { 
        return this.search(results.nextPageOffset); 
       } else { 
        return Observable.empty(); 
       } 
       }); 
    } 
} 

expand позволяет сделать некоторые обработки на основе предыдущих результатов (как проверить, если есть несколько страниц), и указать Observable с большим количеством результатов. Результаты всех этих вызовов будут объединены, не нужно беспокоиться о переносе их над собой.

Answer 2

Вы должны использовать switchmap, чтобы получить ответ от другого наблюдаемого до тех пор, пока есть все данные. Просто конкатенирует все ответы и возвращает их как ответ в последнем наблюдаемом. Что-то вроде:

//emit immediately, then every 5s 
const source = Rx.Observable.timer(0, 5000); 

//switch to new inner observable when source emits, emit items that are emitted 
const example = source.switchMap(() => Rx.Observable.interval(500)); 

//output: 0,1,2,3,4,5,6,7,8,9...0,1,2,3,4,5,6,7,8 
const subscribe = example.subscribe(val => console.log(val));