在 Observables 中继续下一步之前如何取消或取消订阅？

Question

我对 Rxjs/Observables 很陌生，它真的开始困扰我。我使用了一个 Observable 来遍历方法，并确保在每个方法完成后它们被一个一个地调用（使用 next() 函数）。那么，在可能说所有 5 个 next 呼叫中的第二个与某些数据有问题之后，您如何取消整个订阅？下面是一个简化的示例代码，供您了解情况。

我正在导入以下数据：

import { Observable } from 'rxjs';

在构造函数中，我准备了我的 Observable：

this.myObservable = new Observable( observer => {
  observer.next( this.funcOne(observer) );
  observer.next( this.funcTwo(observer) );
  observer.next( this.funcThr(observer) );
  observer.next( this.funcFor(observer) );
  observer.complete();
} );

并且（下）使用“subscribe()”方法，我开始订阅：

this.subscription = this.myObservable.subscribe( x => {
  if( x === -1 )
    this.subscription.unsubscribe(); 
    // I was thinking to unsubscribe here, but that doesn't work. 
    // I get a message saying: Cannot read property 'unsubscribe' of undefined
  else
    this.doSomethingElse(x)
} )

如何通过取消或退订订阅来完全停止一切？希望我的问题很清楚，感谢您抽出宝贵时间阅读我的挫败感。

Answer 1

虽然我不太了解您对流的使用，但取消订阅的一种方法是保留布尔类型的 Subject 并使用 takeUntil （或 pipe(takeUntil) 取决于您的 rxjs 版本）以及何时停止您可以对该主题执行 .next(true) 的订阅。这将结束订阅。 示例：

private destroy$ = new Subject();
...
this.subscription = this.myObservable.pipe(takeUntil(this.destroy$)).subscribe( x => {
  if( x === -1 )
    this.destroy$.next(true); 
    // I was thinking to unsubscribe here, but that doesn't work. 
    // I get a message saying: Cannot read property 'unsubscribe' of undefined
  else
    this.doSomethingElse(x)
} )

希望这会有所帮助。