javascript - 为什么这个 readline 异步迭代器不能正常工作？

Question

这是我在 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 rl1) {
            console.log(line1);
        }
        console.log('finished');
    } finally {
        console.log('finally');
    }
}

test("data/numbers.txt", "data/letters.txt").then(() => {
    console.log(`done`);
}).catch(err => {
    console.log('Got rejected promise:', err);
})

但是，如果我删除 await once(stream, 'open') 中的任何一个语句，然后是 for loop 完全符合预期(列出 rl1 文件的所有行)。因此，显然，来自 readline 接口(interface)和流之间的异步迭代器存在一些计时问题。任何想法可能发生的事情。知道什么可能导致这个问题或如何解决它吗？
仅供引用， await once(stream, 'open')是否因为异步迭代器中的另一个错误，如果打开文件时出现问题，它不会拒绝，而 await once(stream, 'open')如果文件无法打开(基本上是在打开之前)，则会导致您正确地获得拒绝。
如果您想知道为什么存在 stream2 代码，它在较大的项目中使用，但我已将此示例缩减为最小的、可重现的示例，并且只需要这么多代码来演示问题。

编辑:在尝试稍微不同的实现时，我发现如果我将这两个 once(stream, "open")调用 Promise.all() ，然后它就可以工作了。所以，这有效:

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


async function test(file1, file2) {
    try {
        const stream1 = fs.createReadStream(file1);
        const rl1 = readline.createInterface({input: stream1, crlfDelay: Infinity});
        const stream2 = fs.createReadStream(file2);
        const rl2 = readline.createInterface({input: stream2, crlfDelay: Infinity});
        // pre-flight file open to catch any open errors here
        // because of existing bug in async iterator with file open errors
        await Promise.all([once(stream1, "open"), once(stream2, "open")]);

        console.log('before for() loop');
        for await (const line1 of rl1) {
            console.log(line1);
        }
        console.log('finished');
    } finally {
        console.log('finally');
    }
}

test("data/numbers.txt", "data/letters.txt").then(() => {
    console.log(`done`);
}).catch(err => {
    console.log('Got rejected promise:', err);
});

这显然不应该对您等待文件打开的确切方式敏感。某处存在一些计时错误。我想在 readline 或 readStream 上找到该错误并将其归档。有任何想法吗？

Answer 1

事实证明，根本问题是 readline.createInterface()立即，调用它会添加一个data事件监听器 (code reference here) 并恢复流以开始流。

input.on('data', ondata);

和

input.resume();

然后，在 ondata监听器，它解析行的数据，当它找到一行时，它会触发 line事件 here .

for (let n = 0; n < lines.length; n++)
  this._onLine(lines[n]);

但是，在我的示例中，在 readline.createInterface() 之间发生了其他异步事件。被调用并创建了异步迭代器(它将监听 line 事件)。所以， line正在发出事件，但还没有任何东西在监听它们。
所以，要正常工作 readline.createInterface()要求任何要监听 line事件必须在调用 readline.createInterface() 后同步添加或者存在竞争条件和 line 事件可能会丢失。

在我的原始代码示例中，一种可靠的解决方法是不调用 readline.createInterface()直到我完成 await once(...) 之后.然后，异步迭代器将在 readline.createInterface() 之后同步创建。叫做。

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


async function test(file1, file2) {
    try {
        const stream1 = fs.createReadStream(file1);
        const stream2 = fs.createReadStream(file2);
        // wait for both files to be open to catch any "open" errors here
        // since readline has bugs about not properly reporting file open errors
        // this await must be done before either call to readline.createInterface()
        // to avoid race conditions that can lead to lost lines of data
        await Promise.all([once(stream1, "open"), once(stream2, "open")]);

        const rl1 = readline.createInterface({input: stream1, crlfDelay: Infinity});
        const rl2 = readline.createInterface({input: stream2, crlfDelay: Infinity});

        console.log('before for() loop');
        for await (const line1 of rl1) {
            console.log(line1);
        }
        console.log('finished');
    } finally {
        console.log('finally');
    }
}

test("data/numbers.txt", "data/letters.txt").then(() => {
    console.log(`done`);
}).catch(err => {
    console.log('Got rejected promise:', err);
});

解决此一般问题的一种方法是更改​​ readline.createInterface()这样它就不会添加 data事件并恢复流直到有人添加 line事件监听器。这将防止数据丢失。它将允许 readline 接口(interface)对象安静地坐在那里而不会丢失数据，直到其输出的接收器实际准备好。这适用于异步迭代器，并且还可以防止混入其他异步代码的接口(interface)的其他用途可能丢失 line。事件。
对此的注释已添加到相关的 open readline 错误问题 here .