Как завершить Наблюдаемость Наблюдаемых

Question

hi Иметь действие (NgRx), которое запускает функцию (которая должна возвращать наблюдаемую), которая имеет цикл, для каждого цикла цикла мне нужно выдавать (следующий) значение, функцию в цикле, которая выбираетданные возвращают наблюдаемую, я хочу завершить наблюдаемую, которая имеет цикл, когда все наблюдаемые завершены:

вот код:

      public outerFunction(
        collections: someObject
      ): Observable<someOtherObject> {

        const outerObservable$ = new Observable<Observable<someOtherObject>>(
          (observer) => {

            const collKeys = Object.keys(collections);
            for (const id of collKeys) {
              if (collections[id]) {
                const innerObs$ = this.functionThatReturnsObs(
                  collections[id]
                )
                observer.next(innerObs$);
              }
              // observer.complete(); <-- If I complete here the stream is interrupted
            }
          }
        );
return outerObservable$.pipe(mergeAll());
    }

Answer 1

На самом деле я нашел ответ из RxJs Docs

код был полностью в порядке без полного, так как в конце я позвонил mergeAll()

mergeAll подписывается на Observable, который испускает Observables, также известный как Observable высшего порядка.Каждый раз, когда он наблюдает за одной из этих испускаемых внутренних Наблюдаемых, он подписывается на это и доставляет все значения из внутренней Наблюдаемой на выходную Наблюдаемую. Вывод Observable завершается только после того, как все внутренние Observables завершены.

в конце mergeAll () завершит мою наблюдаемую

Answer 2

Можете ли вы попробовать это

return outerObservable$.pipe(mergeMap(res=>merge(res))

, если это не сработает, вы можете обработать его как обычный массив

   public outerFunction(
        collections: someObject
      ): Observable<someOtherObject> {
            const innerObs$=[]
            const collKeys = Object.keys(collections);
            for (const id of collKeys) 
              if (collections[id]) 
                innerObs$.push(this.functionThatReturnsObs(
                  collections[id]
                ))

             return merge(innerObs$);
    }