events: don't call resume after close (fix readline regression)

lib/events.js

@@ -1071,7 +1071,7 @@ function on(emitter, event, options = kEmptyObject) {
         const value = unconsumedEvents.shift();
         size--;
         if (paused && size < lowWatermark) {
-          emitter.resume();
+          emitter.resume(); // Can not be finished yet
           paused = false;
         }
         return PromiseResolve(createIterResult(value, false));
@@ -1188,6 +1188,7 @@ function on(emitter, event, options = kEmptyObject) {
     abortListenerDisposable?.[SymbolDispose]();
     removeAll();
     finished = true;
+    paused = false;
     const doneResult = createIterResult(undefined, true);
     while (!unconsumedPromises.isEmpty()) {
       unconsumedPromises.shift().resolve(doneResult);

test/parallel/test-readline-async-iterators.js

@@ -17,6 +17,7 @@ const testContents = [
   'line 1',
   'line 1\nline 2 南越国是前203年至前111年存在于岭南地区的一个国家\nline 3\ntrailing',
   'line 1\nline 2\nline 3 ends with newline\n',
+  Array(1e4).fill(0).map((_, i) => i).join('\n'), // More that 2 * highWaterMark
 ];
 
 async function testSimple() {
@@ -43,6 +44,29 @@ async function testSimple() {
   }
 }
 
+// Same as testSimple, but with Readable.from() instead of fs.createReadStream
+async function testReadableFrom() {
+  for (const fileContent of testContents) {
+    const readable = Readable.from([fileContent]);
+    const rli = readline.createInterface({
+      input: readable,
+      crlfDelay: Infinity
+    });
+
+    const iteratedLines = [];
+    for await (const k of rli) {
+      iteratedLines.push(k);
+    }
+
+    const expectedLines = fileContent.split('\n');
+    if (expectedLines[expectedLines.length - 1] === '') {
+      expectedLines.pop();
+    }
+    assert.deepStrictEqual(iteratedLines, expectedLines);
+    assert.strictEqual(iteratedLines.join(''), fileContent.replace(/\n/g, ''));
+  }
+}
+
 async function testMutual() {
   for (const fileContent of testContents) {
     fs.writeFileSync(filename, fileContent);
@@ -115,6 +139,7 @@ async function testSlowStreamForLeaks() {
 }
 
 testSimple()
+  .then(testReadableFrom)
   .then(testMutual)
   .then(testSlowStreamForLeaks)
   .then(common.mustCall());