序列化可观察的下一个函数调用

Question

考虑以下 Typescript / rxjs 代码：

import { range } from 'rxjs'; 

range(1, 5).subscribe(async (num) => {
  await longLastingOperation(num);
})

function longLastingOperation(num) {
  console.log('starting', num);

  return new Promise((resolve) => {
    setTimeout(() => {
      console.log('done with', num);
      resolve();
    }, Math.random() * 1000);
  });
}

每个发出的值都会触发一个随机持续时间的持久操作。控制台输出是不可预测的，看起来类似于：

starting 1
starting 2
starting 3
starting 4
starting 5
done with 2
done with 5
done with 1
done with 4
done with 3

现在，我想为每个发出的值“序列化”执行持久操作。例如，我希望 longLastingOperation(2) 在开始之前等待 longLastingOperation(1) 完成。

我希望每次都得到一个看起来完全一样的输出：

starting 1
done with 1
starting 2
done with 2
starting 3
done with 3
starting 4
done with 4
starting 5
done with 5

如何使用 rxjs 和 observables 实现这一目标？

Answer 1

考虑在每个longLastingOperation(num) 调用上使用concatMap 运算符将响应包装在 Observable 中并按顺序订阅结果：

range(1, 5)
.concatMap((num) => {
  return Observable.fromPromise(longLastingOperation(num));
})
.subscribe(res => {
  console.log(`End working on ${res}`) // Shall be ordered here
})

这是文档参考：
http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-concatMap

这是我在这个主题上的一些额外工作：
https://www.linkedin.com/pulse/rx-map-misleading-marbles-tomasz-budzi%C5%84ski/