带有并行内部调用的顺序 rxjs 调用

Question

使用 rxjs，我想使用之前调用返回的值进行一组 http 调用。我希望内部调用并行执行。我还需要在第一次调用可用时立即返回该值。我需要能够订阅结果并处理任何错误。内部调用产生的错误不应导致同级调用取消。

我有以下代码可以工作，但由于使用了forkJoin，所以在内部调用完成之前等待返回初始值：

public create(customer: Customer, groupIds: number[]): Observerable<Customer> {
    const response = this.http.post<Customer>('/customer', customer).pipe(mergeMap(
        (created) => {
            return forkJoin([
                of(created),
                forkJoin(groupIds.map(groupId => {
                    const membership = new Membership(created.Id, groupId);
                    return this.http.post<Membership>('/membership', membership);
                )
            ]);
        }
    )).pipe(map(a => a[0]));

    return response;
}

有没有办法在不等待内部调用完成的情况下返回创建的客户？除此之外，有没有办法把上面的代码写得更简洁？

（注意：this.http 是来自 Angular 的 HttpClient 类型，并返回 Observable<T>）

Answer 1

如果您不需要使用并行调用中的值（就像您当前代码的情况一样）：

public create(customer: Customer, groupIds: number[]): Observerable<Customer> {
  const response = this.http.post <Customer>('/customer', customer)
    .pipe(
      tap(() => groupIds.map(groupId => {
          const membership = new Membership(created.Id, groupId);
          return this.http.post <Membership>('/membership', membership).subscribe();
        ))
      }
    ));

  return response;
}

如果您确实需要使用数据，可以使用startWith 代替forkJoin

public create(customer: Customer, groupIds: number[]): Observerable<Customer> {
    const response = this.http.post<Customer>('/customer', customer).pipe(mergeMap(
        (created) => {
            return forkJoin(groupIds.map(groupId => {
                    const membership = new Membership(created.Id, groupId);
                    return this.http.post<Membership>('/membership', membership);
            ).pipe(startWith(created);
        }
    ));

    return response;
}

Answer 2

如果您有客户的局部变量，您可以执行以下操作：

public create(customer: Customer, groupIds: number[]): Observerable<Customer> {
    const response = this.http.post<Customer>('/customer', customer).pipe(
      tap(created => this.customer = created), // <---
      mergeMap((created) => {
            return forkJoin([
                of(created),
                forkJoin(groupIds.map(groupId => {
                    const membership = new Membership(created.Id, groupId);
                    return this.http.post<Membership>('/membership', membership);
                )
            ])
        }
    )).pipe(map(a => a[0]));

    return response;
}

在mergeMap 之前添加一个点击将客户设置为一个局部变量。然后，您可以在子操作完成之前访问创建的客户。

我在这里做了一个堆栈闪电战：https://stackblitz.com/edit/angular-create-todo-deborahk

（它使用待办事项/帖子而不是客户/组，但在概念上应该相似。）

或者

你可以这样做：

  private todoCreatedSubject = new BehaviorSubject<ToDo>(null);
  todoCreatedAction$ = this.todoCreatedSubject.asObservable();

  private groupSubject = new BehaviorSubject<number[]>([5, 10]);
  groupAction$ = this.groupSubject.asObservable();

  todo$ = this.todoCreatedAction$.pipe(
    mergeMap(todo =>
      this.http.post<ToDo>(this.todoUrl, todo, { headers: this.headers })
    )
  );

  posts$ = combineLatest([this.todo$, this.groupAction$]).pipe(
    switchMap(([created, groupIds]) =>
      forkJoin(
        groupIds.map(groupId => {
          const post = {
            userId: created.userId,
            title: "Post:" + groupId,
            body: created.title
          } as Post;
          return this.http
            .post<Post>(this.postUrl, post, { headers: this.headers })
            .pipe(tap(post => console.log(post)));
        })
      )
    )
  );

它定义了两个独立的流。 todo$ 与您的客户类似，发布后可立即访问。

post$ 将类似于您的会员资格。

我还在这里为这个创建了一个堆栈闪电战：https://stackblitz.com/edit/angular-create-declarative-todo-deborahk

Answer 3

我的理解是你想构建一个 Observable

1. 在获取客户后立即发出客户（“我还需要在第一次调用可用时立即返回值”）但是
2. 使用客户数据并行进行一系列后续 http 调用，以保存会员数据并在所有会员保存后发出（即当所有并行 http 调用返回时）

如果我的理解是正确的，我会如下进行。

首先，我将创建一个函数，该函数接受一个 customerId 和一个 groupIds 数组，并返回一个使用 forkJoin 实现并行调用的 Observable，类似于

function createMemberships(customerId: string, groupIds: number[]) {
  return forkJoin(groupIds.map(gId => 
    createMembershipHttpSimulation(customerId, gId))
  )
}

然后我会创建第二个函数来返回你想要的 Observable，就像这样。

function create(customer: Customer, groupIds: number[]) {
  const createCustomerObs = createCustomerHttpSimulation(myCustomer).pipe(
      shareReplay(1)
  )
  const createMembershipsObs = createCustomerObs.pipe(
    concatMap(customer => createMemberships(customer.id, groupIds))
  )
  return concat(createCustomerObs, createMembershipsObs)
}

此函数首先创建 Observable createCustomerObs，它调用远程 API 来创建 Customer，并确保使用 shareReplay(1) 共享其订阅。然后创建第二个 Observable，createMembershipsObs，将createCustomerObs 传递到forkJoin 中，以触发并行创建成员资格的请求。最后，它通过concat、createCustomerObs 和createMembershipsObs 连接起来。这种连接的结果是另一个 Observable，它在createCustomerObs 发出时立即发出，然后在createMembershipsObs 发出时发出，即在并行调用完成时发出。使用shareReplay 可确保只调用一次客户创建 API。

Here a stackblitz 显示了这种方法。我使用一些延迟在控制台上显示先发出客户，然后发出会员资格