javascript - 在 RxJS 中是否可以在 auditTime() 操作中测试未决值？

Question

我正在使用 RxJS 的 .auditTime(500) 操作 ( docs ) 作为尾随 throttle :我想最多每 500 毫秒发出一次服务器调用。

服务器调用完成后的下游，我需要查询是否还有更多待处理的服务器调用，或者缓冲区现在是否已清除，以便我可以以状态消息的形式将此信息传达给用户，例如“正在保存...”和“已保存”。

大致如下所示。

saveToServerObservable
  .do(() => {
    // gets called every time
    setStatus(Status.SAVING);
  })
  .auditTime(500) // wait 500 ms and emit no more than once per 500 ms
  .flatMap(data => axios({
    method: "post",
    url: "/saveurl",
    data: data,
  }))
  .map(response => response.data)
  .do(data => {
    // here I want to know whether there are pending values from the
    // auditTime() operation above or if the buffer is currently clear
    const pendingSaves = ???;
    if (!pendingSaves) {
     setStatus(Status.SAVED);
    }
  })
  .subscribe();

正如您在最后的 .do() 操作中看到的，我想知道是否有来自 .auditTime(500) 操作的未决值。我怎样才能实现这样的目标？

干杯! 🙏

Answer 1

我认为您可以使用 scan 并通过稍微修改您的链来实现您想要的:

const inc = new Subject();
const dec = new Subject();

const counter = Observable.merge(dec.mapTo(-1), inc.throttleTime(500).mapTo(1))
    .scan((acc, val) => acc + val, 0)
    .map(val => val > 0);

saveToServerObservable
  .do(() => {
    // gets called every time
    setStatus(Status.SAVING);
    inc.next();
  })
  .auditTime(500) // wait 500 ms and emit no more than once per 500 ms
  .flatMap(data => axios({
    method: "post",
    url: "/saveurl",
    data: data,
  }))
  .do(() => dec.next())
  .map(response => response.data)
  .withLatestFrom(counter, (data, pendingSaves) => {
    if (!pendingSaves) {
     setStatus(Status.SAVED);
    }
  })
  .subscribe();

整个想法在 counter Observable 中合并了 inc 和 dec。这两个 Observables 使用 scan() 递增和递减计数器。

inc 还与 .throttleTime(500) 链接，与 .auditTime(500) 正好相反，因为当您调用setStatus(Status.SAVING); 你总是知道这会让 .auditTime(500) 发出一个项目，因此你可以立即增加计数器。

然后 withLatestFrom 只是将计数器与远程调用的结果合并，这是您可以检查 counter 的最新发射的地方。