streams: add whatwg stream interop

nodejs · devsnek · Aug 20, 2018 · Aug 19, 2018 · Aug 20, 2018 · Aug 20, 2018
commit e0117e7963aaf46d5a83056812ef72ff6c558b6a
diff --git a/doc/api/stream.md b/doc/api/stream.md
@@ -341,6 +341,23 @@ reader.pipe(writer);
 reader.unpipe(writer);
 ```
 
+##### writable.acquireStandardStream()
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* Returns: {WritableStream} to feed this stream.
+
+```js
+const fs = require('fs');
+
+const stream = fs.createWriteStream('file').acquireStandardStream();
+const writer = stream.getWriter();
+writer.write('hi!');
+```
+
 ##### writable.cork()
 <!-- YAML
 added: v0.11.2
@@ -827,6 +844,22 @@ If both `'readable'` and [`'data'`][]  are used at the same time, `'readable'`
 takes precedence in controlling the flow, i.e. `'data'` will be emitted
 only when [`stream.read()`][stream-read] is called.
 
+##### readable.acquireStandardStream()
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* Returns: {ReadableStream} to fully consume the stream.
+
+```js
+const fs = require('fs');
+
+const stream = fs.createReadStream('file').acquireStandardStream();
+const reader = stream.getReader();
+```
+
 ##### readable.destroy([error])
 <!-- YAML
 added: v8.0.0
@@ -1266,6 +1299,26 @@ Examples of `Duplex` streams include:
 * [zlib streams][zlib]
 * [crypto streams][crypto]
 
+##### duplex.acquireStandardStream
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* Returns {Object}
+  * `readable` {ReadableStream}
+  * `writable` {WritableStream}
+
+Creates a WHATWG stream pair to represent this duplex stream.
+
+```js
+const stream = getDuplexSomehow();
+const { readable, writable } = stream.acquireStandardStream();
+readable.getReader();
+writable.getWriter();
+```
+
 #### Class: stream.Transform
 <!-- YAML
 added: v0.9.4
@@ -2200,6 +2253,15 @@ by [`stream._transform()`][stream-_transform]. The `'end'` event is emitted
 after all data has been output, which occurs after the callback in
 [`transform._flush()`][stream-_flush] has been called.
 
+##### transform.acquireStandardStream()
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* Returns: {TransformStream}
+
 #### transform.\_flush(callback)
 
 * `callback` {Function} A callback function (optionally with an error

diff --git a/lib/_stream_duplex.js b/lib/_stream_duplex.js
@@ -31,6 +31,7 @@ module.exports = Duplex;
 const util = require('util');
 const Readable = require('_stream_readable');
 const Writable = require('_stream_writable');
+const { emitExperimentalWarning } = require('internal/util');
 
 util.inherits(Duplex, Readable);
 
@@ -66,6 +67,13 @@ function Duplex(options) {
   }
 }
 
+Duplex.prototype.acquireStandardStream = function() {
+  emitExperimentalWarning('Duplex.acquireStandardStream');
+  const readable = Readable.prototype.acquireStandardStream.call(this);
+  const writable = Writable.prototype.acquireStandardStream.call(this);
+  return { readable, writable };
+};
+
 Object.defineProperty(Duplex.prototype, 'writableHighWaterMark', {
   // making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
@@ -988,6 +988,29 @@ Readable.prototype[Symbol.asyncIterator] = function() {
   return new ReadableAsyncIterator(this);
 };
 
+Readable.prototype.acquireStandardStream = function() {
+  emitExperimentalWarning('Readable.acquireStandardStream');
+  return new ReadableStream({
+    start: (controller) => {
+      this.pause();
+      this.on('data', (chunk) => {
+        controller.enqueue(chunk);
+        this.pause();
+      });
+      this.once('end', () => controller.close());
+      this.once('error', (e) => controller.error(e));
+    },
+    pull: () => {
+      this.resume();
+    },
+    cancel: () => {
+      this.destroy();
+    },
+  }, {
+    highWaterMark: this.readableHighWaterMark,
+  });
+};
+
 Object.defineProperty(Readable.prototype, 'readableHighWaterMark', {
   // making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in

diff --git a/lib/_stream_transform.js b/lib/_stream_transform.js
@@ -72,6 +72,7 @@ const {
 } = require('internal/errors').codes;
 const Duplex = require('_stream_duplex');
 const util = require('util');
+const { emitExperimentalWarning } = require('internal/util');
 util.inherits(Transform, Duplex);
 
 
@@ -202,6 +203,32 @@ Transform.prototype._destroy = function(err, cb) {
   });
 };
 
+Transform.prototype.acquireStandardStream = function() {
+  emitExperimentalWarning('Transform.acquireStandardStream');
+  return new TransformStream({
+    start: (controller) => {
+      this.on('data', (chunk) => {
+        controller.enqueue(chunk);
+      });
+      this.once('end', () => controller.close());
+      this.once('error', (e) => controller.error(e));
+    },
+    transform: (chunk) => {
+      return new Promise((resolve) => {
+        const underHighWaterMark = this.write(chunk);
+        if (!underHighWaterMark) {
+          this.once('drain', resolve);
+        } else {
+          resolve();
+        }
+      });
+    },
+    flush: () => {
+      this.end();
+    },
+  });
+};
+
 
 function done(stream, er, data) {
   if (er)

diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
@@ -591,6 +591,33 @@ Writable.prototype.end = function(chunk, encoding, cb) {
   return this;
 };
 
+Writable.prototype.acquireStandardStream = function() {
+  internalUtil.emitExperimentalWarning('Writable.acquireStandardStream');
+  return new WritableStream({
+    start: (controller) => {
+      this.once('error', (e) => controller.error(e));
+    },
+    write: (chunk) => {
+      return new Promise((resolve) => {
+        const underHighWaterMark = this.write(chunk);
+        if (!underHighWaterMark) {
+          this.once('drain', resolve);
+        } else {
+          resolve();
+        }
+      });
+    },
+    close: (controller) => {
+      this.end();
+    },
+    abort: (reason) => {
+      this.destroy(reason);
+    },
+  }, {
+    highWaterMark: this.writableHighWaterMark,
+  });
+};
+
 Object.defineProperty(Writable.prototype, 'writableLength', {
   // making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in

diff --git a/tools/doc/type-parser.js b/tools/doc/type-parser.js
@@ -101,6 +101,10 @@ const customTypesMap = {
 
   'readline.Interface': 'readline.html#readline_class_interface',
 
+  'ReadableStream': 'https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream',
+  'WritableStream': 'https://developer.mozilla.org/en-US/docs/Web/API/WritableStream',
+  'TransformStream': 'https://streams.spec.whatwg.org/#ts-class',
+
   'Stream': 'stream.html#stream_stream',
   'stream.Duplex': 'stream.html#stream_class_stream_duplex',
   'stream.Readable': 'stream.html#stream_class_stream_readable',