RxJS 可观察对象，如果值发生变化或经过特定时间则触发

Question

我有一个每秒触发多次的流。现在我想要一个观察者，如果值发生变化或者在最后一个过程之后经过了某个时间，那么它会进行过程。我想用纯 RxJS 来实现这一点，而不需要额外的“辅助变量”。 场景：

const fooBar$ = timer(0, 100)                         // every 0.1 seconds
.pipe(map(()=> Math.random() > 0.5 ? "foo" : "bar"))  // create randomly "foo" or "bar
.pipe(take(12))                                       // do it 12 times
.pipe(distinctUntilChanged())                         // just fires if the value changes
.subscribe(console.log); 

到目前为止一切顺利。但另外，如果某个时间过去了，我想捕获一个值：

// ⬇
"foo" // ❌
"foo" // ✅
"bar" // ✅
"foo" // ❌
"foo" // ❌
"foo" // ❌
"foo" // ❌
"foo" // ✅ <-- I want this one too, because a certain time (0.5 seconds) went by
"foo" // ❌
"foo" // ✅
"bar" // ✅
"foo" // ✅

我认为它可以与 debounce() 一起使用。但我可能是错的。

Answer 1

const fooBar$ = timer(0, 100).pipe(
  map(() => Math.random() > 0.5 ? "foo" : "bar"),
);

const distinct$ = fooBar$.pipe(
  distinctUntilChanged(),
);

const timer$ = distinct$.pipe(
  mergeMap(value => timer(500).pipe(
    map(() => value + " from timer"))),
);

const result$ = merge(distinct$, timer$).pipe(
  startWith('foo'), 
  take(12)
);

result$.subscribe(console.log);