From f574c2be5c986c4c66b32d51fd343987b0508055 Mon Sep 17 00:00:00 2001
From: Ivan Filenko <ivan.filenko@protonmail.com>
Date: Wed, 21 Feb 2018 19:28:24 +0300
Subject: [PATCH 1/2] readline: add support for async iteration

---
 doc/api/readline.md                           |  57 ++++++
 doc/api/stream.md                             |   3 +
 lib/_stream_readable.js                       |   2 +-
 lib/internal/readline/async_iterator.js       | 166 ++++++++++++++++++
 lib/internal/streams/async_iterator.js        |   5 +-
 lib/readline.js                               |  33 +++-
 node.gyp                                      |   1 +
 .../parallel/test-readline-async-iterators.js |  43 +++++
 8 files changed, 307 insertions(+), 3 deletions(-)
 create mode 100644 lib/internal/readline/async_iterator.js
 create mode 100644 test/parallel/test-readline-async-iterators.js

diff --git a/doc/api/readline.md b/doc/api/readline.md
index a2a4adf3093a9a..945c5ca35ef141 100644
--- a/doc/api/readline.md
+++ b/doc/api/readline.md
@@ -306,6 +306,37 @@ rl.write(null, { ctrl: true, name: 'u' });
 The `rl.write()` method will write the data to the `readline` `Interface`'s
 `input` *as if it were provided by the user*.
 
+### rl[@@asyncIterator]
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+Returns an [AsyncIterator][async-iterator] to fully consume the stream.
+
+```js
+const readline = require('readline');
+const fs = require('fs');
+
+async function processLineByLine(readable) {
+  readable.setEncoding('utf8');
+  const rli = readline.createInterface({
+    input: readable,
+    crlfDelay: Infinity
+  });
+
+  for await (const line of rli) {
+    console.log(line);
+  }
+}
+
+processLineByLine(fs.createReadStream('file')).catch(console.error);
+```
+
+If the loop terminates with a `break` or a `throw`, the stream will be destroyed.
+In other terms, iterating over a stream will consume the stream fully.
+
 ## readline.clearLine(stream, dir)
 <!-- YAML
 added: v0.7.7
@@ -527,6 +558,29 @@ rl.on('line', (line) => {
 });
 ```
 
+> Stability: 1 - Experimental
+
+Another way is to use async [for-await-of][for-await-of] iteration statement:
+
+```js
+const readline = require('readline');
+const fs = require('fs');
+
+async function processLineByLine(readable) {
+  readable.setEncoding('utf8');
+  const rli = readline.createInterface({
+    input: readable,
+    crlfDelay: Infinity
+  });
+
+  for await (const line of rli) {
+    console.log(line);
+  }
+}
+
+processLineByLine(fs.createReadStream('file')).catch(console.error);
+```
+
 [`'SIGCONT'`]: readline.html#readline_event_sigcont
 [`'SIGTSTP'`]: readline.html#readline_event_sigtstp
 [`process.stdin`]: process.html#process_process_stdin
@@ -535,3 +589,6 @@ rl.on('line', (line) => {
 [TTY]: tty.html
 [Writable]: stream.html#stream_writable_streams
 [reading files]: #readline_example_read_file_stream_line_by_line
+[async-iterator]: https://github.com/tc39/proposal-async-iteration
+[for-await-of]: https://github.com/tc39/proposal-async-iteration#the-async-iteration-statement-for-await-of
+
diff --git a/doc/api/stream.md b/doc/api/stream.md
index 483aec1b700fb6..5c9a09d6bb875a 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1241,6 +1241,8 @@ fully. The stream will be read in chunks of size equal to the `highWaterMark`
 option. In the code example above, data will be in a single chunk if the file
 has less then 64kb of data because no `highWaterMark` option is provided to
 [`fs.createReadStream()`][].
+Use [readline][readline-async-iterator] if there is a need for line-by-line
+async file iteration.
 
 ### Duplex and Transform Streams
 
@@ -2518,3 +2520,4 @@ contain multi-byte characters.
 [readable-destroy]: #stream_readable_destroy_error
 [writable-_destroy]: #stream_writable_destroy_err_callback
 [writable-destroy]: #stream_writable_destroy_error
+[readline-async-iterator]: readline.html#readline_rl_asynciterator
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 31a8a11e4a0673..9149c3be473f95 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -38,7 +38,7 @@ const {
   ERR_METHOD_NOT_IMPLEMENTED,
   ERR_STREAM_UNSHIFT_AFTER_END_EVENT
 } = require('internal/errors').codes;
-const ReadableAsyncIterator = require('internal/streams/async_iterator');
+const { ReadableAsyncIterator } = require('internal/streams/async_iterator');
 const { emitExperimentalWarning } = require('internal/util');
 var StringDecoder;
 
diff --git a/lib/internal/readline/async_iterator.js b/lib/internal/readline/async_iterator.js
new file mode 100644
index 00000000000000..ea815f64eed3d8
--- /dev/null
+++ b/lib/internal/readline/async_iterator.js
@@ -0,0 +1,166 @@
+'use strict';
+
+const kLastResolve = Symbol('lastResolve');
+const kLastReject = Symbol('lastReject');
+const kError = Symbol('error');
+const kEnded = Symbol('ended');
+const kLastPromise = Symbol('lastPromise');
+const kHandlePromise = Symbol('handlePromise');
+const kStream = Symbol('stream');
+const kReadlineInterface = Symbol('readlineInterface');
+const { AsyncIteratorRecord } = require('internal/streams/async_iterator');
+
+function readAndResolve(iter) {
+  const resolve = iter[kLastResolve];
+  if (resolve !== null) {
+    const data = iter[kReadlineInterface].read();
+    // we defer if data is null
+    // we can be expecting either 'end' or
+    // 'error'
+    if (data !== null) {
+      iter[kLastPromise] = null;
+      iter[kLastResolve] = null;
+      iter[kLastReject] = null;
+      resolve(new AsyncIteratorRecord(data, false));
+    }
+  }
+}
+
+function writeToBuffer(iter) {
+  const data = iter[kStream].read();
+  if (data !== null) {
+    iter[kReadlineInterface]._normalWrite(data);
+  }
+  process.nextTick(readAndResolve, iter);
+}
+
+function onReadable(iter) {
+  process.nextTick(writeToBuffer, iter);
+}
+
+function onEnd(iter) {
+  const resolve = iter[kLastResolve];
+  if (resolve !== null) {
+    iter[kLastPromise] = null;
+    iter[kLastResolve] = null;
+    iter[kLastReject] = null;
+    resolve(new AsyncIteratorRecord(null, true));
+  }
+  iter[kEnded] = true;
+}
+
+function onError(iter, err) {
+  const reject = iter[kLastReject];
+  // reject if we are waiting for data in the Promise
+  // returned by next() and store the error
+  if (reject !== null) {
+    iter[kLastPromise] = null;
+    iter[kLastResolve] = null;
+    iter[kLastReject] = null;
+    reject(err);
+  }
+  iter.error = err;
+}
+
+function wrapForNext(lastPromise, iter) {
+  return function(resolve, reject) {
+    lastPromise.then(function() {
+      iter[kHandlePromise](resolve, reject);
+    }, reject);
+  };
+}
+
+const ReadlineAsyncIterator = class ReadlineAsyncIterator {
+  constructor(readline_interface) {
+    this[kReadlineInterface] = readline_interface;
+    this[kStream] = readline_interface.input;
+    this[kLastResolve] = null;
+    this[kLastReject] = null;
+    this[kError] = null;
+    this[kEnded] = false;
+    this[kLastPromise] = null;
+
+    this[kStream].on('readable', onReadable.bind(null, this));
+    this[kStream].on('end', onEnd.bind(null, this));
+    this[kStream].on('error', onError.bind(null, this));
+
+    // the function passed to new Promise
+    // is cached so we avoid allocating a new
+    // closure at every run
+    this[kHandlePromise] = (resolve, reject) => {
+      const data = this[kReadlineInterface].read();
+      console.log('iterator qwe 2');
+      console.log(data);
+      // throw new Error("Something unexpected has occurred.");
+      if (data) {
+        this[kLastPromise] = null;
+        this[kLastResolve] = null;
+        this[kLastReject] = null;
+        resolve(new AsyncIteratorRecord(data, false));
+      } else {
+        this[kLastResolve] = resolve;
+        this[kLastReject] = reject;
+      }
+    };
+  }
+
+  get stream() {
+    return this[kStream];
+  }
+
+  next() {
+    // if we have detected an error in the meanwhile
+    // reject straight away
+    const error = this[kError];
+    if (error !== null) {
+      return Promise.reject(error);
+    }
+
+    if (this[kEnded]) {
+      return Promise.resolve(new AsyncIteratorRecord(null, true));
+    }
+
+    // if we have multiple next() calls
+    // we will wait for the previous Promise to finish
+    // this logic is optimized to support for await loops,
+    // where next() is only called once at a time
+    const lastPromise = this[kLastPromise];
+    let promise;
+
+    if (lastPromise) {
+      promise = new Promise(wrapForNext(lastPromise, this));
+      console.log('iterator qwe 4');
+      console.log(promise);
+    } else {
+      // fast path needed to support multiple this.push()
+      // without triggering the next() queue
+      const data = this[kReadlineInterface].read();
+      if (data !== null) {
+        return Promise.resolve(new AsyncIteratorRecord(data, false));
+      }
+
+      promise = new Promise(this[kHandlePromise]);
+    }
+
+    this[kLastPromise] = promise;
+
+    return promise;
+  }
+
+  return() {
+    // destroy(err, cb) is a private API
+    // we can guarantee we have that here, because we control the
+    // Readable class this is attached to
+    return new Promise((resolve, reject) => {
+      this[kStream].destroy(null, (err) => {
+        if (err) {
+          reject(err);
+          return;
+        }
+        resolve(new AsyncIteratorRecord(null, true));
+      });
+    });
+  }
+};
+
+module.exports = ReadlineAsyncIterator;
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index 0e34573d877aee..f454b0013a2e5e 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -156,4 +156,7 @@ const ReadableAsyncIterator = class ReadableAsyncIterator {
   }
 };
 
-module.exports = ReadableAsyncIterator;
+module.exports = {
+  ReadableAsyncIterator,
+  AsyncIteratorRecord
+};
diff --git a/lib/readline.js b/lib/readline.js
index 124fc8111b2f09..579c6cce5196f6 100644
--- a/lib/readline.js
+++ b/lib/readline.js
@@ -34,8 +34,11 @@ const {
 } = require('internal/errors').codes;
 const { debug, inherits } = require('util');
 const { Buffer } = require('buffer');
+const BufferList = require('internal/streams/BufferList');
 const EventEmitter = require('events');
 const { StringDecoder } = require('string_decoder');
+const { emitExperimentalWarning } = require('internal/util');
+const ReadlineAsyncIterator = require('internal/readline/async_iterator');
 const {
   CSI,
   emitKeys,
@@ -77,6 +80,8 @@ function Interface(input, output, completer, terminal) {
   this.isCompletionEnabled = true;
   this._sawKeyPress = false;
   this._previousKey = null;
+  this._enableBuffer = false;
+  this._buffer = new BufferList();
 
   EventEmitter.call(this);
   var historySize;
@@ -238,6 +243,24 @@ Interface.prototype.setPrompt = function(prompt) {
   this._prompt = prompt;
 };
 
+Interface.prototype.read = function() {
+  if (this._buffer.length === 0)
+    return null;
+
+  return this._buffer.shift();
+};
+
+
+Interface.prototype[Symbol.asyncIterator] = function() {
+  emitExperimentalWarning('readline Interface[Symbol.asyncIterator]');
+
+  this._enableBuffer = true;
+  this.close();
+  this.input.setEncoding('utf8');
+
+  return new ReadlineAsyncIterator(this);
+};
+
 
 Interface.prototype._setRawMode = function(mode) {
   const wasInRawMode = this.input.isRaw;
@@ -424,8 +447,16 @@ Interface.prototype._normalWrite = function(b) {
     // either '' or (conceivably) the unfinished portion of the next line
     string = lines.pop();
     this._line_buffer = string;
-    for (var n = 0; n < lines.length; n++)
+    for (var n = 0; n < lines.length; n++) {
       this._onLine(lines[n]);
+
+      if (this._enableBuffer) {
+        this._buffer.push(lines[n]);
+      }
+    }
+    if (this._enableBuffer) {
+      this._buffer.push(string);
+    }
   } else if (string) {
     // no newlines this time, save what we have for next time
     this._line_buffer = string;
diff --git a/node.gyp b/node.gyp
index 78a15bd7b838a7..c5ecf18da56da7 100644
--- a/node.gyp
+++ b/node.gyp
@@ -127,6 +127,7 @@
       'lib/internal/querystring.js',
       'lib/internal/process/write-coverage.js',
       'lib/internal/readline.js',
+      'lib/internal/readline/async_iterator.js',
       'lib/internal/repl.js',
       'lib/internal/repl/await.js',
       'lib/internal/socket_list.js',
diff --git a/test/parallel/test-readline-async-iterators.js b/test/parallel/test-readline-async-iterators.js
new file mode 100644
index 00000000000000..e4202dd764af46
--- /dev/null
+++ b/test/parallel/test-readline-async-iterators.js
@@ -0,0 +1,43 @@
+'use strict';
+
+const common = require('../common');
+const fs = require('fs');
+const { join } = require('path');
+const readline = require('readline');
+const assert = require('assert');
+
+common.crashOnUnhandledRejection();
+
+const tmpdir = require('../common/tmpdir');
+tmpdir.refresh();
+
+const filename = join(tmpdir.path, 'test.txt');
+const file_content = 'abc123\n' +
+                     '南越国是前203年至前111年存在于岭南地区的一个国家\n';
+const file_lines_count = 3;
+const without_line_breaks = file_content.replace(/(\n)/gm, '');
+
+fs.writeFileSync(filename, file_content);
+
+async function tests() {
+  await (async () => {
+    const readable = fs.createReadStream(filename);
+    const rli = readline.createInterface({
+      input: readable,
+      crlfDelay: Infinity
+    });
+
+    let data = '';
+    let iterations = 0;
+    for await (const k of rli) {
+      data += k;
+      iterations += 1;
+    }
+
+    assert.strictEqual(data, without_line_breaks);
+    assert.strictEqual(iterations, file_lines_count);
+  })();
+}
+
+// to avoid missing some tests if a promise does not resolve
+tests().then(common.mustCall());

From 2f5c5584276eaa7160c295425f228f5106e925df Mon Sep 17 00:00:00 2001
From: Ivan Filenko <ivan.filenko@protonmail.com>
Date: Sat, 5 May 2018 20:12:39 +0300
Subject: [PATCH 2/2] stream: move common async iteration logic to
 base_async_iterator

---
 doc/api/readline.md                         |   5 +-
 lib/_stream_readable.js                     |   2 +-
 lib/internal/readline/async_iterator.js     | 211 ++++++++---------
 lib/internal/streams/async_iterator.js      | 238 ++++++++------------
 lib/internal/streams/base_async_iterator.js | 107 +++++++++
 lib/readline.js                             |   2 +-
 node.gyp                                    |   1 +
 7 files changed, 296 insertions(+), 270 deletions(-)
 create mode 100644 lib/internal/streams/base_async_iterator.js

diff --git a/doc/api/readline.md b/doc/api/readline.md
index 945c5ca35ef141..0f00ac5af5a2e3 100644
--- a/doc/api/readline.md
+++ b/doc/api/readline.md
@@ -334,8 +334,9 @@ async function processLineByLine(readable) {
 processLineByLine(fs.createReadStream('file')).catch(console.error);
 ```
 
-If the loop terminates with a `break` or a `throw`, the stream will be destroyed.
-In other terms, iterating over a stream will consume the stream fully.
+If the loop terminates with a `break` or a `throw`, the stream will be 
+destroyed. In other terms, iterating over a stream will consume the stream 
+fully.
 
 ## readline.clearLine(stream, dir)
 <!-- YAML
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 9149c3be473f95..31a8a11e4a0673 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -38,7 +38,7 @@ const {
   ERR_METHOD_NOT_IMPLEMENTED,
   ERR_STREAM_UNSHIFT_AFTER_END_EVENT
 } = require('internal/errors').codes;
-const { ReadableAsyncIterator } = require('internal/streams/async_iterator');
+const ReadableAsyncIterator = require('internal/streams/async_iterator');
 const { emitExperimentalWarning } = require('internal/util');
 var StringDecoder;
 
diff --git a/lib/internal/readline/async_iterator.js b/lib/internal/readline/async_iterator.js
index ea815f64eed3d8..3fd88999e1ab16 100644
--- a/lib/internal/readline/async_iterator.js
+++ b/lib/internal/readline/async_iterator.js
@@ -1,14 +1,20 @@
 'use strict';
 
-const kLastResolve = Symbol('lastResolve');
-const kLastReject = Symbol('lastReject');
-const kError = Symbol('error');
-const kEnded = Symbol('ended');
-const kLastPromise = Symbol('lastPromise');
-const kHandlePromise = Symbol('handlePromise');
-const kStream = Symbol('stream');
 const kReadlineInterface = Symbol('readlineInterface');
-const { AsyncIteratorRecord } = require('internal/streams/async_iterator');
+const {
+  kLastResolve,
+  kLastReject,
+  kError,
+  kEnded,
+  kLastPromise,
+  kHandlePromise,
+  kStream,
+  onEnd,
+  onError,
+  wrapForNext,
+  AsyncIteratorRecord,
+  BaseAsyncIterator
+} = require('internal/streams/base_async_iterator');
 
 function readAndResolve(iter) {
   const resolve = iter[kLastResolve];
@@ -38,129 +44,94 @@ function onReadable(iter) {
   process.nextTick(writeToBuffer, iter);
 }
 
-function onEnd(iter) {
-  const resolve = iter[kLastResolve];
-  if (resolve !== null) {
-    iter[kLastPromise] = null;
-    iter[kLastResolve] = null;
-    iter[kLastReject] = null;
-    resolve(new AsyncIteratorRecord(null, true));
-  }
-  iter[kEnded] = true;
-}
-
-function onError(iter, err) {
-  const reject = iter[kLastReject];
-  // reject if we are waiting for data in the Promise
-  // returned by next() and store the error
-  if (reject !== null) {
-    iter[kLastPromise] = null;
-    iter[kLastResolve] = null;
-    iter[kLastReject] = null;
-    reject(err);
-  }
-  iter.error = err;
-}
+const ReadlineAsyncIterator =
+  class ReadlineAsyncIterator extends BaseAsyncIterator {
+    constructor(readline_interface) {
+      super();
+      this[kReadlineInterface] = readline_interface;
+      this[kStream] = readline_interface.input;
+      this[kLastResolve] = null;
+      this[kLastReject] = null;
+      this[kError] = null;
+      this[kEnded] = false;
+      this[kLastPromise] = null;
+
+      this[kStream].on('readable', onReadable.bind(null, this));
+      this[kStream].on('end', onEnd.bind(null, this));
+      this[kStream].on('error', onError.bind(null, this));
+
+      // the function passed to new Promise
+      // is cached so we avoid allocating a new
+      // closure at every run
+      this[kHandlePromise] = (resolve, reject) => {
+        const data = this[kReadlineInterface].read();
+        if (data) {
+          this[kLastPromise] = null;
+          this[kLastResolve] = null;
+          this[kLastReject] = null;
+          resolve(new AsyncIteratorRecord(data, false));
+        } else {
+          this[kLastResolve] = resolve;
+          this[kLastReject] = reject;
+        }
+      };
+    }
 
-function wrapForNext(lastPromise, iter) {
-  return function(resolve, reject) {
-    lastPromise.then(function() {
-      iter[kHandlePromise](resolve, reject);
-    }, reject);
-  };
-}
+    get stream() {
+      return this[kStream];
+    }
 
-const ReadlineAsyncIterator = class ReadlineAsyncIterator {
-  constructor(readline_interface) {
-    this[kReadlineInterface] = readline_interface;
-    this[kStream] = readline_interface.input;
-    this[kLastResolve] = null;
-    this[kLastReject] = null;
-    this[kError] = null;
-    this[kEnded] = false;
-    this[kLastPromise] = null;
-
-    this[kStream].on('readable', onReadable.bind(null, this));
-    this[kStream].on('end', onEnd.bind(null, this));
-    this[kStream].on('error', onError.bind(null, this));
-
-    // the function passed to new Promise
-    // is cached so we avoid allocating a new
-    // closure at every run
-    this[kHandlePromise] = (resolve, reject) => {
-      const data = this[kReadlineInterface].read();
-      console.log('iterator qwe 2');
-      console.log(data);
-      // throw new Error("Something unexpected has occurred.");
-      if (data) {
-        this[kLastPromise] = null;
-        this[kLastResolve] = null;
-        this[kLastReject] = null;
-        resolve(new AsyncIteratorRecord(data, false));
-      } else {
-        this[kLastResolve] = resolve;
-        this[kLastReject] = reject;
+    next() {
+      // if we have detected an error in the meanwhile
+      // reject straight away
+      const error = this[kError];
+      if (error !== null) {
+        return Promise.reject(error);
       }
-    };
-  }
 
-  get stream() {
-    return this[kStream];
-  }
+      if (this[kEnded]) {
+        return Promise.resolve(new AsyncIteratorRecord(null, true));
+      }
 
-  next() {
-    // if we have detected an error in the meanwhile
-    // reject straight away
-    const error = this[kError];
-    if (error !== null) {
-      return Promise.reject(error);
-    }
+      // if we have multiple next() calls
+      // we will wait for the previous Promise to finish
+      // this logic is optimized to support for await loops,
+      // where next() is only called once at a time
+      const lastPromise = this[kLastPromise];
+      let promise;
 
-    if (this[kEnded]) {
-      return Promise.resolve(new AsyncIteratorRecord(null, true));
-    }
+      if (lastPromise) {
+        promise = new Promise(wrapForNext(lastPromise, this));
+      } else {
+        // fast path needed to support multiple this.push()
+        // without triggering the next() queue
+        const data = this[kReadlineInterface].read();
+        if (data !== null) {
+          return Promise.resolve(new AsyncIteratorRecord(data, false));
+        }
 
-    // if we have multiple next() calls
-    // we will wait for the previous Promise to finish
-    // this logic is optimized to support for await loops,
-    // where next() is only called once at a time
-    const lastPromise = this[kLastPromise];
-    let promise;
-
-    if (lastPromise) {
-      promise = new Promise(wrapForNext(lastPromise, this));
-      console.log('iterator qwe 4');
-      console.log(promise);
-    } else {
-      // fast path needed to support multiple this.push()
-      // without triggering the next() queue
-      const data = this[kReadlineInterface].read();
-      if (data !== null) {
-        return Promise.resolve(new AsyncIteratorRecord(data, false));
+        promise = new Promise(this[kHandlePromise]);
       }
 
-      promise = new Promise(this[kHandlePromise]);
-    }
-
-    this[kLastPromise] = promise;
+      this[kLastPromise] = promise;
 
-    return promise;
-  }
+      return promise;
+    }
 
-  return() {
-    // destroy(err, cb) is a private API
-    // we can guarantee we have that here, because we control the
-    // Readable class this is attached to
-    return new Promise((resolve, reject) => {
-      this[kStream].destroy(null, (err) => {
-        if (err) {
-          reject(err);
-          return;
-        }
-        resolve(new AsyncIteratorRecord(null, true));
+    return() {
+      // destroy(err, cb) is a private API
+      // we can guarantee we have that here, because we control the
+      // Readable class this is attached to
+      return new Promise((resolve, reject) => {
+        this[kStream].destroy(null, (err) => {
+          if (err) {
+            reject(err);
+            return;
+          }
+          resolve(new AsyncIteratorRecord(null, true));
+        });
       });
-    });
-  }
-};
+    }
+  };
 
 module.exports = ReadlineAsyncIterator;
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index f454b0013a2e5e..27f2d250c71be7 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -1,162 +1,108 @@
 'use strict';
 
-const kLastResolve = Symbol('lastResolve');
-const kLastReject = Symbol('lastReject');
-const kError = Symbol('error');
-const kEnded = Symbol('ended');
-const kLastPromise = Symbol('lastPromise');
-const kHandlePromise = Symbol('handlePromise');
-const kStream = Symbol('stream');
-
-const AsyncIteratorRecord = class AsyncIteratorRecord {
-  constructor(value, done) {
-    this.done = done;
-    this.value = value;
-  }
-};
-
-function readAndResolve(iter) {
-  const resolve = iter[kLastResolve];
-  if (resolve !== null) {
-    const data = iter[kStream].read();
-    // we defer if data is null
-    // we can be expecting either 'end' or
-    // 'error'
-    if (data !== null) {
-      iter[kLastPromise] = null;
-      iter[kLastResolve] = null;
-      iter[kLastReject] = null;
-      resolve(new AsyncIteratorRecord(data, false));
+const {
+  kLastResolve,
+  kLastReject,
+  kError,
+  kEnded,
+  kLastPromise,
+  kHandlePromise,
+  kStream,
+  onReadable,
+  onEnd,
+  onError,
+  wrapForNext,
+  AsyncIteratorRecord,
+  BaseAsyncIterator
+} = require('internal/streams/base_async_iterator');
+
+const ReadableAsyncIterator =
+  class ReadableAsyncIterator extends BaseAsyncIterator {
+    constructor(stream) {
+      super();
+      this[kStream] = stream;
+      this[kLastResolve] = null;
+      this[kLastReject] = null;
+      this[kError] = null;
+      this[kEnded] = false;
+      this[kLastPromise] = null;
+
+      stream.on('readable', onReadable.bind(null, this));
+      stream.on('end', onEnd.bind(null, this));
+      stream.on('error', onError.bind(null, this));
+
+      // the function passed to new Promise
+      // is cached so we avoid allocating a new
+      // closure at every run
+      this[kHandlePromise] = (resolve, reject) => {
+        const data = this[kStream].read();
+        if (data) {
+          this[kLastPromise] = null;
+          this[kLastResolve] = null;
+          this[kLastReject] = null;
+          resolve(new AsyncIteratorRecord(data, false));
+        } else {
+          this[kLastResolve] = resolve;
+          this[kLastReject] = reject;
+        }
+      };
     }
-  }
-}
-
-function onReadable(iter) {
-  // we wait for the next tick, because it might
-  // emit an error with process.nextTick
-  process.nextTick(readAndResolve, iter);
-}
-
-function onEnd(iter) {
-  const resolve = iter[kLastResolve];
-  if (resolve !== null) {
-    iter[kLastPromise] = null;
-    iter[kLastResolve] = null;
-    iter[kLastReject] = null;
-    resolve(new AsyncIteratorRecord(null, true));
-  }
-  iter[kEnded] = true;
-}
-
-function onError(iter, err) {
-  const reject = iter[kLastReject];
-  // reject if we are waiting for data in the Promise
-  // returned by next() and store the error
-  if (reject !== null) {
-    iter[kLastPromise] = null;
-    iter[kLastResolve] = null;
-    iter[kLastReject] = null;
-    reject(err);
-  }
-  iter[kError] = err;
-}
 
-function wrapForNext(lastPromise, iter) {
-  return function(resolve, reject) {
-    lastPromise.then(function() {
-      iter[kHandlePromise](resolve, reject);
-    }, reject);
-  };
-}
-
-const ReadableAsyncIterator = class ReadableAsyncIterator {
-  constructor(stream) {
-    this[kStream] = stream;
-    this[kLastResolve] = null;
-    this[kLastReject] = null;
-    this[kError] = null;
-    this[kEnded] = false;
-    this[kLastPromise] = null;
-
-    stream.on('readable', onReadable.bind(null, this));
-    stream.on('end', onEnd.bind(null, this));
-    stream.on('error', onError.bind(null, this));
+    get stream() {
+      return this[kStream];
+    }
 
-    // the function passed to new Promise
-    // is cached so we avoid allocating a new
-    // closure at every run
-    this[kHandlePromise] = (resolve, reject) => {
-      const data = this[kStream].read();
-      if (data) {
-        this[kLastPromise] = null;
-        this[kLastResolve] = null;
-        this[kLastReject] = null;
-        resolve(new AsyncIteratorRecord(data, false));
-      } else {
-        this[kLastResolve] = resolve;
-        this[kLastReject] = reject;
+    next() {
+      // if we have detected an error in the meanwhile
+      // reject straight away
+      const error = this[kError];
+      if (error !== null) {
+        return Promise.reject(error);
       }
-    };
-  }
-
-  get stream() {
-    return this[kStream];
-  }
 
-  next() {
-    // if we have detected an error in the meanwhile
-    // reject straight away
-    const error = this[kError];
-    if (error !== null) {
-      return Promise.reject(error);
-    }
+      if (this[kEnded]) {
+        return Promise.resolve(new AsyncIteratorRecord(null, true));
+      }
 
-    if (this[kEnded]) {
-      return Promise.resolve(new AsyncIteratorRecord(null, true));
-    }
+      // if we have multiple next() calls
+      // we will wait for the previous Promise to finish
+      // this logic is optimized to support for await loops,
+      // where next() is only called once at a time
+      const lastPromise = this[kLastPromise];
+      let promise;
 
-    // if we have multiple next() calls
-    // we will wait for the previous Promise to finish
-    // this logic is optimized to support for await loops,
-    // where next() is only called once at a time
-    const lastPromise = this[kLastPromise];
-    let promise;
+      if (lastPromise) {
+        promise = new Promise(wrapForNext(lastPromise, this));
+      } else {
+        // fast path needed to support multiple this.push()
+        // without triggering the next() queue
+        const data = this[kStream].read();
+        if (data !== null) {
+          return Promise.resolve(new AsyncIteratorRecord(data, false));
+        }
 
-    if (lastPromise) {
-      promise = new Promise(wrapForNext(lastPromise, this));
-    } else {
-      // fast path needed to support multiple this.push()
-      // without triggering the next() queue
-      const data = this[kStream].read();
-      if (data !== null) {
-        return Promise.resolve(new AsyncIteratorRecord(data, false));
+        promise = new Promise(this[kHandlePromise]);
       }
 
-      promise = new Promise(this[kHandlePromise]);
-    }
+      this[kLastPromise] = promise;
 
-    this[kLastPromise] = promise;
-
-    return promise;
-  }
+      return promise;
+    }
 
-  return() {
-    // destroy(err, cb) is a private API
-    // we can guarantee we have that here, because we control the
-    // Readable class this is attached to
-    return new Promise((resolve, reject) => {
-      this[kStream].destroy(null, (err) => {
-        if (err) {
-          reject(err);
-          return;
-        }
-        resolve(new AsyncIteratorRecord(null, true));
+    return() {
+      // destroy(err, cb) is a private API
+      // we can guarantee we have that here, because we control the
+      // Readable class this is attached to
+      return new Promise((resolve, reject) => {
+        this[kStream].destroy(null, (err) => {
+          if (err) {
+            reject(err);
+            return;
+          }
+          resolve(new AsyncIteratorRecord(null, true));
+        });
       });
-    });
-  }
-};
+    }
+  };
 
-module.exports = {
-  ReadableAsyncIterator,
-  AsyncIteratorRecord
-};
+module.exports = ReadableAsyncIterator;
diff --git a/lib/internal/streams/base_async_iterator.js b/lib/internal/streams/base_async_iterator.js
new file mode 100644
index 00000000000000..e71cd8e2a76056
--- /dev/null
+++ b/lib/internal/streams/base_async_iterator.js
@@ -0,0 +1,107 @@
+'use strict';
+
+const kLastResolve = Symbol('lastResolve');
+const kLastReject = Symbol('lastReject');
+const kError = Symbol('error');
+const kEnded = Symbol('ended');
+const kLastPromise = Symbol('lastPromise');
+const kHandlePromise = Symbol('handlePromise');
+const kStream = Symbol('stream');
+const {
+  ERR_METHOD_NOT_IMPLEMENTED
+} = require('internal/errors').codes;
+
+const AsyncIteratorRecord = class AsyncIteratorRecord {
+  constructor(value, done) {
+    this.done = done;
+    this.value = value;
+  }
+};
+
+function readAndResolve(iter) {
+  const resolve = iter[kLastResolve];
+  if (resolve !== null) {
+    const data = iter[kStream].read();
+    // we defer if data is null
+    // we can be expecting either 'end' or
+    // 'error'
+    if (data !== null) {
+      iter[kLastPromise] = null;
+      iter[kLastResolve] = null;
+      iter[kLastReject] = null;
+      resolve(new AsyncIteratorRecord(data, false));
+    }
+  }
+}
+
+function onReadable(iter) {
+  // we wait for the next tick, because it might
+  // emit an error with process.nextTick
+  process.nextTick(readAndResolve, iter);
+}
+
+function onEnd(iter) {
+  const resolve = iter[kLastResolve];
+  if (resolve !== null) {
+    iter[kLastPromise] = null;
+    iter[kLastResolve] = null;
+    iter[kLastReject] = null;
+    resolve(new AsyncIteratorRecord(null, true));
+  }
+  iter[kEnded] = true;
+}
+
+function onError(iter, err) {
+  const reject = iter[kLastReject];
+  // reject if we are waiting for data in the Promise
+  // returned by next() and store the error
+  if (reject !== null) {
+    iter[kLastPromise] = null;
+    iter[kLastResolve] = null;
+    iter[kLastReject] = null;
+    reject(err);
+  }
+  iter[kError] = err;
+}
+
+function wrapForNext(lastPromise, iter) {
+  return function(resolve, reject) {
+    lastPromise.then(function() {
+      iter[kHandlePromise](resolve, reject);
+    }, reject);
+  };
+}
+
+// Abstract class. Methods should be overridden
+// in specific implementation classes
+const BaseAsyncIterator = class BaseAsyncIterator {
+  constructor(stream) {}
+
+  get stream() {
+    throw new ERR_METHOD_NOT_IMPLEMENTED('stream()');
+  }
+
+  next() {
+    throw new ERR_METHOD_NOT_IMPLEMENTED('next()');
+  }
+
+  return() {
+    throw new ERR_METHOD_NOT_IMPLEMENTED('return()');
+  }
+};
+
+module.exports = {
+  kLastResolve,
+  kLastReject,
+  kError,
+  kEnded,
+  kLastPromise,
+  kHandlePromise,
+  kStream,
+  onReadable,
+  onEnd,
+  onError,
+  wrapForNext,
+  BaseAsyncIterator,
+  AsyncIteratorRecord
+};
diff --git a/lib/readline.js b/lib/readline.js
index 579c6cce5196f6..8132d414158335 100644
--- a/lib/readline.js
+++ b/lib/readline.js
@@ -34,7 +34,7 @@ const {
 } = require('internal/errors').codes;
 const { debug, inherits } = require('util');
 const { Buffer } = require('buffer');
-const BufferList = require('internal/streams/BufferList');
+const BufferList = require('internal/streams/buffer_list');
 const EventEmitter = require('events');
 const { StringDecoder } = require('string_decoder');
 const { emitExperimentalWarning } = require('internal/util');
diff --git a/node.gyp b/node.gyp
index c5ecf18da56da7..bd3f3749e0e237 100644
--- a/node.gyp
+++ b/node.gyp
@@ -152,6 +152,7 @@
       'lib/internal/stream_base_commons.js',
       'lib/internal/vm/module.js',
       'lib/internal/streams/lazy_transform.js',
+      'lib/internal/streams/base_async_iterator.js',
       'lib/internal/streams/async_iterator.js',
       'lib/internal/streams/buffer_list.js',
       'lib/internal/streams/duplexpair.js',