当一个条件不被遵守时，如何停止RXJS链？

import { interval, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

// 创建一个 Subject 作为 notifier
const stopSignal$ = new Subject();

// 创建一个简单的 Observable，每秒发出一个递增的值
const source$ = interval(1000);

// 使用 takeUntil 操作符来停止 Observable 链
const final$ = source$.pipe(
  takeUntil(stopSignal$)
);

// 订阅最终的 Observable
const subscription = final$.subscribe(
  value => console.log(value),
  error => console.error(error),
  () => console.log('Observable 完成')
);

// 在特定条件不被遵守时发出 stopSignal
if (条件不被遵守) {
  stopSignal$.next();
}

// 当不再需要时，取消订阅
subscription.unsubscribe();