javascript - 使用 Observable 排队 promise

Question

我有一个 promise 流进来，我不希望它们在所有以前的 promises 都解决之前开始，但是同时使用 Observables。

我会尝试这样的事情:

let queue = Rx.Observable.create();

function addToQueue() {
  // How do I add this to the queue.
}

function removeFromQueue() {
  // How do i remove this item from the queue.
}

function getNewObservable() {
  const queueItem = Rx.Observable.fromPromise(new Promise(resolve => {
    setTimeout(() => {
      removeFromQueue();
      resolve();
    }, 1000);
  }));

  addToQueue();
  return queue;
}

getNewObservable().then(() => console.log('After 1000'));
getNewObservable().then(() => console.log('After 2000'));

// This one shouldn't run until the top two are finished
getNewObservable().then(() => console.log('After 3000'));

// This one would just wait for the previous one to resolve since
// the first two already resolved
setTimeout(() => {
  getNewObservable().then(() => console.log('After 4000'));
}, 2500);

Answer 1

那只是 concatMap 运算符。唯一的问题是您在哪里创建 Promise:

const queue = Rx.Observable.create();
const results = queue.concatMap(p => createPromise(p));

results.subscribe(console.log);

queue.add(getNewObservable(...));
queue.add(getNewObservable(...));
queue.add(getNewObservable(...));