如何在@ngrx/effects中终止管道操作？

import { Injectable } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { of, Subject } from 'rxjs';
import { catchError, map, switchMap, takeUntil } from 'rxjs/operators';

@Injectable()
export class MyEffects {
  private cancel$ = new Subject<void>();

  myEffect$ = createEffect(() =>
    this.actions$.pipe(
      ofType('MY_ACTION'),
      switchMap(action =>
        // 在这里执行你的异步操作，比如发送HTTP请求
        // 当取消信号发出时，这个操作会被立即取消
        this.myService.someAsyncOperation().pipe(
          takeUntil(this.cancel$),
          map(response => ({ type: 'SUCCESS', payload: response })),
          catchError(error => of({ type: 'ERROR', payload: error }))
        )
      )
    )
  );

  constructor(private actions$: Actions, private myService: MyService) {}

  cancelMyEffect() {
    this.cancel$.next(); // 发出取消信号
  }
}