Observable.forkJoin() не выполняется

У меня есть следующий код:

//Loop: For each user ID/Role ID, get the data
userMeta.forEach((businessRole) => {
  Observable.forkJoin(
    af.database.object('/roles/'+businessRole.$value),
    af.database.object('/users/'+businessRole.$key)
  ).subscribe(
    data => {
      console.log("Data received");
      data[1].role = data[0];
      this.users.push(data[1]);
    },
    err => console.error(err)
  );

Я пытаюсь подписаться на результат из 2 наблюдаемых с помощью forkJoin.

По некоторым причинам сообщение "Полученные данные" не отображается.

Мои переменные userMeta отлично смотрятся на console.log:

Что не так?

Обновление: следующий код ничего не возвращает ничего

let source = Observable.forkJoin(
        af.database.object('/roles/'+businessRole.$value),
        af.database.object('/users/'+businessRole.$key)
    );
    let subscription = source.subscribe(
      function (x) {
    console.log("GOT: " + x);
  },
  function (err) {
    console.log('Error: %s', err);
  },
  function () {
    console.log('Completed');
  });

То, что я на самом деле пытаюсь сделать, это повысить производительность следующего кода:

//Subscription 3: role ID to role Name
        af.database.object('/roles/'+businessRole.$value)
        .subscribe((roleData) => {
        //Subscription 4: Get user info
        af.database.object('/users/'+businessRole.$key).subscribe(user => {

Ответ 1

forkJoin() требует, чтобы все исходные Observables forkJoin() хотя бы один раз и завершили.

Следующая демонстрация завершается, как и ожидалось:

const source = forkJoin(
  from([1,2,3]),
  from([9,8,7,6])
).subscribe(
  x => console.log('GOT:', x),
  err => console.log('Error:', err),
  () => console.log('Completed')
);

Демонстрационная версия: https://stackblitz.com/edit/rxjs-urhkni

GOT: 3,6
Completed

Январь 2019: обновлено для RxJS 6

Ответ 2

У меня была аналогичная проблема с использованием Angular 2/Angularfire 2, в частности, где я искал, существуют ли пользователи по электронной почте. В одном случае пользователь существует, и я получил массив из одного объекта из Observable. В другом случае пользователь не существовал, и я получил пустой массив.

Когда я использовал forkJoin с resultSelector и подпиской, ни функция выбора, ни функция подписки никогда не работали. Однако, когда я пробовал

Observable.zip(
  FirebaseListObservable,
  FirebaseListObservable,
  (...results) => {
    return results.map(some code here)
  }
).subscribe(res => console.log(res));

Работают как селектор, так и подписка. Я предполагаю, что это связано с ответом @martin, где forkJoin требует, чтобы наблюдаемые заканчивались, потому что по определению он возвращает последние выбросы. Если наблюдаемое никогда не завершается, я полагаю, что он никогда не может иметь последнего излучения.

Возможно, наблюдаемые в списке угловых огней (или объектные наблюдаемые в вашем случае) никогда не завершатся, что делает невозможным использование forkJoin. К счастью, zip имеет подобное поведение и все еще работает, разница в том, что он может повторяться несколько раз, если данные изменяются в Firebase, где forkJoin объединяет только последний ответ.

В моем случае я рассматриваю либо 1) использование zip и принятие того, что мой код может выполняться несколько раз, если пользовательские данные изменяются, пока .zip все еще работает, 2) вручную отключает zip после первого набора данных возвращается, или 3) канава Angularfire и попробуйте Firebase api напрямую, используя что-то вроде .once, чтобы увидеть, могу ли я получить наблюдаемое, которое завершает и запускает forkJoin.

Ответ 3

У меня возникла аналогичная проблема: я создавал список наблюдаемых динамически, и я заметил, что forkjoin() никогда не испускает и не завершает работу, если список наблюдаемых пуст, тогда как Promise.all() разрешается с пустым списком:

Observable.forkJoin([])
    .subscribe(() => console.log('do something here')); // This is never called

Обходной путь, который я нашел, - проверить длину списка и не использовать этот оператор, когда он пуст.

return jobList.length ? Observable.forkJoin(jobList) : Observable.of([]);

Ответ 4

Просто добавьте наблюдатель .complete();

Не будет работать:

observer.next(...)

Буду работать:

observer.next(...);
observer.complete();

Надеюсь, поможет.