readline: add error handling for input stream

lib/readline.js

@@ -190,6 +190,10 @@ function Interface(input, output, completer, terminal) {
     this._ttyWrite = _ttyWriteDumb.bind(this);
   }
 
+  function onerror(err) {
+    self.emit('error', err);
+  }
+
   function ondata(data) {
     self._normalWrite(data);
   }
@@ -227,9 +231,12 @@ function Interface(input, output, completer, terminal) {
 
   this[kLineObjectStream] = undefined;
 
+  input.on('error', onerror);
+
   if (!this.terminal) {
     function onSelfCloseWithoutTerminal() {
       input.removeListener('data', ondata);
+      input.removeListener('error', onerror);
       input.removeListener('end', onend);
     }
 
@@ -240,6 +247,7 @@ function Interface(input, output, completer, terminal) {
   } else {
     function onSelfCloseWithTerminal() {
       input.removeListener('keypress', onkeypress);
+      input.removeListener('error', onerror);
       input.removeListener('end', ontermend);
       if (output !== null && output !== undefined) {
         output.removeListener('resize', onresize);
@@ -1098,12 +1106,17 @@ Interface.prototype[SymbolAsyncIterator] = function() {
     });
     const lineListener = (input) => {
       if (!readable.push(input)) {
+        // TODO(rexagod): drain to resume flow
         this.pause();
       }
     };
     const closeListener = () => {
       readable.push(null);
     };
+    const errorListener = (err) => {
+      readable.destroy(err);
+    };
+    this.on('error', errorListener);
     this.on('line', lineListener);
     this.on('close', closeListener);
     this[kLineObjectStream] = readable;

test/parallel/test-readline-input-onerror.js

@@ -0,0 +1,40 @@
+'use strict';
+const common = require('../common');
+const fs = require('fs');
+const readline = require('readline');
+const path = require('path');
+
+async function processLineByLine_SymbolAsyncError(filename) {
+  const fileStream = fs.createReadStream(filename);
+  const rl = readline.createInterface({
+    input: fileStream,
+    crlfDelay: Infinity
+  });
+  // eslint-disable-next-line no-unused-vars
+  for await (const line of rl) {
+    /* check SymbolAsyncIterator `errorListener` */
+  }
+}
+
+const f = path.join(__dirname, 'file.txt');
+
+// catch-able SymbolAsyncIterator `errorListener` error
+processLineByLine_SymbolAsyncError(f).catch(common.expectsError({
+  code: 'ENOENT',
+  message: `ENOENT: no such file or directory, open '${f}'`
+}));
+
+async function processLineByLine_InterfaceErrorEvent(filename) {
+  const fileStream = fs.createReadStream(filename);
+  const rl = readline.createInterface({
+    input: fileStream,
+    crlfDelay: Infinity
+  });
+  rl.on('error', common.expectsError({
+    code: 'ENOENT',
+    message: `ENOENT: no such file or directory, open '${f}'`
+  }));
+}
+
+// check Interface 'error' event
+processLineByLine_InterfaceErrorEvent(f);