rxjs：结合可观察的结果，同时已经用异步管道显示第一个答案

【问题标题】：rxjs: Combine result of observables while already displaying the first with async piperxjs：结合可观察的结果，同时已经用异步管道显示第一个
【发布时间】：2019-12-02 07:37:54
【问题描述】：

在 rxjs 中使用 Angular 来显示第一个 observable 的结果并在其他 observable 完成时组合数据的最佳方式是什么？

例子：

@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 管道？

【问题讨论】：

为了清楚起见，你想用这些项目获得一次发射，然后你想用这些项目与那些项目组合一次发射？
是的，但理论上这可能是相反的； thoseItems 被发射之前 theseItems 被发射

标签： angular rxjs rxjs-observables

【解决方案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>

【讨论】：

这个方法不是要等两个 observables 都发射后才显示成组合吗？
不，那是 forkJoin 操作符（它等待 observable 完成）。每当某些可观察对象发出一个值时，组合就会发出。当然，您应该始终检查是否有一些要连接的值。
@devqon 关于两个 Observable 必须在组合发射之前发射是正确的，但这可以通过简单的 startsWith 来解决。取自rxjs docs：请注意，在每个 observable 发出至少一个值之前，combineLatest 不会发出初始值。
@AthanasiosKataras 这似乎不起作用，请参阅this forked stackblitz。只有在两个 observables 都被解析后才会显示这些值
那是因为您将延迟应用于第一个值。检查更新的 stackblitz 并正确应用延迟。

【解决方案2】：

您可以使用合并和扫描。

  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 扫描类似于array.reduce...您可以累积每个可观察发射的结果。

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

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

请注意，在每个 observable 发出至少一个值之前，combineLatest 不会发出初始值。

【讨论】：

与其他答案相反，这似乎正是我想要的，谢谢:)

【解决方案3】：

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[]>
)

【讨论】：