javascript - 检测异步系统中的结束事件

Question

我正在使用 RxJS 编写一些后端内容。当一切都完成后，我需要引发一些事件。

1) 架构涉及 Subject，因此 onCompleted 事件永远不会自然发生。它可以手动发出，但要做到这一点，我需要依赖一些 END 事件......循环循环。

2) 架构有 pending$ Observable 来跟踪待处理的任务。 不幸的是，由于系统的异步特性，该状态可以被清空多次，因此它本身的“空”不能用作 END 指示器。

这是我的解决方案，这是一个相当糟糕的解决方案，因为它需要显式的间隔常数。

import {Observable, Subject} from "rx";

let pendingS = new Subject();
let pending$ = pendingS
  .startWith(new Set())
  .scan((memo, [action, op]) => {
    if (action == "+") {
      memo.add(op);
      return memo;
    } else if (action == "-") {
      memo.delete(op);
      return memo;
    } else {
      return memo;
    }
  }).share();

let pendingDone$ = pending$ // a guess that
  .debounce(2000) // if nothing happens within 2 sec window
  .filter(pending => !pending.size); // and nothing is pending => we're done

pendingDone$.subscribe(function () {
  console.log("pendingDone!");
});

setTimeout(function () {
  pendingS.onNext(["+", "op1"]);
  pendingS.onNext(["+", "op2"]);
  pendingS.onNext(["+", "op3"]);
  pendingS.onNext(["-", "op2"]);
  pendingS.onNext(["-", "op1"]);
  pendingS.onNext(["-", "op3"]);
}, 100);

setTimeout(function () {
  pendingS.onNext(["+", "op4"]);
  pendingS.onNext(["-", "op4"]);
}, 500);

对于这个(非常普遍的)问题有更优雅的解决方案吗？

Answer 1

到目前为止，我已经找到了 3 种解决该主题的方法。

1)“空闲超时”

// next event
let pendingDone$ = pending$
  .debounce(2000)
  .filter(pending => !pending.size);

// next + completed events
let pendingDone$ = pending$
  .filter(pending => !pending.size)
  .timeout(2000, Observable.just("done"));

2)“同步设置，异步取消”

如果挂起的删除被挂起到下一个事件循环迭代，则下一个挂起的任务应该在此之前出现，从而删除“挂起为空”的时间条件。

pendingS.onNext(["+", "op"]);
// do something
setImmediate(() => 
  pendingS.onNext(["-", "op"]);
);

pendingDone$
  .filter(state => !state.size)
  .skip(1); // skip initial empty state

3) 调度程序 (?)

使用调度程序似乎可以实现类似的结果，但这可能是一种非常规用法，因此我不会深入研究这个方向。