RX:同步承诺

Question

假设我有一个相当典型的rx用法，每次出现一些更改事件时都会发出请求(我以.NET风格编写，但我真的在考虑Javascript(：

myChanges
    .Throttle(200)
    .Select(async data => {
        await someLongRunningWriteRequest(data);
    })

如果请求花费的时间超过 200 毫秒，则有可能在旧请求完成之前开始新请求 - 甚至可能首先完成新请求。

如何同步？

请注意，这与多线程无关，这是我在谷歌搜索"rx同步"或类似内容时唯一能找到的信息。

Answer 1

您可以使用 concatMap 运算符，该运算符只有在上一项完成后才会开始处理下一项。

下面是一个示例，其中events$以 200 毫秒的间隔出现，然后以不同的持续时间连续处理：

const { Observable } = Rx;
const fakeWriteRequest = data => {
  console.log('started working on: ', data);
  return Observable.of(data).delay(Math.random() * 2000);
}
const events$ = Observable.interval(200);
events$.take(10)
  .concatMap(i => fakeWriteRequest(i))
  .subscribe(e => console.log(e));

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>