Angular 处理多个依赖订阅

Question

有什么帮助吗？

let notificationsMessages = []
countries.forEach((country: any) => {
    this.isActiveCountry(country.isActive).subscribe((data) => { // // CALL #1 TO API
        country.serverId = data.serverId;
        this.uploadPhotoToApi(country.fileSource).subscribe((response) => { // CALL #2 TO API
          // server return the uploaded file ID
          country.serverFileID = response.serverFileId;
          this.sendCountryToApi(country).subscribe((response) => { // CALL #3 TO API
            this.countriesTable.delete(country.id).then(() => {
              // Delete the uploaded country from local database
              // if is the last country EMIT EVENT
            }, (error) => {
              // if is the last country EMIT EVENT
              notificationsMessages.push(error); // Error to delete country from indexedDB 
            });
          }, (error) => {
              // if is the last country EMIT EVENT
              notificationsMessages.push(error); // // Error to upload country to API
          });
        }, (errorCode) => {
              // if is the last country EMIT EVENT
          notificationsMessages.push(error); // Error on sending file to API
        });
      }, (error) => {
              // if is the last country EMIT EVENT
            notificationsMessages.push(error); // // Error on country identification
      });
  });
  

处理完所有country 列表后，如何发出事件？ 我需要知道有多少国家上传成功，有多少没有。

例如，如果我有一个包含 50 个国家/地区的列表，则在处理最后一个国家/地区时，我想发出一个包含 2 个数组的事件……如下所示： 成功：[countryId1, countryId2...] 错误：['Country Id 2 failed on upload', 'Country Id 10 failed on file upload']

所有这 3 个调用都是依赖的，必须按该顺序执行……我无法更改此流程。 我应该在 CALL #3 成功以及所有错误函数上发出事件吗？ 谢谢！

Answer 1

这是执行此操作的一种方法。这可能有点过头了，因为它可以让您对错误处理进行大量精细控制，但基本上总是以相同的方式处理错误。

即便如此，这可能比最直接的解决方案更容易扩展。

这里：

interface TaggedCountry{
  success: boolean,
  country: any,
  error?: any
}

class ArbiratryClassName {

  processCountry(country: any): Observable<TaggedCountry>{

    return this.isActiveCountry(country.isActive).pipe(
      // country now has serverId set
      map(({serverId}) => ({...country, serverId})),
      catchError(error => throwError(() => ({
        success: false,
        country,
        error
      }) as TaggedCountry)),

      mergeMap((resultCountry: any) => this.uploadPhotoToApi(resultCountry.fileSource).pipe(
        // country now has serverFileId set
        map(({serverFileId}) => ({...resultCountry, serverFileId})),
        catchError(error => throwError(() => ({
          success: false,
          country: resultCountry,
          error
        }) as TaggedCountry))
      )),

      mergeMap((resultCountry: any) => this.sendCountryToApi(resultCountry).pipe(
        // Ignore response from sendCountryToApi
        mapTo(resultCountry),
        catchError(error => throwError(() => ({
          success: false,
          country: resultCountry,
          error
        }) as TaggedCountry))
      )),

      mergeMap((resultCountry: any) => from(this.countriesTable.delete(resultCountry.id)).pipe(
        // Ignore response from countriesTable.delete
        mapTo(resultCountry),
        catchError(error => throwError(() => ({
          success: false,
          country: resultCountry,
          error
        }) as TaggedCountry))
      )),

      map((resultCountry: any) => ({
        success: true,
        country: resultCountry
      }) as TaggedCountry),

      // Convert errors into regular emissions
      catchError((tagged:TaggedCountry) => of(tagged))
    );
  }

  processCountries(countries: any[]): Observable<{success: TaggedCountry[], errors: TaggedCountry[]}>{
    return forkJoin(countries.map(c => this.processCountry(c))).pipe(
      map((tagged: TaggedCountry[]) => ({
        success: tagged.filter(tag => tag.success),
        errors: tagged.filter(tag => !tag.success)
      }))
    )
  }

  doSomethingWith(countries: any[]): void {
    this.processCountries(countries).subscribe({
      next: countries => console.log("All countries processed. Result: ", countries),
      complete: () => console.log("There's only one emission, so this should get called immediately after .next() was called"),
      error: err => console.log("This is a surprise, is there an error we didn't catch earlier? Error: ", err)
    })
  }
}

如果看到以不同的方式完成相同的事情会有所帮助，这里是 processCountry 的较短实现

processCountry(country: any): Observable<TaggedCountry>{

  return this.isActiveCountry(country.isActive).pipe(
    tap((res:any) => country.serverId = res.serverId),

    switchMap(_ => this.uploadPhotoToApi(country.fileSource)),
    tap((res:any) => country.serverFileId = res.serverFileId),

    switchMap(_ => this.sendCountryToApi(country)),
    switchMap(_ => this.countriesTable.delete(country.id)),

    // Tag our result as a success
    map(_ => ({
      success: true,
      country
    }) as TaggedCountry),

    // Tag our errors and convert errors into regular emissions
    catchError(error => of(({
      success: false,
      country,
      error
    }) as TaggedCountry))
  );
}

Answer 2

尽量避免将多个.subscribe(s 相互嵌套的诱惑。正如@praveen-soni 提到的，switchMap 可以提供帮助。

然后要在所有国家/地区都已处理后获得状态，我认为forkJoin 非常适合：它接收一个可观察的列表，并在它们全部完成后发出。

如何构建 observables 列表？您最初有一个国家/地区列表，因此您可以将每个国家/地区映射到处理该国家/地区的 observable。我们也可以使用一个catchError，这样错误就不会关闭整个流，而只会关闭该特定国家/地区的一个。

我认为它看起来像：

const result$ = forkJoin(
  countries.map((country) =>
    this.isActiveCountry(country.isActive).pipe(
      switchMap((data) => {
        country.serverId = data.serverId;
        return this.uploadPhotoToApi(country.fileSource);
      }),
      switchMap((response) => {
        country.serverFileId = response.serverFileId;
        return this.sendCountryToApi(country);
      }),
      switchMap(() => {
        return this.countriesTable.delete(country.id);
      }),
      map(() => {
        // Everything went OK, map it to an OK status
        return {
          type: "success",
        };
      }),
      catchError((error) => {
        // If any step fails, map it to an error type
        return of({
          type: "error",
          error,
        });
      }),
      take(1) // Make sure the observable completes
    )
  )
);

// result$ can now be used as any other observable
result$.subscribe(result => {
  console.log(result);
})