forkJoin 的成功回调没有被调用

Question

我必须在我的 Angular 应用程序中进行两个并行 API 调用，每个调用的数据都将在组件中使用。下面是我的服务和组件文件的代码。我使用forkJoin 进行并行调用和处理。因为根据我的理解forkJoin 确保在两个 API 调用都成功时会调用成功回调。

在运行时，我可以看到成功回调在每个服务中被调用，但在组件文件中没有。

api.services.ts

$getRequestById(id: string): Subject<any> {
    const data_subject = new Subject<any>();
    this.$http.get(API_URL + '/' + id).subscribe(
      (dto) => data_subject.next(new Request(dto)), // THIS IS GETTING EXECUTED
      (error) => data_subject.error(error)
    );
    return data_subject;
  }

  $getAllRequestsForId(id: string): Subject<any> {
    const data_subject = new Subject<any>();

    this.$http.get(API_URL + '?request_id=' + id).subscribe(
      (dto) => data_subject.next(dto.map(i => new Request(i))),
      (error) => data_subject.error(error)
    );

    return data_subject;
  }

在我的组件文件中，

home.component.ts

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests])
 .subscribe(
    (response) => this.setResponse(response), // THIS LINE IS NOT GETTING CALLED.
    (error) => this.notification.error('Error occured while fetching request details! Please try again.')
 );

但是，如果我在任何组件中独立运行此 API 调用，如下所示，它可以正常工作。

this.service.$getRequestById(this.id).subscribe(
(response) => this.setResponse(response), // THIS GETS EXECUTED 
(error) => this.notification.error("An error occured")
)

我在应用程序的其他部分独立使用这些服务没有任何问题，但是当与forkJoin 结合时，没有调用成功回调。知道如何解决这个在forkJoin 和独立调用中都有效的问题吗？

Answer 1

我会说使用 RxJS Subject 会使调用变得不必要地复杂化。您可以从 HttpClient 返回 observable 并使用 forkJoin 组合它们。

来自forkJoindocs：

等待 Observable 完成，然后合并它们的最后一个值 发射。

因此，如果 Subject 之一未完成，forkJoin 将不会发出响应。如果您希望在订阅之前修改来自 observable 的响应，您可以使用 RxJS map 和 catchError 运算符。您可能不需要catchError，因为您没有修改来自 HTTP 请求的错误。

试试下面的

服务

import { pipe } from 'rxjs';
import { map } from 'rxjs/operators';

$getRequestById(id: string): Observable<any> {
  return this.$http.get(API_URL + '/' + id).pipe(
    map(dto => new Request(dto))
  );
}

$getAllRequestsForId(id: string): Observable<any> {
  return this.$http.get(API_URL + '?request_id=' + id).pipe(
    map(response => response.map(item => new Request(i)))     // <-- inner `map` is array method, outer `map` is RxJS operator
  );
}

控制器

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests]).subscribe(
  (response) => this.setResponse(response),
  (error) => this.notification.error('Error occured while fetching request details! Please try again.')
);

更新：catchError

使用catchError 要记住的最重要的事情是返回一个可观察对象。您可以使用 RxJS throwError 发出错误并完成。或者你可以使用 RxJS of 返回一个 observable。

import { pipe, throwError, of } from 'rxjs';
import { map, catchError } from 'rxjs/operators';

// throwing error using `throwError`
$getRequestById(id: string): Observable<any> {
  return this.$http.get(API_URL + '/' + id).pipe(
    map(dto => new Request(dto)),
    catchError(error => throwError('Ran into an error: ', error))       
  );
}

// throwing error using `of`
$getRequestById(id: string): Observable<any> {
  return this.$http.get(API_URL + '/' + id).pipe(
    map(dto => new Request(dto)),
    catchError(error => of('Ran into an error: ', error))               
  );
}

Answer 2

你的代码应该是

$getRequestById(id: string): Observable<any> {
    return this.$http.get(API_URL + '/' + id).pipe(map(
      (dto) => new Request(dto)), // THIS IS GETTING EXECUTED
      (catchError) => throwError(err)
    );
  }

  $getAllRequestsForId(id: string): Observable<any> {
    return this.$http.get(API_URL + '?request_id=' + id).pipe(map(
      (dto) => dto.map(i => new Request(i))),
      (catchError) => throwError(error)
    );
  }

home.component.ts

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests])
 .subscribe(
    (response) => this.setResponse(response), // THIS LINE IS NOT GETTING CALLED.
    (error) => this.notification.error('Error occured while fetching request details! Please try again.')
 );

Answer 3

API 没有被调用，因为您已经使用 subscribe 使用了 API，因此请删除订阅并像这样重试

api.services.ts

$getRequestById(id: string) {
    return API_URL + '/' + id;
  }

  $getAllRequestsForId(id: string) {
     return API_URL + '?request_id=' + id;
  }

home.component.ts

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests])
 .subscribe(
    (response) => {
     console.log(response);
     this.setResponse(response)
    },
    (error) => this.notification.error('Error occured while fetching request details! Please try again.')
 );

Answer 4

forkjoin 只会在每个源 observable 完成后调用 subscribe 回调。在您的服务函数中，您返回一个永远不会完成的主题，因此 forkjoin 将永远等待。

此外，您无需在服务功能中创建新主题。直接返回 http.get 的结果。如果你需要在服务中转换结果，你可以通过一些 rxjs 操作符来管道化结果

api.services.ts

$getRequestById(id: string): Subject<any> {
    return this.$http.get(API_URL + '/' + id)
  }

$getAllRequestsForId(id: string): Subject<any> {
    return this.$http.get(API_URL + '?request_id=' + id).pipe(
      concatAll(),
      map(i => new Request(i)),
      toArray()
    );
  }