异步生成器：产生被拒绝的承诺

Question

我一直在玩异步生成器，试图创建一个“承诺排序”生成器，它接受一组承诺并按照它们解决或拒绝的顺序一个接一个地产生承诺。所以像：

async function* orderProms(prom_arr) {

    // Make a copy so the splices don't mess it up.
    const proms = [...prom_arr];

    while (proms.length) {
        // Tag each promise with it's index, so that we can remove it for the next loop.
        const {prom, index} = await Promise.race(proms.map((prom, index) => prom.then(
            () => ({prom, index}),
            () => ({prom, index})
        )));

        proms.splice(index, 1);
        yield prom;
    }
}

想像这样使用这个生成器：

const resAfter = (val, delay) => new Promise(res => setTimeout(() => res(val), delay));
const rejAfter = (val, delay) => new Promise((_, rej) => setTimeout(() => rej(val), delay));

const promises = [
    resAfter("Third", 3000),
    resAfter("First", 1000),
    rejAfter("Second", 2000), // NOTE: this one rejects!
];

(async () => {

    let ordered = orderProms(promises);

    let done = false;
    for (let next_promise = ordered.next(); !done; next_promise = ordered.next()) {
        const next = await next_promise
            .catch(err => ({done: false, value: `Caught error: ${err}`}));

        done = next.done;
        if (!done) console.log(next.value);
    }
})()

但是，我注意到这将达到第二个承诺，然后生成器将停止。似乎是因为拒绝了“第二个”承诺。当prom 被拒绝时，在生成器中调用yield prom 将在生成器中创建一个异常。

但这是我困惑的根源。我不想在这里创建异常，我只想将被拒绝的承诺作为迭代器结果的value 产生。我不希望它被打开。这几乎就像被视为yield await prom;，但正如您所见，没有await 调用。

这里发生了什么，我怎样才能简单地从这个生成器中产生一个被拒绝的承诺。

---

这是可运行的 sn-p 中的上述代码：

async function* orderProms(prom_arr) {

    // Make a copy so the splices don't mess it up.
    const proms = [...prom_arr];

    while (proms.length) {
        // Tag each promise with it's index, so that we can remove it for the next loop.
        const {prom, index} = await Promise.race(proms.map((prom, index) => prom.then(
            () => ({prom, index}),
            () => ({prom, index})
        )));

        proms.splice(index, 1);
        yield prom;
    }
}

const resAfter = (val, delay) => new Promise(res => setTimeout(() => res(val), delay));
const rejAfter = (val, delay) => new Promise((_, rej) => setTimeout(() => rej(val), delay));

const promises = [
    resAfter("Third", 3000),
    resAfter("First", 1000),
    rejAfter("Second", 2000), // NOTE: this one rejects!
];

(async () => {

    let ordered = orderProms(promises);

    let done = false;
    for (let next_promise = ordered.next(); !done; next_promise = ordered.next()) {
        const next = await next_promise
            .catch(err => ({done: false, value: `Caught error: ${err}`}));

        done = next.done;
        if (!done) console.log(next.value);
    }
})()

Answer 1

这几乎就像被视为yield await prom。这是怎么回事？

Exactly that 是异步生成器的行为方式。

我怎样才能简单地从这个生成器中产生一个被拒绝的承诺。

你不能。请注意，异步迭代器预计会被

使用

try {
    for await (const value of orderProms(promises)) {
        console.log(value);
    }
} catch(err) {
    console.error('Caught error: ', err);
}

语法中没有对个别错误处理的便利。当出现异常时，循环停止，生成器完成。点。

那你能做什么？我看到三个选择：

• 保持原样，尽早将失败视为一项功能（类似于Promise.all）

• 处理错误（在orderProms 中或在将承诺传递给它之前）并产生承诺状态和值的元组

  for await (const value of orderProms(promises.map(prom =>
      prom.catch(err => `Caught error: ${err}`)
  ))) {
      console.log(value);
  }
  

• 使用普通的（非async）生成器手动生成一个又一个promise，以便能够以您想要的方式使用它

Answer 2

您可以让承诺解析为类似于您从 Promise.allSettled 获得的内容：

async function* orderProms(prom_arr) {
    // Make a copy so the splices don't mess it up.
    const proms = new Set(prom_arr.map((prom, index) => ({prom, index})));
    while (proms.size) {
        const settled = await Promise.race(Array.from(proms, obj => obj.prom.then(
            value => Object.assign(obj, { value, status: "fulfilled" }),
            error => Object.assign(obj, { error, status: "rejected" }),
        )));
        proms.delete(settled);
        let { prom, ...rest } = settled;
        yield rest;
    }
}

const resAfter = (val, delay) => new Promise(res => setTimeout(() => res(val), delay));
const rejAfter = (val, delay) => new Promise((_, rej) => setTimeout(() => rej(val), delay));

const promises = [
    resAfter("Third", 3000),
    resAfter("First", 1000),
    rejAfter("Second", 2000), // NOTE: this one rejects!
];

(async () => {
    for await (let result of orderProms(promises)) {
        console.log(JSON.stringify(result));
    }
})().catch(err => console.log(err.message));

Answer 3

我不能说接受的答案是错误的，但也不是很正确。特别是

当出现异常时，循环停止，生成器完成。 点。

部分有问题。

根据您的问题，虽然现代 JS 允许我们对这个问题采取一些优雅的方法，但按照您的要求，我们仍然可以让它工作，即使我认为它不是......那么好。

第一部分 - 对原始问题的回答

我不会详细介绍任何细节，只注意finally 在生成器函数orderProms 中的用法，在从内部抛出异常后，事情仍然会发生（生成器未完成）。所以.. 一种方法可能是这样的；

async function* orderProms(prom_arr) {

    // Make a copy so the splices don't mess it up.
    var proms = [...prom_arr];

        // Tag each promise with it's index, so that we can remove it for the next loop.
     try {
      while (proms.length) {
            var {prom, index} = await Promise.race(proms.map((prom, index) => prom.then(
                () => ({prom, index}),
                () => ({prom, index})
            )));
            
            proms.splice(index, 1);
            yield prom;
          }
    } finally {
        proms.length && (ordered = orderProms(proms));
      }
}

var resAfter = (val, delay) => new Promise(res => setTimeout(() => res(val), delay)),
    rejAfter = (val, delay) => new Promise((_, rej) => setTimeout(() => rej(val), delay)),
    promises = [ resAfter("Third", 3000)
               , resAfter("First", 1000)
               , rejAfter("Second", 2000) // NOTE: this one rejects!
               ],
    ordered  = orderProms(promises);

async function endPoint() {
    try {
      for await (var value of ordered) {
        console.log(value)
      }
    }
    catch(e){
      console.log(`caught rejection ${e} at endpoint`);
      endPoint();
    }
}

endPoint();

第 II 部分 - 问题的优雅解决方案

现在想象一下.. 如果我们有一个数组，我们可以像处理普通数组一样用 Promise 填充它，并且其中的 Promise 会根据它们的解决/拒绝时间自动排序。

首先让我们扩展Array 类型并赋予它特殊的异步能力。下面的代码定义了SortedAsyncArray 类型并且只是一个骨架。它没有经过全面测试，但应该足以给出一个想法。再次注意finally 部分，因为它仅在yield 由于异常或耗尽（生成器完成情况）挂起时才会执行。

class SortedPromisesArray extends Array {
  constructor(...args){
    super(...args);
  };
  async *[Symbol.asyncIterator]() {
    try {
      while(this.length){
        var {v,i} = await Promise.race(this.map((p,i) => p.then(v => ({v,i}))));
        this.splice(i,1);
        yield v;
      }
    } finally {
        this.length && this.splice(i,1);
    };
  };
};

那么我们应该如何使用这个异步数组呢？我想出的方法如下。

var promise  = (val, delay, resolves) => new Promise((v,x) => setTimeout(() => resolves ? v(val) : x(val), delay)),
    promises = [ promise("Third",  3000, true)
               , promise("First",  1000, true)
               , promise("Second", 2000, false) // NOTE: this one rejects!
               ],
    sortedPS = new SortedPromisesArray(...promises);

async function sink() {
  try {
    for await (let value of sortedPS){
      console.log(`Got: ${value}`);
    }
  } catch(err) {
    console.log(`caught at endpoint --> exception ${err}`);
    sink();
  };
};
sink();

第三部分正确的方法

我们在 第二部分 中看到的内容很好而且很优雅，但是它滥用了Promise.race()。如果这不是一个异步代码，那将是一种罪过。嗯……在我的书中，这仍然是一种罪过。为什么你会多次竞速最慢的承诺.length？

现在我们将尝试通过深入研究来正确解决这种愚蠢问题

• Private Class Fields 隐藏那些我们不想让任何人弄乱的变量。
• 我们甚至会检查提供的值是否是真正的承诺，并丢弃那些不是承诺的值。尽管在检查之后，如果您愿意，可以通过 Promise.resolve(notAPromise) 声明向他们承诺。
• 去掉finally 部分中的finally 块。

我们神奇地排序的异步数组现在变成了

class SortedPromisesArray extends Array {
  #RESOLVE;
  #REJECT;
  #COUNT;
  constructor(...args){
    super(...args.filter(p => Object(p).constructor === Promise));
    this.#COUNT = this.length;
    this.forEach(p => p.then(v => this.#RESOLVE(v), e => this.#REJECT(e)));
  };
  async *[Symbol.asyncIterator]() {
    while(this.#COUNT--) {
      yield new Promise((resolve,reject) => ( this.#RESOLVE = resolve
                                            , this.#REJECT  = reject
                                            ));
    };
  };
};

而sink() 函数只是与第二部分中的相同。

var promise  = (val, delay, resolves) => new Promise((v,x) => setTimeout(() => resolves ? v(val) : x(val), delay)),
    promises = [ promise("Third", 3000, true)
               , promise("First", 1000, true)
               , promise("Second", 2000, false) // NOTE: this one rejects!
               ],
    sortedPS = new SPA(...promises);

async function sink() {
  try {
    for await (let value of sortedPS){
      console.log(`Got: ${value}`);
    }
  } catch(err) {
    console.log(`caught at endpoint --> exception ${err}`);
    sink();
  }
}

sink();

再次...这不是生产代码。此处仅用于演示目的，但它向现代 JS/TS 中的良好设计模式迈出了一步。