如果 RxJS forkjoin 在它正在等待的一个 observable 中有一个 flatMap，则停止订阅

Question

我正在尝试使用带有嵌套 forkjoins 的 rxjs 进行嵌套调用。但是，我面临一个问题，当内部可观察对象内部有一个 flatMap 时，外部 forkJoin 停止 returning 结果。

下面是代码来说明。

  // Only returns a value when the flatMap statements are removed from the inner observables it is waiting for
  searchAdmin(searchKey: string): Observable<[StudentAccountSearchResult[], InstructorAccountSearchResult[]]> {
    return forkJoin(
      this.getStudentAccountSearchResults(searchKey),
      this.getInstructorAccountSearchResults(searchKey)
    );
  }

  private getStudentAccountSearchResults(searchKey: string): Observable<StudentAccountSearchResult[]> {
    return this.getStudents(searchKey).pipe(
      map((students: Students) => students.students),
      // Outer forkjoin in searchAdmin returns when I remove the line below
      flatMap((studentsArray: Student[]) => forkJoin(studentsArray.map(student => this.createStudentAccountSearchResult(student)))),
    );
  }

  private getInstructorAccountSearchResults(searchKey: string): Observable<InstructorAccountSearchResult[]> {
    return this.getInstructors(searchKey).pipe(
      map((instructors: Instructors) => instructors.instructors),
      // Outer forkjoin in searchAdmin returns when I remove the line below
      flatMap((instructorsArray: Instructor[]) => forkJoin(instructorsArray.map(instructor => this.createInstructorAccountSearchResult(instructor))))
    )
  }

getStudentAccountSearchResults 和 getInstructorAccountSearchResults return 都是我直接订阅它们时的正确值，因此在映射可观察对象时没有错误。有谁知道为什么 innerObservables 中的 flatMap 导致外部 forkJoin 停止 returning 任何结果？谢谢！

我正在使用 rxjs 6.4

Answer 1

我想说内部 forkJoin 可能是不必要的，因为已经有外部 forkJoin。目前生成的代码可能类似于 forkJoin(forkJoin(...), forkJoin(...))。相反，我们可以将所有 observable 直接通过管道传输到外部 forkJoin。尝试以下

private getStudentAccountSearchResults(searchKey: string): Observable<StudentAccountSearchResult[]> {
  return this.getStudents(searchKey).pipe(
    map((students: Students) => students.students.map(student => this.createStudentAccountSearchResult(student))),
  );
}

private getInstructorAccountSearchResults(searchKey: string): Observable<InstructorAccountSearchResult[]> {
  return this.getInstructors(searchKey).pipe(
    map((instructors: Instructors) => instructors.instructors.map(instructor => this.createInstructorAccountSearchResult(instructor))),
  )
}

Answer 2

答案如@martin post在对原始 post 的评论中编辑的那样。

我通过在执行平面图之前检查数组是否为空来修复它

 private getInstructorAccountSearchResults(searchKey: string): Observable<InstructorAccountSearchResult[]> {
    return this.getInstructors(searchKey).pipe(
      map((instructors: Instructors) => instructors.instructors),
      // Outer forkjoin in searchAdmin returns when I remove the line below
      flatMap((instructorsArray: Instructor[]) => instructorsArray.length > 0 
        ? forkJoin(instructorsArray.map(instructor => this.createInstructorAccountSearchResult(instructor)))
        : of([]))
    )
  }

如果 RxJS forkjoin 在它正在等待的一个 observable 中有一个 flatMap，则停止订阅

RxJS forkjoin stops subscribing if theres a flatMap inside one of the observables it is waiting for

observable

rxjs

angular

rxjs6