标签: async-iterator

假设我们有一个异步生成器：

exports.asyncGen = async function* (items) {
  for (const item of items) {
    const result = await someAsyncFunc(item)
    yield result;
  }
}

是否可以映射该生成器？本质上我想这样做：

const { asyncGen } = require('./asyncGen.js')

exports.process = async function (items) {
  return asyncGen(items).map(item => {
    //... do something
  })
}

截至目前.map无法识别异步迭代器。

另一种方法是使用for await ... of ，但这远不如使用.map

我正在尝试了解 javascript 的Symbol.asyncIterator和await of。我写了一些简单的代码，它抛出一个错误说：

    TypeError: undefined is not a function

在尝试使用for await (let x of a).

我不明白这样做的原因。

let a = {}


function test() {
        for(let i=0; i < 10; i++) {
                if(i > 5) {
                        return Promise.resolve(`Greater than 5: (${i})`)
                }else {
                        return Promise.resolve(`Less than 5: (${i})`)
                }
        }
}

a[Symbol.asyncIterator] = test;


async function main() {
        for await (let x of a) { // LINE THAT THROWS AN ERROR
                console.log(x)
        }
}


main() …

我一直在尝试使用异步生成器来尝试制作一个“承诺排序”生成器，该生成器接受一系列承诺并按照它们解决或拒绝的顺序逐个产生承诺。所以像这样：

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 …

这是我在 node v14.4.0 中提炼为最小的、可重现的示例的更大过程的一部分。在此代码中，它从for循环内部不输出任何内容。

我在控制台中只看到这个输出：

before for() loop
finished
finally
done

该for await (const line1 of rl1)循环永远不会进入for循环-它只是跳过就在它：

const fs = require('fs');
const readline = require('readline');
const { once } = require('events');

async function test(file1, file2) {
    try {
        const stream1 = fs.createReadStream(file1);
        await once(stream1, 'open');
        const rl1 = readline.createInterface({input: stream1, crlfDelay: Infinity});

        const stream2 = fs.createReadStream(file2);
        await once(stream2, 'open');
        const rl2 = readline.createInterface({input: stream2, crlfDelay: Infinity});

        console.log('before for() loop');
        for await (const line1 of …

设想

我得到了一个带有异步回调的函数，例如

let readFile:   (path: string, callback: (line: string, eof: boolean) => void) => void

虽然我更喜欢使用AsyncIterable / AsyncGenerator签名的函数：

let readFileV2: (path: string) => AsyncIterable<string>

问题

如果没有readFileV2，我必须读取类似的文件

let file = await new Promise((res, err) => {
    let file = ''
    readFile('./myfile.txt', (line, eof) => {
        if (eof) { return res(file) }
        file += line + '\n'
    })
})

..同时readFileV2允许我做得更干净，就像

let file = '';
for await (let line of readFileV2('./myfile.txt')) {
    file += line + '\n'
} …

假设我有一个发电机：

function* source() {
  yield "hello"; yield "world";
}

我创建了可迭代对象，使用 for 循环进行迭代，然后在迭代器完全完成之前退出循环（返回完成）。

function run() {
  for (let item of source()) {
    console.log(item);
    break;
  }
}

问题：如何从可迭代端发现迭代器提前终止？

如果您尝试直接在生成器本身中执行此操作，似乎没有任何反馈：

function* source2() {
  try {
    let result = yield "hello";
    console.log("foo");
  } catch (err) {
    console.log("bar");
  }
}

...既没有记录“foo”也没有记录“bar”。

考虑来自 MDN 的这个基本的AsyncIterator 示例：

var asyncIterable = {
  [Symbol.asyncIterator]() {
    return {
      i: 0,
      next() {
        if (this.i < 3) {
          return Promise.resolve({ value: this.i++, done: false });
        }

        return Promise.resolve({ done: true });
      }
    };
  }
};

(async function() {
   for await (let num of asyncIterable) {
     console.log(num);
   }
})();

在节点 10.16.0 上运行它可以正常工作。但是，我似乎无法通过 Typescript 运行它。使用这个 tsconfig：

{
  "compilerOptions": {
    "lib": ["es2016", "esnext.asynciterable"],
    "target": "es2016"
  }
}

导致错误The type returned by the 'next()' method of …

基于processLineByLine()的示例，我注意到如果给定的文件名不存在，我们将无法捕获错误。在这种情况下，程序以如下内容结束：

UnhandledPromiseRejectionWarning：错误：ENOENT：没有这样的文件或目录

因此，我用来引发可捕获错误的最简单方法是对processLineByLine()函数进行 2 处修改：

1. 将其放入发电机中，例如 function*
2. await 文件存在检查 await access(filename, fs.constants.F_OK)

最后我不得不将readline.Interface实例转换为异步生成器。我特别不喜欢这最后一部分。结果lines()函数是这样的：

export async function* lines(filename) {
    await access(filename, fs.constants.F_OK)
    const lines = readline.createInterface({
        input: fs.createReadStream(filename),
        crlfDelay: Infinity
    })
    for await (const l of lines) {
        yield l
    }
}

问题：lines()如果文件名不存在，是否有更好的方法来返回异步迭代器或抛出错误？

BUG 报告：关于@jfriend00 观察，我在 nodejs 上打开了一个 Bug 问题：https : //github.com/nodejs/node/issues/30831