如何处理 RxJs 运营商的依赖 http 服务调用订阅？

Question

我有一个名为 fetchServiceProviders 的函数，它将返回一个服务提供者列表，并让我通过调用另一个服务来设置服务提供者的徽标。我通过使用两个订阅实现了这一点。

第一次订阅时，我得到了列表。然后我映射它并通过所有服务提供商的 id 调用 fetchServiceProviderLogo。

这可以通过一次订阅来完成吗？我的方法好吗？如果不是，问题是什么？一些 rxjs 运营商如何做到这一点？

我尝试了 map 和 flatMap 运算符，但没有得到任何线索。

fetchServiceProviders ()
{
    this.subscribers.fetchServiceProvidersLogoSub = this.serviceProviderService.fetchServiceProviders( new Map() ).subscribe(
        data =>
        {
            this.providers = data._embedded.serviceProviders;
            this.providers.map(
                provider =>
                {  
                    this.subscribers.fetchServiceProvidersLogoSub =
                        this.serviceProviderService.fetchServiceProviderLogo( { 'id': provider.id } ).subscribe(
                            logo =>
                            {
                                this.createIndividualLogoFromBlob( provider, logo );
                            }
                        );
                }
            );
        }
    )
}

Answer 1

如果您的所有 Observables 都已完成并且 this.serviceProviderService.fetchServiceProviderLogo 中的调用顺序无关紧要，您可以使用 switchMap（或 mergeMap）和 merge 的组合。如果您的提取请求全部完成，您也不必取消订阅。

fetchServiceProviders () {
  this.serviceProviderService.fetchServiceProviders(new Map())
    .pipe(
      map(data => data._embedded.serviceProviders),
      mergeMap(providers => merge(
        providers.map(provider =>
          this.serviceProviderService.fetchServiceProviderLogo({ 'id': provider.id })
           .pipe(map(logo => ({ provider, logo })))
        )
      ))
    )
    .subscribe(({ provider, logo }) => this.createIndividualLogoFromBlob(provider, logo));
}

您可以使用forkJoin 代替merge 来并行执行this.serviceProviderService.fetchServiceProviderLogo 调用。

fetchServiceProviders () {
  this.serviceProviderService.fetchServiceProviders(new Map())
    .pipe(
      map(data => data._embedded.serviceProviders),
      mergeMap(providers => forkJoin(
        providers.map(provider =>
          this.serviceProviderService.fetchServiceProviderLogo({ 'id': provider.id })
           .pipe(map(logo => ({ provider, logo })))
        )
      ))
    )
    .subscribe(providerLogos =>
       providerLogos.forEach(({ provider, logo }) => this.createIndividualLogoFromBlob(provider, logo))
    );
}

Answer 2

这是一个简单的例子。您可以考虑 getItems 和 getStreetNameById 您的提供商。

import { of, from } from 'rxjs';
import { switchMap, groupBy, map, mergeMap } from 'rxjs/operators';

function getItems() {
  return of([
    { streetId: 1, person: 'Person 1' },
    { streetId: 2, person: 'Person 2' },
    { streetId: 1, person: 'Person 3' },
  ]);
}

function getStreetNameById(id: number) {
  console.log('requesting street name for ', id);
  return of('Street ' + id);
}

getItems().pipe( //get all the items
  switchMap(items => from(items)), //create a new stream with each item from the list
  groupBy(item => item.streetId), //group by the streetId (we want to request the street name once per id
  mergeMap(group => { //merge all the transformed groups back into one stream
    return getStreetNameById(group.key) //request the street name by id
      .pipe(
        switchMap(streetName => group.pipe( //merge the street name back to all the items in the group
          map(item => ({ ...item, streetName })),
        ))
      );
  })
).subscribe(items => {
  console.log(items); //now all the items have the street name
});

https://stackblitz.com/edit/rxjs-merge-providers?file=index.ts

Answer 3

在您的代码中，您正在为每个 serviceProvider 进行订阅。 如果您想一次完成所有操作，可以查看合并运算符：https://www.learnrxjs.io/operators/combination/merge.html