Implement streams by adopting the reference implementation

lib/jsdom/browser/Window.js

@@ -4,7 +4,7 @@ const webIDLConversions = require("webidl-conversions");
 const { CSSStyleDeclaration } = require("cssstyle");
 const { Performance: RawPerformance } = require("w3c-hr-time");
 const notImplemented = require("./not-implemented");
-const { installInterfaces } = require("../living/interfaces");
+const { installInterfaces, defineVMGlobals } = require("../living/interfaces");
 const { define, mixin } = require("../utils");
 const Element = require("../living/generated/Element");
 const EventTarget = require("../living/generated/EventTarget");
@@ -105,6 +105,9 @@ function setupWindow(windowInstance, { runScripts }) {
   if (runScripts === "outside-only" || runScripts === "dangerously") {
     contextifyWindow(windowInstance);
 
+    // Workaround for https://github.com/nodejs/node/issues/38918
+    defineVMGlobals(windowInstance);
+
     // Without this, these globals will only appear to scripts running inside the context using vm.runScript; they will
     // not appear to scripts running from the outside, including to JSDOM implementation code.
     for (const [globalName, globalPropDesc] of jsGlobalEntriesToInstall) {

lib/jsdom/living/interfaces.js

@@ -1,6 +1,8 @@
 /* eslint-disable global-require */
 "use strict";
 
+const vm = require("vm");
+
 const style = require("../level2/style");
 const xpath = require("../level3/xpath");
 
@@ -185,7 +187,18 @@ const generatedInterfaces = {
 
   Headers: require("./generated/Headers"),
   AbortController: require("./generated/AbortController"),
-  AbortSignal: require("./generated/AbortSignal")
+  AbortSignal: require("./generated/AbortSignal"),
+
+  ByteLengthQueuingStrategy: require("./generated/ByteLengthQueuingStrategy"),
+  CountQueuingStrategy: require("./generated/CountQueuingStrategy"),
+  ReadableStream: require("./generated/ReadableStream"),
+  ReadableStreamDefaultController: require("./generated/ReadableStreamDefaultController"),
+  ReadableStreamDefaultReader: require("./generated/ReadableStreamDefaultReader"),
+  TransformStream: require("./generated/TransformStream"),
+  TransformStreamDefaultController: require("./generated/TransformStreamDefaultController"),
+  WritableStream: require("./generated/WritableStream"),
+  WritableStreamDefaultController: require("./generated/WritableStreamDefaultController"),
+  WritableStreamDefaultWriter: require("./generated/WritableStreamDefaultWriter")
 };
 
 function install(window, name, interfaceConstructor) {
@@ -196,6 +209,22 @@ function install(window, name, interfaceConstructor) {
   });
 }
 
+function defineVMGlobal(context, name) {
+  vm.runInContext(`Object.defineProperty(this, "${name}", {
+    configurable: true,
+    writable: true
+  })`, context);
+}
+
+// Predefines interface global variables in the VM, to work around
+// https://github.com/nodejs/node/issues/38918
+exports.defineVMGlobals = window => {
+  for (const generatedInterfaceName of Object.keys(generatedInterfaces)) {
+    defineVMGlobal(window, generatedInterfaceName);
+  }
+  defineVMGlobal(window, "HTMLDocument");
+};
+
 exports.installInterfaces = (window, globalNames) => {
   // Install generated interface.
   for (const generatedInterface of Object.values(generatedInterfaces)) {

lib/jsdom/living/streams/ByteLengthQueuingStrategy-impl.js

@@ -0,0 +1,27 @@
+"use strict";
+
+const sizeFunctionWeakMap = new WeakMap();
+
+exports.implementation = class ByteLengthQueuingStrategyImpl {
+  constructor(globalObject, [{ highWaterMark }]) {
+    this._globalObject = globalObject;
+    this.highWaterMark = highWaterMark;
+  }
+
+  get size() {
+    initializeSizeFunction(this._globalObject);
+    return sizeFunctionWeakMap.get(this._globalObject);
+  }
+};
+
+function initializeSizeFunction(globalObject) {
+  if (sizeFunctionWeakMap.has(globalObject)) {
+    return;
+  }
+
+  // We need to set the 'name' property:
+  // eslint-disable-next-line prefer-arrow-callback
+  sizeFunctionWeakMap.set(globalObject, function size(chunk) {
+    return chunk.byteLength;
+  });
+}

lib/jsdom/living/streams/ByteLengthQueuingStrategy.webidl

@@ -0,0 +1,7 @@
+[Exposed=(Window,Worker,Worklet)]
+interface ByteLengthQueuingStrategy {
+  constructor(QueuingStrategyInit init);
+
+  readonly attribute unrestricted double highWaterMark;
+  readonly attribute Function size;
+};

lib/jsdom/living/streams/CountQueuingStrategy-impl.js

@@ -0,0 +1,27 @@
+"use strict";
+
+const sizeFunctionWeakMap = new WeakMap();
+
+exports.implementation = class CountQueuingStrategyImpl {
+  constructor(globalObject, [{ highWaterMark }]) {
+    this._globalObject = globalObject;
+    this.highWaterMark = highWaterMark;
+  }
+
+  get size() {
+    initializeSizeFunction(this._globalObject);
+    return sizeFunctionWeakMap.get(this._globalObject);
+  }
+};
+
+function initializeSizeFunction(globalObject) {
+  if (sizeFunctionWeakMap.has(globalObject)) {
+    return;
+  }
+
+  // We need to set the 'name' property:
+  // eslint-disable-next-line prefer-arrow-callback
+  sizeFunctionWeakMap.set(globalObject, function size() {
+    return 1;
+  });
+}

lib/jsdom/living/streams/CountQueuingStrategy.webidl

@@ -0,0 +1,7 @@
+[Exposed=(Window,Worker,Worklet)]
+interface CountQueuingStrategy {
+  constructor(QueuingStrategyInit init);
+
+  readonly attribute unrestricted double highWaterMark;
+  readonly attribute Function size;
+};

lib/jsdom/living/streams/QueuingStrategy.webidl

@@ -0,0 +1,6 @@
+dictionary QueuingStrategy {
+  unrestricted double highWaterMark;
+  QueuingStrategySize size;
+};
+
+callback QueuingStrategySize = unrestricted double (optional any chunk);

lib/jsdom/living/streams/QueuingStrategyInit.webidl

@@ -0,0 +1,3 @@
+dictionary QueuingStrategyInit {
+  required unrestricted double highWaterMark;
+};

lib/jsdom/living/streams/ReadableByteStreamController-impl.js

@@ -0,0 +1,123 @@
+"use strict";
+const { CancelSteps, PullSteps } = require("./abstract-ops/internal-methods.js");
+const { resetQueue } = require("./abstract-ops/queue-with-sizes.js");
+const aos = require("./abstract-ops/readable-streams.js");
+
+const ReadableStreamBYOBRequest = require("../generated/ReadableStreamBYOBRequest.js");
+
+exports.implementation = class ReadableByteStreamControllerImpl {
+  get byobRequest() {
+    if (this._byobRequest === null && this._pendingPullIntos.length > 0) {
+      const firstDescriptor = this._pendingPullIntos[0];
+      const view = new this._globalObject.Uint8Array(
+        firstDescriptor.buffer,
+        firstDescriptor.byteOffset + firstDescriptor.bytesFilled,
+        firstDescriptor.byteLength - firstDescriptor.bytesFilled
+      );
+
+      const byobRequest = ReadableStreamBYOBRequest.new(this._globalObject);
+      byobRequest._controller = this;
+      byobRequest._view = view;
+      this._byobRequest = byobRequest;
+    }
+
+    return this._byobRequest;
+  }
+
+  get desiredSize() {
+    return aos.readableByteStreamControllerGetDesiredSize(this);
+  }
+
+  close() {
+    if (this._closeRequested === true) {
+      throw new TypeError("The stream has already been closed; do not close it again!");
+    }
+
+    const state = this._stream._state;
+    if (state !== "readable") {
+      throw new TypeError(`The stream (in ${state} state) is not in the readable state and cannot be closed`);
+    }
+
+    aos.readableByteStreamControllerClose(this);
+  }
+
+  enqueue(chunk) {
+    if (chunk.byteLength === 0) {
+      throw new TypeError("chunk must have non-zero byteLength");
+    }
+    if (chunk.buffer.byteLength === 0) {
+      throw new TypeError("chunk's buffer must have non-zero byteLength");
+    }
+
+    if (this._closeRequested === true) {
+      throw new TypeError("stream is closed or draining");
+    }
+
+    const state = this._stream._state;
+    if (state !== "readable") {
+      throw new TypeError(`The stream (in ${state} state) is not in the readable state and cannot be enqueued to`);
+    }
+
+    aos.readableByteStreamControllerEnqueue(this._globalObject, this, chunk);
+  }
+
+  error(e) {
+    aos.readableByteStreamControllerError(this, e);
+  }
+
+  [CancelSteps](reason) {
+    aos.readableByteStreamControllerClearPendingPullIntos(this);
+
+    resetQueue(this);
+
+    const result = this._cancelAlgorithm(reason);
+    aos.readableByteStreamControllerClearAlgorithms(this);
+    return result;
+  }
+
+  [PullSteps](readRequest) {
+    const stream = this._stream;
+    // Assert: aos.readableStreamHasDefaultReader(stream) === true
+
+    if (this._queueTotalSize > 0) {
+      // Assert: aos.readableStreamGetNumReadRequests(stream) === 0
+
+      const entry = this._queue.shift();
+      this._queueTotalSize -= entry.byteLength;
+
+      aos.readableByteStreamControllerHandleQueueDrain(this);
+
+      const view = new this._globalObject.Uint8Array(entry.buffer, entry.byteOffset, entry.byteLength);
+
+      readRequest.chunkSteps(view);
+      return;
+    }
+
+    const autoAllocateChunkSize = this._autoAllocateChunkSize;
+    if (autoAllocateChunkSize !== undefined) {
+      let buffer;
+      try {
+        buffer = new this._globalObject.ArrayBuffer(autoAllocateChunkSize);
+      } catch (bufferE) {
+        readRequest.errorSteps(bufferE);
+        return;
+      }
+
+      const pullIntoDescriptor = {
+        buffer,
+        bufferByteLength: autoAllocateChunkSize,
+        byteOffset: 0,
+        byteLength: autoAllocateChunkSize,
+        bytesFilled: 0,
+        elementSize: 1,
+        ViewConstructor: this._globalObject.Uint8Array,
+        readerType: "default"
+      };
+
+      this._pendingPullIntos.push(pullIntoDescriptor);
+    }
+
+    aos.readableStreamAddReadRequest(stream, readRequest);
+    aos.readableByteStreamControllerCallPullIfNeeded(this);
+  }
+};

lib/jsdom/living/streams/ReadableByteStreamController.webidl

@@ -0,0 +1,9 @@
+[Exposed=(Window,Worker,Worklet)]
+interface ReadableByteStreamController {
+  readonly attribute ReadableStreamBYOBRequest? byobRequest;
+  readonly attribute unrestricted double? desiredSize;
+
+  void close();
+  void enqueue(ArrayBufferView chunk);
+  void error(optional any e);
+};