如何使用 Promise 等待 observable 的每个值

假设我有这个可观察的：

const obs = new Observable((observer) => {
    observer.next(0.25);
    observer.next(0.75);
    observer.next(new ArrayBuffer(100));
    observer.complete();
});

我怎样才能用承诺来等待每个值？

以下代码将仅返回最后一个值（调用complete()之前的值）：

const value = await obs.toPromise();

但我希望能够一路获得每个值。我可以做这样的事情：

const value1 = await obs.pipe(take(1)).toPromise();
const value2 = await obs.pipe(take(2)).toPromise();

但这并不理想，因为我每次都必须增加数字，而且take(1000)即使只有 3 个值，示例中仍会返回一些内容。我正在寻找类似的东西：

const value1 = await obs.pipe(next()).toPromise(); // 0.25
const value2 = await obs.pipe(next()).toPromise(); // 0.75
const value3 = await obs.pipe(next()).toPromise(); // ArrayBuffer(100)
const value4 = await obs.pipe(next()).toPromise(); // null

这更类似于发电机。

有没有办法完成这样的事情？

看来您要求的是一种将可观察量转换为异步可迭代量的方法，以便您可以“手动”或使用新的异步迭代其值等待-等待 https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of语言特征。

以下是如何执行此操作的示例（我尚未测试此代码，因此它可能存在一些错误）：

// returns an asyncIterator that will iterate over the observable values
function asyncIterator(observable) {
  const queue = []; // holds observed values not yet delivered
  let complete = false;
  let error = undefined;
  const promiseCallbacks = [];

  function sendNotification() {
    // see if caller is waiting on a result
    if (promiseCallbacks.length) {
      // send them the next value if it exists
      if (queue.length) {
        const value = queue.shift();
        promiseCallbacks.shift()[0]({ value, done: false });
      }
      // tell them the iteration is complete
      else if (complete) {
        while (promiseCallbacks.length) {
          promiseCallbacks.shift()[0]({ done: true });
        }
      }
      // send them an error
      else if (error) {
        while (promiseCallbacks.length) {
          promiseCallbacks.shift()[1](error);
        }
      }
    }
  }

  observable.subscribe(
    value => {
      queue.push(value);
      sendNotification();
    },
    err => {
      error = err;
      sendNotification();
    },
    () => {
      complete = true;
      sendNotification();
    });

  // return the iterator
  return {
    next() {
      return new Promise((resolve, reject) => {
        promiseCallbacks.push([resolve, reject]);
        sendNotification();
      });
    }
  }
}

与 for-wait-of 语言功能一起使用：

async someFunction() {
  const obs = ...;
  const asyncIterable = {
    [Symbol.asyncIterator]: () => asyncIterator(obs)
  };

  for await (const value of asyncIterable) {
    console.log(value);
  }
}

使用无需等待的语言功能：

async someFunction() {
  const obs = ...;
  const it = asyncIterator(obs);

  while (true) {
    const { value, done } = await it.next();
    if (done) {
      break;
    }

    console.log(value);
  }
}