rxjs 中的嵌套 HTTP 控制流

Question

试图围绕 Observables 和链接/嵌套 HTTP 请求。

假设我的遛狗 API 后端有以下 REST 端点，这些端点无法更改：

• GET /dogs（返回所有狗）：

  [
      { id: 1, name: 'Fido' },
      { id: 2, name: 'Barky' },
      { id: 3, name: 'Chip' },
      { id: 4, name: 'Bracken' }
  ]
  

• GET /walker/:id（返回一个遛狗者）：

  { id: 1, name: 'John Doe' }
  

• GET /pairings（返回狗和步行者之间的所有配对）：

  [
      { id: 1, dogIds: [2], walkerId: 1 },
      { id: 2, dogIds: [1, 3], walkerId: 2 }
  ]
  

业务规则

• 一对正好有 1 个步行者
• 配对包含 1 只或多只狗的列表
• 狗可以是 0 或 1 对的一部分
• 助行器可以是 0 或 1 对的一部分

目标

我想提供一个步行者和狗之间所有配对的列表，按步行者名称排序。我想按名称对每个步行者的狗进行分类。我不想展示任何没有主动配对的步行者或狗，例如：

Walker       | Dogs
-------------+-----------
John Doe     | Barky
Jan Kowalksi | Chip, Fido

我的思考过程

1. 同时请求所有 /pairings 和 /dogs
2. 等待这两个请求完成
3. 遍历每个pairing 并填充dogs 字段
4. 从每个pairing 中提取walkerId 并并行请求每个/walker/:id
5. 等待所有这些请求完成
6. 遍历每个 pairing 并填充 walker 字段

我觉得我可以使用 Promises 轻松做到这一点，但我正在努力让我的大脑适应 Observables 中的思考。这是我到目前为止所得到的（使用 Angular 的 HttpClient）：

function getDogWalkerPairings() {
    return Observable.forkJoin([
        this.http.get('/pairings'),
        this.http.get('/dogs')
    ])
        .map(
            (res) => {
                const pairings = res[0];
                const dogs = res[1];

                return pairings.map(p => {
                    const pDogs = p.dogIds.map(dogId =>
                        dogs.find(d => (d.id === dogId)
                    );
                    return Object.assign({ dogs: pDogs }, p);
                });
            }
        )
        .map((pairingsWithDogs) => {
            return Observable.forkJoin(
                pairingsWithDogs.map(p => this.http.get('/walkers/' + p.walkerId))
            );
        })
        .map((walkers) => {
            // uhhh... where to now?
            // I don't have a reference to pairings in this scope :/
        });
}

Answer 1

好的，我试试看:-)

我的方法是尽可能多地提取到函数中。对我来说，这有助于获得更好的画面。 我将它从“.map()”更改为“pipe(map())”，这是自 v5.5 以来的新 RxJs 样式。

function getDogWalkerPairings() {
    return Observable.forkJoin([
        this.http.get('/pairings'),
        this.http.get('/dogs')
    ]).pipe(
        map([pairings, dogs] => createPairingsWithDogs(pairings, dogs) ),
        switchMap( pairingsWithDogs => getWalkersForPairs(pairingsWithDogs) )
    )
}

function createPairingsWithDogs(pairing, dogs){
    return pairings.map(pairing => {
        const dogPairings = pairing.dogIds.map(
            dogId => dogs.find( dog => dog.id === dogId)
        );
        return Object.assign( {dogs: dogPairings }, pairing )
    }
}

function getWalkersForPairs(pairingsWithDogs):Observable<any>{
    return Observable.forkJoin(
        pairingsWithDogs.map(p => this.http.get('/walkers/' + p.walkerId))
    ).pipe(
        map( walkerArray => createWalkerDogPairs(walkerArray, pairingsWithDogs) )
    );
}

function createWalkerDogPairs(walkerArray, pairingsWithDogs){
    ...
    return finalResultTable;
}

它是如何工作的？ 首先，我像您一样创建配对。

然后改变流（switchMap）。在这里，我将您的技巧与 forkJoin 一起使用。但是，当我将它提取到它自己的函数中时，我创建了一个新的范围......在那里我拥有了我需要的一切。 （好吧，那里没有 cookie，所以不是所有的东西...... :-( )

如果这是我的编码，我还会添加很多类型。特别是当我切换类型（使用“地图”）时，这有助于我掌握它

Observable.of( [1,2,3,4] ).pipe(
 map( (numbers: number[]): boolean[] => checkOddNumbers(numbers) ),
 tap( (data: boolean[] => console.log(data) )
)

希望对你有所帮助。

热烈的问候

PS：我知道，我的方法名称很棒... :-(

Answer 2

使用concatMap 链接最终的http 调用。棘手的部分是您需要将您为第一个forkJoin 获得的配对传回。这是我的答案，也是一个有效的example on stackblitz。我以 500 毫秒的延迟模拟了您的 http 调用，只需在我使用它们的地方使用您适当的 http 调用即可。

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Observable, of, forkJoin, merge, Subject } from 'rxjs';
import { map, delay, concatMap, takeUntil } from 'rxjs/operators';

interface IdNamePair {
  id: number;
  name: string;
}

interface Pairing {
  id: number;
  dogIds: Array<number>;
  walkerId: number;
  dogs?: Array<IdNamePair>;
  walker?: IdNamePair;
}

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit, OnDestroy {

  pairings: Array<Pairing>;
  private delay = 500;
  private ngUnsubscribe: Subject<any> = new Subject();

  constructor() { }

  ngOnInit() {
    this.getDogWalkerPairings();
  }

  ngOnDestroy() {
    this.ngUnsubscribe.next();
    this.ngUnsubscribe.complete();
  }

  private getDogWalkerPairings() {
    forkJoin(this.getPairings(), this.getDogs())
      .pipe(
        map(this.mapToPairingsWithDogs),
        concatMap((pairingsWithDogs: Array<Pairing>) => {
          return forkJoin(pairingsWithDogs.map(pair => {
            return forkJoin(this.getWalker(pair.walkerId), of(pair));
          }));
        }),
        map(this.mapToPairingsWithDogsAndWalker),
        takeUntil(this.ngUnsubscribe)
      )
      .subscribe((pairings: Array<Pairing>) => {
        console.log(pairings);
        this.pairings = pairings;
      });
  }

  private mapToPairingsWithDogs(data: [Array<Pairing>, Array<IdNamePair>]): Array<Pairing> {
    const pairings = data[0];
    const dogs = data[1];
    return pairings.map(pairing => {
      const pDogs = pairing.dogIds.map(dogId => dogs.find(d => (d.id === dogId)));
      pairing.dogs = pDogs;
      return pairing;
    });
  }

  private mapToPairingsWithDogsAndWalker(data: Array<[IdNamePair, Pairing]>): Array<Pairing> {
    return data.map(d => {
      const pairing: Pairing = d[1];
      pairing.walker = d[0];
      return pairing;
    });
  }

  private getDogs(): Observable<Array<IdNamePair>> {
    return of([
      { id: 1, name: 'Fido' },
      { id: 2, name: 'Barky' },
      { id: 3, name: 'Chip' },
      { id: 4, name: 'Bracken' }
    ]).pipe(delay(this.delay));
  }

  private getWalker(id: number): Observable<IdNamePair> {
    return of({ id: id, name: id === 1 ? 'John Doe' : 'Jane Doe'}).pipe(delay(this.delay));
  }

  private getPairings(): Observable<Array<Pairing>> {
    return of([
      { id: 1, dogIds: [2], walkerId: 1 },
      { id: 2, dogIds: [1, 3], walkerId: 2 }
    ]).pipe(delay(this.delay));
  }

}

编辑

解释：

1. forkJoin - 将同时返回 pairings 和 dogs
2. map - 将 dogs 与 pairings 配对，我们在其中找到它们的 id
3. concatMap - 将执行下一个调用，为此我们需要做一些事情
   • 我们需要为每个配对调用getWalker，并且我们需要同时获得它们的所有结果，因此我们将每个配对映射到@987654329 @方法调用返回一个Observable，最后我们forkJoin映射到Observable数组
   • 棘手的部分是我们需要将仅在concatMap 范围内可用的配对 传递给下一个map 或subscribe，为此我们需要@987654336 @ 每个Observable 我们从getWalker 调用中获得一个Observable，该Observable 是从单个配对创建的。
4. map - 将 walker 与 pairings 配对，我们在其中找到他们的 id