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,236 @@
+'use strict';
+
+const {
+ ObjectDefineProperties,
+ Symbol,
+} = primordials;
+
+const {
+ TextDecoder,
+ TextEncoder,
+} = require('internal/encoding');
+
+const {
+ TransformStream,
+} = require('internal/webstreams/transformstream');
+
+const {
+ kEnumerableProperty,
+} = require('internal/webstreams/util');
+
+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: 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,
 };