使用测试调度程序时如何链接 RxJS 承诺？答案

【问题标题】：How can I chain RxJS promises when using the test scheduler?使用测试调度程序时如何链接 RxJS 承诺？
【发布时间】：2017-06-16 07:46:46
【问题描述】：

在使用带有延迟交付的 Observables 的代码中，我可以使用 TestScheduler 使交付在虚拟时间内发生而没有实际延迟。但是，如果我将这些 Observables 转换为 Promise 并尝试将它们链接在一起，似乎只有第一个 Promise 得到解决。如何在不等待实时解决方案出现的情况下测试使用这种延迟的系统？

请参阅下面的片段：请注意，将 'sched' 更改为 'async' 并注释掉 'flush' 行会导致它按预期运行，但速度很慢。我怎样才能让这个承诺链顺利运行？

var sched = new Rx.TestScheduler();
//var sched = Rx.Scheduler.async;

Rx.Observable.timer (100, sched).toPromise ().then (
  () => $("#test").text("100")).then (
  () => Rx.Observable.timer (100, sched).toPromise ()).then (
  () => $("#test").text("200")).then (
  () => Rx.Observable.timer (100, sched).toPromise ()).then (
  () => $("#test").text("300")).then (
  () => Rx.Observable.timer (100, sched).toPromise ()).then (
  () => $("#test").text("400"));

sched.flush ();

<script src="https://ajax.googleapis.com/ajax/libs/jquery/2.1.1/jquery.min.js"></script>
<script src="https://unpkg.com/@reactivex/rxjs@5.4.0/dist/global/Rx.min.js"></script>

<div id="test"></div>

【问题讨论】：

标签： javascript promise rxjs

【解决方案1】：

问题在于，在解决 Promise 之后，直到控制流返回到顶层之后才会调用它的回调。这意味着在调用sched.flush() 期间解决第一个promise 后，一个新的回调将排队，但直到sched.flush() 完成后才会发生。但是我们仍然需要sched.flush() 再次运行以使第二个 Observable 完成，依此类推。

原则上，这意味着我们只需要异步重新调用sched.flush() 几次，就可以完成所有的 observables。我们可以通过创建一个单独的 Promises 链来完成此操作，该链在每个回调上刷新调度程序。唯一的问题是尚不清楚如何判断 Observables 是否已准备好，因为对于不相关的 Promise 回调没有明确定义的执行顺序。

我的解决方案是简单地创建一个 Promise 链，该链执行一个刷新调度程序的回调函数，然后递归地重新插入自身，直到它发现它不再需要这样做。这依赖于产生某种可以测试的副作用的原始回调（但由于要求是单元测试，这似乎是合理的——如果没有这样的效果，它在第一种情况下是不可测试的)。在实际代码中，它还需要在检测到代码结束后执行任何断言、测试拆卸等。我还对其进行了安排，使其具有有限的迭代次数，因此如果测试代码以某种方式失败，它不会无限期地递归。

var sched = new Rx.TestScheduler();
var finished = false;
//var sched = Rx.Scheduler.async;

Rx.Observable.timer (100, sched).toPromise ().then (
  () => $("#test").text("100")).then (
  () => Rx.Observable.timer (100, sched).toPromise ()).then (
  () => $("#test").text("200")).then (
  () => Rx.Observable.timer (100, sched).toPromise ()).then (
  () => $("#test").text("300")).then (
  () => Rx.Observable.timer (100, sched).toPromise ()).then (
  () => $("#test").text("400")).then (
  () => finished = true );

let f = n => { 
   sched.flush (); 
   if (n > 0 && !finished) {
       return Promise.resolve(n - 1).then(f); 
   } 
   else {
       $("#test").append (" [" + n + " iterations left]");
   }
}
Promise.resolve(100).then(f);

<script src="https://ajax.googleapis.com/ajax/libs/jquery/2.1.1/jquery.min.js"></script>
<script src="https://unpkg.com/@reactivex/rxjs@5.4.0/dist/global/Rx.min.js"></script>

<div id="test"></div>

【讨论】：