关于 2 个 rxjs 可观察对象和一个计时器的问题

Question

我有一个关于nestjs CQRS sagas 上下文中的rxjs 的问题。

假设我有一个场景，有两个事件一个接一个地发布。一个设置值，另一个取消设置。我需要能够监听将值设置为 3 秒的事件，并在其他事件尚未发布时执行一些操作。

这里是一些初学者的代码：

valuePersisted = (events$: Observable<any>): Observable<ICommand> => {
return events$
  .pipe(
    ofType(ValueHasBeenSet),
    map(event => {
      return new SomeCommand();
    }),
  );
}

我需要以某种方式收听ValueHasBeenUnset 事件并从流中取消，以防在一段时间内收到此事件。

编辑

我刚刚意识到事件ValueHasBeenSet 和ValueHasBeenUnset 可以设置和取消设置不同的值类型，代码应该区分这一点。例如，两个事件都有一个名为type 的属性，其值可以是'blue' | 'yellow'。有没有办法保留每个事件类型的逻辑，只保留两个通用事件 ValueHasBeenSet 和 ValueHasBeenUnset？

Answer 1

考虑通过以下方式实现它：

 return events$
  .pipe(
    ofType(ValueHasBeenSet), // <- listen for ValueHasBeenSet event
    switchMap(x => {  // <- use switchMap to return a timer 
      return timer(3000).pipe(
        takeUntil(events$.pipe(ofType(ValueHasBeenUnset))), // <- unsubscribe from the timer on ValueHasBeenUnset event
        map(event => {
          return new SomeCommand(); // <- map to SomeCommand once 3 seconds have passed
        })
      )
    })