Массив RxJs наблюдаемого массива

Для веб-приложения, написанного с помощью Angular2 в TypeScript, мне нужно работать с RxJs Observable s.
Поскольку я никогда не использовал rxjs раньше, и я новичок в реактивном программировании в целом, у меня иногда возникают некоторые трудности с поиском правильного пути для выполнения определенных задач.
Теперь я столкнулся с проблемой. где мне нужно преобразовать Array<Observable<T>> в Observable<Array<T>>.
Я попытаюсь объяснить это на примере:
- У меня есть Observable, который дает мне список Users (Observable<Array<User>>)
- User -class имеет функцию getPosts возвращающую Observable<Array<Post>>.
- Мне нужно сопоставить Observable<Array<User>> в Observable<Array<Post>> чтобы оценить все Post внутри функции onNext.

Я могу легко отобразить из Observable<Array<User>> в Observable<Array<Observable<Array<Post>>>> используя
map((result: Array<User>) => result.map((user: User) => user.getPosts()))
ans Я могу свернуть Array<Array<Post>> в Array<Post>.
Однако я просто не могу найти правильный способ сопоставить Observable<Array<Observable<Array<Post>>>> в Observable<Array<Array<Post>>>
До сих пор я использовал функцию combineLatest вместе с flatMap.
Мне казалось, что это работает, и редактор, который я использовал (редактор Atom), не обнаружил ошибок. Однако теперь я использую Netbeans, который показывает мне ошибку в этом коде. Также компиляция кода с использованием "tsc" приводит к ошибкам.
Это выглядит так:

Argument of type '(result: Post[]) => void' is not assignable to parameter of type 'NextObserver<[Observable<Post>]> | ErrorObserver<[Observable<Post>]> | CompletionObserver<[...'.
Type '(result: Post[]) => void' is not assignable to type '(value: [Observable<Post>]) => void'.

Поэтому мой вопрос:
Как я могу "сгладить" Array Observables в Array?

Ответ 1

Оператор flatMap позволяет это сделать. Я не совсем понимаю, что вы пытаетесь сделать, но я постараюсь дать ответ...

Если вы хотите загрузить все

getPostsPerUser() {
  return this.http.get('/users')
    .map(res => res.json())
    .flatMap((result : Array<User>) => {
      return Observable.forkJoin(
        result.map((user : User) => user.getPosts());
    });
}

Observable.forkJoin позволяет подождать, пока все наблюдаемые данные будут получены.

В приведенном выше коде предполагается, что user.getPosts() возвращает наблюдаемый...

При этом вы получите массив массивов сообщений:

this.getPostsPerUser().subscribe(result => {
  var postsUser1 = result[0];
  var postsUser2 = result[1];
  (...)
});

Ответ 2

Вы можете использовать функцию apply по методу rxjs, который вы хотите, например:

const source1 = Rx.Observable.interval(100)
  .map(function (i) { return 'First: ' + i; });

const source2 = Rx.Observable.interval(150)
  .map(function (i) { return 'Second: ' + i; });

const observablesArray = [source1, source2];

const sources = Rx.Observable.combineLatest
.apply(this, observablesArray).take(4)

/* or you can write it without "apply" this way:
const sources = Rx.Observable
                .combineLatest(...observablesArray)
                .take(4)
*/
sources.subscribe(
  (response) => {
    console.log(response);
  }
)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>

Ответ 3

Используйте forkJoin в этом массиве наблюдаемых, тогда это будет наблюдаемый массив.