rxjs：在已经显示第一个异步管道的同时组合可观察结果的结果

Question

在 angular 中使用 rxjs 显示第一个可观察对象的结果并在其他可观察对象完成时合并数据的最佳方法是什么？

示例：

@Component({
    selector: 'app-component',
    template: `
<div *ngFor="let group of groups$ | async">
    <div *ngFor="let thisItem of group.theseItems">
        ...
    </div>
    <div *ngFor="let thatItem of group.thoseItems">
        ...
    </div>
</div>
`
})
export class AppComponent implements OnInit {
    ...
    ngOnInit() {
        this.groups$ = this.http.get<IThisItem[]>('api/theseItems').pipe(
            map(theseItems => {
                return theseItems.groupBy('groupCode');
            })
        );

        // combine these results? This operation can take 5 seconds
        this.groups$$ = this.http.get<IThatItem[]>('api/thoseItems').pipe(
            map(thoseItems => {
                return thoseItems.groupBy('groupCode');
            })
        );
    }
}

我知道可以通过订阅两者来完成，然后合并结果。但是是否可以为此使用管道运算符，并使用 async 管道？

Answer 1

我认为您可以使用 combineLatest rxjs 运算符。这可能意味着您也稍微更改了模板中的处理方式。

我无法使用你的示例，因为我不知道你的 get 函数，但基本上适用相同的原则。

Check stackblitz here 例如：

export class AppComponent  {

  private firstObservable = of([{name: 'name1'}]).pipe(startWith([]));
  private secondObservable = of([{name: 'name2'}]).pipe(startWith([]));

  combined = combineLatest(this.firstObservable, this.secondObservable).pipe(
        map(([firstResult, secondResult]) => {
          return [].concat(firstResult).concat(secondResult)
        }) 
   );
}

Html 输出：

<span *ngFor="let item of combined | async">{{item.name}}<span>

Answer 2

Async pipe 只是一个 observable 的订阅者...要回答您的问题，您可以使用任何可能的方式...例如：

<div *ngFor="let group of groups$ | async as groups">
    <div *ngFor="let thisItem of group.theseItems">
        ...
    </div>
</div>

public groups$: Observable<type> = this.http.get<IThatItem[]>.pipe(
  startWith(INITIAL_VALUE)
);

或

public groups$: Observable<type> = combineLatest(
  of(INITIAL_VALUE),
  this.http.get<IThatItem[]>
)

Answer 3

您可以使用合并和扫描。

  first$: Observable<Post[]> = this.http.get<Post[]>('https://jsonplaceholder.typicode.com/posts?userId=1');
  second$: Observable<Post[]>  = this.http.get<Post[]>('https://jsonplaceholder.typicode.com/posts?userId=2');
  combinedPosts$: Observable<Post[]> = merge(this.first$, this.second$).pipe(
    scan((acc: Post[], curr: Post[]) => [...acc, ...curr], [])
  )

https://www.learnrxjs.io/operators/combination/merge.html 使一个可以从多个中观察到。

https://www.learnrxjs.io/operators/transformation/scan.html scan类似于array.reduce...可以累加每次可观察发射的结果。

工作示例： https://stackblitz.com/edit/angular-lrwwxw

combineLatest 运算符不太理想，因为它要求每个可观察对象在组合可观察对象发出之前发出：https://www.learnrxjs.io/operators/combination/combinelatest.html

Be aware that combineLatest will not emit an initial value until each observable emits at least one value.

rxjs：在已经显示第一个异步管道的同时组合可观察结果的结果

rxjs: Combine result of observables while already displaying the first with async pipe

rxjs

angular

rxjs-observables