停止 Rxjs 观察者的下一次执行？

Question

我正在尝试使用 Rxjs 观察器的基本用例。这就是我正在做的事情：

const { Observable } = require('rxjs');

Observable.create(o => { 
  setTimeout(() => o.next('hi'), 1000); 
  setTimeout(() => { throw new Error('A mistake') }, 1500); 
  setTimeout(() => o.next('this should not print'), 2000) }
).subscribe({
  next: x => console.log(x),
  error: y => console.log('error', y),
  complete: () => console.log('done')
});

输出结果如下：

> hi
Error: A mistake
    at Timeout.setTimeout [as _onTimeout] (repl:1:89)
    at ontimeout (timers.js:498:11)
    at tryOnTimeout (timers.js:323:5)
    at Timer.listOnTimeout (timers.js:290:5)
> this should not print

我需要做的是停止下一次执行，即使我 unsubscribe 我的订阅也不会停止下一次执行。

我也这样试过：

让订阅；

let source = Observable.create(o => { 
  try {
    setTimeout(() => o.next('hi'), 1000); 
    setTimeout(() => { throw new Error('A mistake') }, 1500); 
    setTimeout(() => o.next('this should not print'), 2000) 
  } catch (e) {
    subscription.unsubscribe();
    return 'error';
  }
});
subscription = source.subscribe({
  next: x => console.log(x),
  error: y => console.log('error', y),
  complete: () => console.log('done')
});

但没有机会……它没有停止。

我所拥有的代码不只是设置超时我有这样的异步和等待代码：

let subscription;

let source = Observable.create(async o => { 
  try {
    o.next(await anEvent()); 
    o.next(await anEventThatThrowsAnException()); 
    o.next(await anEventThatIWantToAvoidDueToTheException()); 
  } catch (e) {
    subscription.unsubscribe();
    return 'error';
  }
});
subscription = source.subscribe({
  next: x => console.log(x),
  error: y => console.log('error', y),
  complete: () => console.log('done')
});

如何实现此代码以停止 "this should not print"？

Answer 1

用setTimeout设置定时器后，你不会用unsubscribe阻止它的执行。

您将需要手动清除超时。

请注意，您可以将超时保存在变量中 var myTimeout = setTimeout(f,ms); 然后您可以取消 clearTimeout(myTimeout);

阅读更多：

Answer 2

这是使它与 ABOS 所述的 observer.error 一起工作的方法。事实上，这是他的例子并且有效。

因此，出于记录目的，我用他在评论中的回答回答了这个问题。

ABOS answer

我只是想，如果有人在尝试处理错误时遇到问题，请抛出错误，而不是按需要的方式捕获和执行 observer.error。

停止 Rxjs 观察者的下一次执行？

Stop next execution of Rxjs observer?

javascript

node.js

rxjs

rxjs6