streams: implement TextEncoderStream and TextDecoderStream

Experimental as part of the web streams implementation

Signed-off-by: James M Snell <jasnell@gmail.com>

doc/api/webstreams.md

@@ -1118,5 +1118,103 @@ added: REPLACEME
   * `chunk` {any}
   * Returns: {number}
 
+### Class: `TextEncoderStream`
+<!-- YAML
+added: REPLACEME
+-->
+
+#### `new TextEncoderStream()`
+<!-- YAML
+added: REPLACEME
+-->
+
+Creates a new `TextEncoderStream` instance.
+#### `textEncoderStream.encoding`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {string}
+
+The encoding supported by the `TextEncoderStream` instance.
+
+#### `textEncoderStream.readable`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {ReadableStream}
+
+#### `textEncoderStream.writable`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {WritableStream}
+
+### Class: `TextDecoderStream`
+<!-- YAML
+added: REPLACEME
+-->
+
+#### `new TextDecoderStream([encoding[, options]])`
+<!-- YAML
+added: REPLACEME
+-->
+
+* `encoding` {string} Identifies the `encoding` that this `TextDecoder` instance
+  supports. **Default:** `'utf-8'`.
+* `options` {Object}
+  * `fatal` {boolean} `true` if decoding failures are fatal.
+  * `ignoreBOM` {boolean} When `true`, the `TextDecoderStream` will include the
+    byte order mark in the decoded result. When `false`, the byte order mark
+    will be removed from the output. This option is only used when `encoding` is
+    `'utf-8'`, `'utf-16be'` or `'utf-16le'`. **Default:** `false`.
+
+Creates a new `TextDecoderStream` instance.
+
+#### `textDecoderStream.encoding`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {string}
+
+The encoding supported by the `TextDecoderStream` instance.
+
+#### `textDecoderStream.fatal`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {boolean}
+
+The value will be `true` if decoding errors result in a `TypeError` being
+thrown.
+
+#### `textDecoderStream.ignoreBOM`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {boolean}
+
+The value will be `true` if the decoding result will include the byte order
+mark.
+
+#### `textDecoderStream.readable`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {ReadableStream}
+
+#### `textDecoderStream.writable`
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {WritableStream}
+
 [Streams]: stream.md
 [WHATWG Streams Standard]: https://streams.spec.whatwg.org/

lib/internal/webstreams/encoding.js

@@ -0,0 +1,232 @@
+'use strict';
+
+const {
+  ObjectDefineProperties,
+  Symbol,
+} = primordials;
+
+const {
+  TextDecoder,
+  TextEncoder,
+} = require('internal/encoding');
+
+const {
+  TransformStream,
+} = require('internal/webstreams/transformstream');
+
+const {
+  codes: {
+    ERR_INVALID_THIS,
+  },
+} = require('internal/errors');
+
+const {
+  inspect,
+} = require('internal/util/inspect');
+
+const {
+  customInspectSymbol: kInspect
+} = require('internal/util');
+
+const kHandle = Symbol('kHandle');
+const kTransform = Symbol('kTransform');
+const kType = Symbol('kType');
+
+/**
+ * @typedef {import('./readablestream').ReadableStream} ReadableStream
+ * @typedef {import('./writablestream').WritableStream} WritableStream
+ */
+
+function isTextEncoderStream(value) {
+  return typeof value?.[kHandle] === 'object' &&
+         value?.[kType] === 'TextEncoderStream';
+}
+
+function isTextDecoderStream(value) {
+  return typeof value?.[kHandle] === 'object' &&
+         value?.[kType] === 'TextDecoderStream';
+}
+
+class TextEncoderStream {
+  constructor() {
+    this[kType] = 'TextEncoderStream';
+    this[kHandle] = new TextEncoder();
+    this[kTransform] = new TransformStream({
+      transform: (chunk, controller) => {
+        const value = this[kHandle].encode(chunk);
+        if (value)
+          controller.enqueue(value);
+      },
+      flush: (controller) => {
+        const value = this[kHandle].encode();
+        if (value.byteLength > 0)
+          controller.enqueue(value);
+        controller.terminate();
+      },
+    });
+  }
+
+  /**
+   * @readonly
+   * @type {string}
+   */
+  get encoding() {
+    if (!isTextEncoderStream(this))
+      throw new ERR_INVALID_THIS('TextEncoderStream');
+    return this[kHandle].encoding;
+  }
+
+  /**
+   * @readonly
+   * @type {ReadableStream}
+   */
+  get readable() {
+    if (!isTextEncoderStream(this))
+      throw new ERR_INVALID_THIS('TextEncoderStream');
+    return this[kTransform].readable;
+  }
+
+  /**
+   * @readonly
+   * @type {WritableStream}
+   */
+  get writable() {
+    if (!isTextEncoderStream(this))
+      throw new ERR_INVALID_THIS('TextEncoderStream');
+    return this[kTransform].writable;
+  }
+
+  [kInspect](depth, options) {
+    if (!isTextEncoderStream(this))
+      throw new ERR_INVALID_THIS('TextEncoderStream');
+    if (depth < 0)
+      return this;
+
+    const opts = {
+      ...options,
+      depth: options.depth == null ? null : options.depth - 1
+    };
+
+    return `${this[kType]} ${inspect({
+      encoding: this[kHandle].encoding,
+      readable: this[kTransform].readable,
+      writable: this[kTransform].writable,
+    }, opts)}`;
+  }
+}
+
+class TextDecoderStream {
+  /**
+   * @param {string} [encoding]
+   * @param {{
+   *   fatal? : boolean,
+   *   ignoreBOM? : boolean,
+   * }} [options]
+   */
+  constructor(encoding = 'utf-8', options = {}) {
+    this[kType] = 'TextDecoderStream';
+    this[kHandle] = new TextDecoder(encoding, options);
+    this[kTransform] = new TransformStream({
+      transform: (chunk, controller) => {
+        const value = this[kHandle].decode(chunk, { stream: true });
+        if (value)
+          controller.enqueue(value);
+      },
+      flush: (controller) => {
+        const value = this[kHandle].decode();
+        if (value)
+          controller.enqueue(value);
+        controller.terminate();
+      },
+    });
+  }
+
+  /**
+   * @readonly
+   * @type {string}
+   */
+  get encoding() {
+    if (!isTextDecoderStream(this))
+      throw new ERR_INVALID_THIS('TextDecoderStream');
+    return this[kHandle].encoding;
+  }
+
+  /**
+   * @readonly
+   * @type {boolean}
+   */
+  get fatal() {
+    if (!isTextDecoderStream(this))
+      throw new ERR_INVALID_THIS('TextDecoderStream');
+    return this[kHandle].fatal;
+  }
+
+  /**
+   * @readonly
+   * @type {boolean}
+   */
+  get ignoreBOM() {
+    if (!isTextDecoderStream(this))
+      throw new ERR_INVALID_THIS('TextDecoderStream');
+    return this[kHandle].ignoreBOM;
+  }
+
+  /**
+   * @readonly
+   * @type {ReadableStream}
+   */
+  get readable() {
+    if (!isTextDecoderStream(this))
+      throw new ERR_INVALID_THIS('TextDecoderStream');
+    return this[kTransform].readable;
+  }
+
+  /**
+   * @readonly
+   * @type {WritableStream}
+   */
+  get writable() {
+    if (!isTextDecoderStream(this))
+      throw new ERR_INVALID_THIS('TextDecoderStream');
+    return this[kTransform].writable;
+  }
+
+  [kInspect](depth, options) {
+    if (!isTextDecoderStream(this))
+      throw new ERR_INVALID_THIS('TextDecoderStream');
+    if (depth < 0)
+      return this;
+
+    const opts = {
+      ...options,
+      depth: options.depth == null ? null : options.depth - 1
+    };
+
+    return `${this[kType]} ${inspect({
+      encoding: this[kHandle].encoding,
+      fatal: this[kHandle].fatal,
+      ignoreBOM: this[kHandle].ignoreBOM,
+      readable: this[kTransform].readable,
+      writable: this[kTransform].writable,
+    }, opts)}`;
+  }
+}
+
+ObjectDefineProperties(TextEncoderStream.prototype, {
+  encoding: { enumerable: true },
+  readable: { enumerable: true },
+  writable: { enumerable: true },
+});
+
+ObjectDefineProperties(TextDecoderStream.prototype, {
+  encoding: { enumerable: true },
+  fatal: { enumerable: true },
+  ignoreBOM: { enumerable: true },
+  readable: { enumerable: true },
+  writable: { enumerable: true },
+});
+
+module.exports = {
+  TextEncoderStream,
+  TextDecoderStream,
+};

lib/stream/web.js

@@ -31,6 +31,11 @@ const {
   CountQueuingStrategy,
 } = require('internal/webstreams/queuingstrategies');
 
+const {
+  TextEncodingStream,
+  TextDecodingStream,
+} = require('internal/webstreams/encoding');
+
 module.exports = {
   ReadableStream,
   ReadableStreamDefaultReader,
@@ -45,4 +50,6 @@ module.exports = {
   WritableStreamDefaultController,
   ByteLengthQueuingStrategy,
   CountQueuingStrategy,
+  TextEncodingStream,
+  TextDecodingStream,
 };