如何在其中一个流依赖于其他流之一的情况下使用 combineLatest？答案

【问题标题】：How to use combineLatest where one of the streams is dependent of one of the others?如何在其中一个流依赖于其他流之一的情况下使用 combineLatest？
【发布时间】：2020-01-30 16:21:59
【问题描述】：

我有一个从后端获取数据的 Angular 解析器。我有以下调用要执行：

GetProject(projectId): Observable<IProject>
GetSites(projectId): Observable<ISites[]>
GetPersons(siteId): Observable<IPerson[]>

我正在尝试使用 combineLatest，但不确定如何在我的场景中使用 RxJs。我希望在解决之前完成所有请求，但是 GetPersons() 应该将 GetSites() 结果中第一项的 id 作为输入。这是怎么做到的？

【问题讨论】：

combineLatest 应该用于热门的 observables。对于冷的，你应该使用forkJoin。此外，它不适合相关的 observables。

标签： angular typescript rxjs combinelatest

【解决方案1】：

看起来你只是想连接几个调用：

forkJoin([GetProject(projectId), GetSites(projectId)]).pipe(
  concatMap(([project, sites]) => {
    const siteId = /* whatever here */;
    return GetPersons(siteId);
  }),
).subscribe(...);

这还取决于您是希望在观察者中接收所有响应还是仅接收最后一个响应。如果您想要所有回复，那么您需要将GetPersons 与map 链接起来，并附加前两个回复：

GetPersons(siteId).pipe(
  map(persons => [project, sites, persons]),
)

【讨论】：

我喜欢这种方法，但我需要创建一个已解析对象才能返回到路由器。我想要类似的东西：resolvedData(project, sites, people);
您可以在 map 运算符中执行此操作，而不是返回数组。

【解决方案2】：

创建一个回放主题：

const sub = new ReplaySubject(3);

然后打电话

this.getProject(1).pipe(
  tap(project => sub.next(project)),
  switchMap(project => this.getSites(1)),
  tap(sites => sub.next(sites)),
  switchMap(sites => this.getPersons(sites[0].id)),
  tap(person => sub.next(person))
);

您的回放主题将包含作为第一个值的项目，作为第二个值的站点，作为第三个值的人。

您可以使用combineLatest 格式和BehaviorSubject。

const obs = new BehaviorSubject([]);
const add = val => obs.pipe(
  take(1),
  map(v => ([...v, val]))
).subscribe(v => obs.next(v));

this.getProject(1).pipe(
  tap(project => add(project)),
  switchMap(project => this.getSites(1)),
  tap(sites => add(sites)),
  switchMap(sites => this.getPersons(sites[0].id)),
  tap(person => add(person))
);

这一次，返回的值将是一个包含所有值的数组。

最后，您可以使用复杂的语法来连接它们，而无需使用主题。

this.getProject(1).pipe(
  switchMap(project => this.getSites(1).pipe(map(sites => ([project, sites])))),
  switchMap(([project, sites]) => this.getPersons(sites[0].id).pipe(map(person => ([project, sites, map])))),
);

【讨论】：

【解决方案3】：

this.project$ = this.myService.getProject(projectId);
this.sites$ = this.myService.getSites(projectId);
this.persons$ = this.sites$.pipe(
  switchMap(
    (sites: ISites[]) => merge(...sites.map((site: ISites) => this.myService.getPersons(site.id))),
  ),
); // that should result in Observable<IPerson[][]>, you likely need to flatten it

【讨论】：