javascript - 如果 source1 超时，RxJS 从 source2 发出，当 source1 返回时从 source1 发出

Question

我有一个发出值的 Observable 源 source1，如果它没有发出任何东西超过 2 秒，我想切换到后备源 source2。如果 source1 再次发射，我想从中发射。依此类推，无限期。

到目前为止，我有以下内容

import { timeout, catchError, takeUntil, concat } from 'rxjs/operators';

declare const source1: Observable;
declare const source2: Observable;

source1.pipe(
    timeout(2000),
    catchError(() => {
      return source2.pipe(
        takeUntil(source1)
      );
    }),
    concat(source1)
).subscribe(val => console.log(val));

这几乎行得通。如果 source1 2 秒后没有发出，它从 source2 发出，直到 source1 再次发出，然后切换到 source1 .但是有两个主要缺陷:

1. 当 source1 再次发出时，第一个发出的值被 takeUntil“捕获”(source1 是一个热可观察对象)并且不会在 concat(source1)
中
2. 如果 source1 第二次停止发射，我希望有相同的行为。在我的实现中，它只工作一次。

知道如何解决这个问题吗？

Answer 1

我猜你可以通过共享 source1 然后使用 repeat 重新订阅同一条链来做到这一点(我没有测试过):

const shared1 = source1.pipe(share());

source1.pipe(
  timeout(2000),
  catchError(() => merge(source1, source2).pipe(
    takeUntil(source1),
  )),
  repeat(),
).subscribe(val => console.log(val));