stream: implement TextEncoderStream and TextDecoderStream

Experimental as part of the web streams implementation

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

PR-URL: https://github.com/nodejs/node/pull/39347
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

doc/api/webstreams.md

@@ -1118,5 +1118,104 @@ added: v16.5.0
   * `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,217 @@
+'use strict';
+
+const {
+  ObjectDefineProperties,
+  Symbol,
+} = primordials;
+
+const {
+  TextDecoder,
+  TextEncoder,
+} = require('internal/encoding');
+
+const {
+  TransformStream,
+} = require('internal/webstreams/transformstream');
+
+const {
+  customInspect,
+  kEnumerableProperty,
+} = require('internal/webstreams/util');
+
+const {
+  codes: {
+    ERR_INVALID_THIS,
+  },
+} = require('internal/errors');
+
+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');
+    return customInspect(depth, options, 'TextEncoderStream', {
+      encoding: this[kHandle].encoding,
+      readable: this[kTransform].readable,
+      writable: this[kTransform].writable,
+    });
+  }
+}
+
+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');
+    return customInspect(depth, options, 'TextDecoderStream', {
+      encoding: this[kHandle].encoding,
+      fatal: this[kHandle].fatal,
+      ignoreBOM: this[kHandle].ignoreBOM,
+      readable: this[kTransform].readable,
+      writable: this[kTransform].writable,
+    });
+  }
+}
+
+ObjectDefineProperties(TextEncoderStream.prototype, {
+  encoding: kEnumerableProperty,
+  readable: kEnumerableProperty,
+  writable: kEnumerableProperty,
+});
+
+ObjectDefineProperties(TextDecoderStream.prototype, {
+  encoding: kEnumerableProperty,
+  fatal: kEnumerableProperty,
+  ignoreBOM: kEnumerableProperty,
+  readable: kEnumerableProperty,
+  writable: kEnumerableProperty,
+});
+
+module.exports = {
+  TextEncoderStream,
+  TextDecoderStream,
+};

lib/stream/web.js

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

test/parallel/test-whatwg-webstreams-encoding.js

@@ -0,0 +1,102 @@
+// Flags: --no-warnings
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  TextEncoderStream,
+  TextDecoderStream,
+} = require('stream/web');
+
+const kEuroBytes = Buffer.from([0xe2, 0x82, 0xac]);
+const kEuro = Buffer.from([0xe2, 0x82, 0xac]).toString();
+
+[1, false, [], {}, 'hello'].forEach((i) => {
+  assert.throws(() => new TextDecoderStream(i), {
+    code: 'ERR_ENCODING_NOT_SUPPORTED',
+  });
+});
+
+[1, false, 'hello'].forEach((i) => {
+  assert.throws(() => new TextDecoderStream(undefined, i), {
+    code: 'ERR_INVALID_ARG_TYPE',
+  });
+});
+
+{
+  const tds = new TextDecoderStream();
+  const writer = tds.writable.getWriter();
+  const reader = tds.readable.getReader();
+  reader.read().then(common.mustCall(({ value, done }) => {
+    assert(!done);
+    assert.strictEqual(kEuro, value);
+    reader.read().then(common.mustCall(({ done }) => {
+      assert(done);
+    }));
+  }));
+  Promise.all([
+    writer.write(kEuroBytes.slice(0, 1)),
+    writer.write(kEuroBytes.slice(1, 2)),
+    writer.write(kEuroBytes.slice(2, 3)),
+    writer.close(),
+  ]).then(common.mustCall());
+
+  assert.strictEqual(tds.encoding, 'utf-8');
+  assert.strictEqual(tds.fatal, false);
+  assert.strictEqual(tds.ignoreBOM, false);
+
+  assert.throws(
+    () => Reflect.get(TextDecoderStream.prototype, 'encoding', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(TextDecoderStream.prototype, 'fatal', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(TextDecoderStream.prototype, 'ignoreBOM', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(TextDecoderStream.prototype, 'readable', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(TextDecoderStream.prototype, 'writable', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+}
+
+{
+  const tds = new TextEncoderStream();
+  const writer = tds.writable.getWriter();
+  const reader = tds.readable.getReader();
+  reader.read().then(common.mustCall(({ value, done }) => {
+    assert(!done);
+    const buf = Buffer.from(value.buffer, value.byteOffset, value.byteLength);
+    assert.deepStrictEqual(kEuroBytes, buf);
+    reader.read().then(common.mustCall(({ done }) => {
+      assert(done);
+    }));
+  }));
+  Promise.all([
+    writer.write(kEuro),
+    writer.close(),
+  ]).then(common.mustCall());
+
+  assert.strictEqual(tds.encoding, 'utf-8');
+
+  assert.throws(
+    () => Reflect.get(TextEncoderStream.prototype, 'encoding', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(TextEncoderStream.prototype, 'readable', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(TextEncoderStream.prototype, 'writable', {}), {
+      code: 'ERR_INVALID_THIS',
+    });
+}

tools/doc/type-parser.mjs

@@ -253,6 +253,10 @@ const customTypesMap = {
     'webstreams.md#webstreamsapi_class_bytelengthqueuingstrategy',
   'CountQueuingStrategy':
     'webstreams.md#webstreamsapi_class_countqueuingstrategy',
+  'TextEncoderStream':
+    'webstreams.md#webstreamsapi_class_textencoderstream',
+  'TextDecoderStream':
+    'webstreams.md#webstreamsapi_class_textdecoderstream',
 };
 
 const arrayPart = /(?:\[])+$/;