Как вы отмените или отмените подписку, прежде чем переходить к следующей в Observables?

Question

Я очень новичок в Rxjs / Observables, и это действительно начинает беспокоить меня. Я использовал Observable, чтобы пройти через методы и убедиться, что они вызываются (используя функцию next ()) один за другим после завершения каждого из них. Итак, как вы отмените всю подписку после того, как, скажем, у второго из 5 следующих вызовов возникли проблемы с некоторыми данными? Ниже приведен упрощенный пример кода для понимания ситуации.

Я импортирую следующие данные:

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();
} );

И (ниже), используя метод подписки (), я запускаю подписку:

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

Хотя я не совсем понимаю, как вы используете потоки, один из способов отменить подписку состоит в том, чтобы сохранить субъект с логическим типом и использовать 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)
} )

Надеюсь, это поможет.