带过滤器的 takeUntil

Question

我在 NestJS 中有一些带有 mergeMap 运算符的代码，我需要根据来自外部和内部可观察对象的一些值来停止可观察对象流。考虑这个例子：

type SomeEvent1 {
  value: string;
}

type SomeEvent2 {
  value: string;
}

@Injectable()
export class TradeSagas {
  private readonly logger = new MyLogger(TradeSagas.name);
  @Saga()
  someSaga = (events$: Observable<IEvent>): Observable<ICommand> => {
    return events$.pipe(
      ofType(SomeEvent1),
      mergeMap((event: SomeEvent1) => {
        events$.pipe(
          ofType(SomeEvent2),
          map((event) => new SomeEvent3())
        );
      })
    );
  };

我需要能够取消订阅收听SomeEvent1 的流或收听SomeEvent2 的流（尚不确定我的特定用例需要什么），例如 SomeEvent1.value === SomeEvent2.value.

我知道我可以使用 takeUntil 运算符，但它接受另一个 observable，所以我很困惑如何使它工作。

我想取消订阅的原因是因为 mergeMap 会跟踪所有以前的事件值，有时它会弄乱流程。我知道我可以使用 switchMap 但这不是我特定问题的解决方案，因为我不需要仅仅因为新事件到达而取消外部 observable - 我有一些更复杂的逻辑来构建事件流。 如果对如何处理此问题有其他想法，请告诉我。

感谢您的帮助。

Answer 1

对于您的情况，takeWhile 会是一个更好的选择，因为它需要一个函数而不是一个 observable：

  someSaga = (events$: Observable<IEvent>): Observable<ICommand> => {
    return events$.pipe(
      ofType(SomeEvent1),
      mergeMap(event1 => events$.pipe(
          ofType(SomeEvent2),
          takeWhile(event2 => event2.value === event1.value),
          map(event2 => new SomeEvent3())
        );
      })
    );
  };

编辑：

有时，如果您将来源分解为单独的变量，则更容易理解。所以我认为这可能是您正在寻找的：

const event1$ = events$.pipe(ofType(SomeEvent1));
const event2$ = events$.pipe(ofType(SomeEvent2));
const event3$ = events$.pipe(ofType(SomeEvent3));

const someSaga2 = events1$.pipe(
  mergeMap(event1 => events2$.pipe(
    takeUntil(events3$.pipe(
        filter(event3 => event3.value === event1.value)
    ))
  ))
);

行为描述：当Event1 发出时，发出Event2，直到发出与value 具有相同value 的Event3。