使用第二个 API 的可观察对象的问题

Question

好吧，我对 RxJs 真的很不好，我正面临一个问题。

在我的一个 Angular 组件中，我加载如下数据：

this._obs$.push(this.store1.loadTypes());
this._obs$.push(this.store2.loadLimitQuantities());
// [...]
    
forkJoin(this._obs$).subscribe(
  next => console.log(next),
  error => console.log(error)
);

其中一个 observables 从 API 1 接收数据，然后需要从 API 1 接收的数据中从 API 2 检索数据：

loadLimitQuantities(): Observable<LimitQuantity[]> {
    // API 1
    const obs = this.limitQuantityService.getLimitQuantities().pipe(share());
    obs.subscribe(data => {
        if (data) {
            const ifs = [];
            data.forEach(element => {
                // API 2
                ifs.push(this.technicalAttribTextIFSService.getType(element.ifsType).pipe(take(1), tap((type) => { 
                    type ? element.type = type.valueText : element.type = null;
                })));
            });
            forkJoin(ifs).subscribe(end => {
                this.limitQuantitiesSubject.next(data);
            });
        } else {
            console.error("Erreur lors du chargement des quantités limites");
        }
    });
    // AP1
    return obs;
}

不幸的是，我的加载数据的 Angular 组件的 next 接收得太早了，因为 API 1 已完成返回数据，但 API 2 未完成。

怎么办？ return 应该是 API 2 的 forkJoin。

StackBlitz : here

感谢您的帮助。

Answer 1

我不确定我是否 100% 解决了您的问题 - 但我认为这会满足您的需求。

limitedQuantitiesWithTypeFromApi2$ = this.loadLimitQuantities()
    .pipe(
        switchMap(
          x => forkJoin(x.map(y => this.getType(y.ifsType))), 
        ),
      );

  ngOnInit() {
    forkJoin(this.loadTypes(), this.limitedQuantitiesWithTypeFromApi2$)
      .subscribe(([types, lq]) => console.log('Original 1', types, lq));

    forkJoin(this.loadTypes(), this.loadLimitQuantities())
      .pipe(
        map(([types, lq]) => {
          return lq.map(x => ({
            ...x, 
            type: types.find(y => y.ifsType === x.ifsType)
          }))
        })
      )
      .subscribe(x => console.log('Alternate Approach', x));

  }

  loadTypes() {
    return timer(1000).pipe(
      map(x => ([
        { ifsType: 1, type: 'vText1'}, 
        { ifsType: 2, type: 'vText2'}
      ])),
      share()
    );
  }

  loadLimitQuantities() {
    return timer(1000).pipe(
      map(x => ([
        {ifsType: 1, type: null, }, 
        {ifsType: 2, type: null }
      ])),
      share(),
    );
  }

  getType(ifsType: number) {
    return timer(1000).pipe(
      map(x => ifsType === 1 ? 'vText1' : 'vText2'),
      map(x => ({ ifsType, type: x })),
      share()
    );
  }

但是，我认为假设您的 API2 是一个 HTTP 调用是安全的 - 虽然它可以工作，但我建议您只批量获取所有“类型关联”，这样您就可以制作一个更简单的流。

我的意思是如果loadTypes 已经有一个ifsType 到type 的关联，那么你可以只订阅“方法2”——如果你仍然使用角度，你可以只使用async 管道订阅loadTypes和loadLimitedQuantities。

Answer 2

也许你需要这样的东西？

class MyClass {

    loadLimitQuantities(): Observable < LimitQuantity[] > {
        // API 1
        return this.limitQuantityService.getLimitQuantities().pipe(
            map((data) => this.getTypeRequests(data)),
            switchMap((requests) => forkJoin(requests))
        );
    }

    getTypeRequests(data) {
      if(!data) {
          console.error("Erreur lors du chargement des quantités limites");
          return [];
      }

      return data.map(element => this.getType(element));
    }

    getType(element) {
        return this.technicalAttribTextIFSService.getType(element.ifsType)
            .pipe(map((type) => ({...element, type: type ? type.valueText : null})))
    }
}

您获得限制数量，将它们映射到类型请求，然后将它们连接在一起。