concatAll 发射多个 observables 并且不订阅它们

Question

我正在尝试从一个 API 端点获取课程列表，并从另一个端点获取与每门课程关联的作业。我正在使用 concatMap 合并可观察对象，但它没有按我预期的那样工作。它不是发出每门课程的作业，而是发出可观察的对象。

function getCourses() {
  return fromFetch("https://canvas.instructure.com/api/v1/courses", {
    method: "GET",
    headers: {
      "Content-Type": "application/json",
      Authorization:
        "Bearer 2~WSiTptufbMw24WKz0gWbb34tSabWEbKKVR5bQY3NEL2b5pd7Ah8OEzM8P64FS4ta",
    },
    selector: responseAsJson,
  });
}

function getAssignments(id: number) {
  return fromFetch(
    `https://canvas.instructure.com/api/v1/courses/${id}/assignments?bucket=upcoming`,
    {
      method: "GET",
      headers: {
        "Content-Type": "application/json",
        Authorization:
          "Bearer ***",
      },
      selector: responseAsJson,
    }
  );
}

const assignments = getCourses()
  .pipe(
    map((response) => response.map((r: any) => getAssignments(r.id))),
    concatAll(),
)
  
assignments.pipe(tap(r => console.log(r))).subscribe();
/*
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
Observable { _subscribe: [Function (anonymous)] }
*/

Answer 1

我能够通过使用 switchMap 和 forkJoin

使其工作

const assignments = getCourses().pipe(
  // unsubscribe from the outer observable after the inner emits
  switchMap((response) =>
    // convert a collection of observables into an observable collection
    forkJoin(response.map((r: any) => getAssignments(r.id)))
  )
);

Answer 2

发生这种情况是因为您的 map() 运算符 returns Promise[] 这意味着它 return 一个 Promise 数组。 concatAll() 将数组作为类似可观察的输入，迭代其值并将它们作为单独的发射发射。换句话说，concatAll() 将“订阅”数组而不是数组内的 Observables。

因此您需要将数组展平为单独的发射，例如 mergeMap():

const assignments = getCourses()
  .pipe(
    mergeMap((response) => response.map((r: any) => getAssignments(r.id))),
    concatAll(),
)

... 或者您也可以使用两个 cocantAll() 但这看起来真的很奇怪：

const assignments = getCourses()
  .pipe(
    map((response) => response.map((r: any) => getAssignments(r.id))),
    concatAll(),
    concatAll(),
)

第一个 concatAll 将展平数组，第二个将订阅单个 Observables。

...或者你可以合并所有内部 Observables

const assignments = getCourses()
  .pipe(
    mergeMap((response) => merge(...response.map((r: any) => getAssignments(r.id)))),
)

concatAll 发射多个 observables 并且不订阅它们

concatAll emits multiple observables and is not subscribing to them

observable

rxjs

typescript