组合多个流/观测值以使用映射运算符

Question

我有两台服务器，具有独立的流/可观察对象，如何使这两台服务器使用相同的链，以便我可以使用switchMap，flatMap等操作符控制它们？

if (serverNumber == 1) {
  this.streamService.getStream('test 1')
    .pipe(
      map(image => console.log(image))
    ).subscribe();
}
else  {
  this.streamService.getFastStream()
    .pipe(
      map(image => console.log(image))
    ).subscribe();
}

例如，如果我从服务器获取，而getStream需要一段时间，并且在getFastStream完成后仍在进行中，我希望通过切换映射阻止第一个getStream的覆盖。我怎么能这么做？我是rxjs的新手，所以我尝试的可能不是它应该如何使用。

可观察的

  images$: Observable<string> = new Observable<string>();

然后，我必须将流推到图像$观察与运算符之一？

this.images$.pipe(
  switchMap(this.streamService.getStream(1))
)

模板：

<div class="section">
    Click to send a request ...
</div>
<div class="section">
    <button (click)="click(1)">Server 1</button>
</div>
<div class="section">
    <button (click)="click(5)">Server 2 (Longer)</button>
</div>

Answer 1

RXJSraceobserveable creator可能会有所帮助。

var server1$ = this.streamService.getStream('test 1');
var fastServer$ = this.streamService.getFastStream();

var fastest$ = race(server1$, fastServer$)
              .pipe(map(image=> console.log(image)));
fastest$.subscribe();

Answer 2

首先在fast observable上使用takeUntil操作符。其次，使用merge获取快速可观察结果，直到缓慢可观察结果开始出现。

import { merge } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

const slow$ = this.streamService.getStream('test 1').pipe(
  map(image => console.log(image))
);

const fast$ = this.streamService.getFastStream().pipe(
  map(image => console.log(image)),
  takeUntil(slow$)
);

merge(
  fast$,
  slow$
).subscribe()

这里是一个StackBlitz演示演示。