streams: fixes for webstreams

lib/internal/webstreams/readablestream.js

@@ -11,7 +11,6 @@ const {
   FunctionPrototypeCall,
   MathMin,
   NumberIsInteger,
-  ObjectCreate,
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
   Promise,
@@ -95,9 +94,9 @@ const {
   AsyncIterator,
   cloneAsUint8Array,
   copyArrayBuffer,
+  createPromiseCallback,
   customInspect,
   dequeueValue,
-  ensureIsPromise,
   enqueueValueWithSize,
   extractHighWaterMark,
   extractSizeAlgorithm,
@@ -251,19 +250,7 @@ class ReadableStream {
     markTransferMode(this, false, true);
     if (source === null)
       throw new ERR_INVALID_ARG_VALUE('source', 'Object', source);
-    this[kState] = {
-      disturbed: false,
-      reader: undefined,
-      state: 'readable',
-      storedError: undefined,
-      stream: undefined,
-      transfer: {
-        writable: undefined,
-        port1: undefined,
-        port2: undefined,
-        promise: undefined,
-      },
-    };
+    this[kState] = createReadableStreamState();
 
     this[kIsClosedPromise] = createDeferredPromise();
     this[kControllerErrorFunction] = () => {};
@@ -647,19 +634,7 @@ ObjectDefineProperties(ReadableStream, {
 function InternalTransferredReadableStream() {
   markTransferMode(this, false, true);
   this[kType] = 'ReadableStream';
-  this[kState] = {
-    disturbed: false,
-    reader: undefined,
-    state: 'readable',
-    storedError: undefined,
-    stream: undefined,
-    transfer: {
-      writable: undefined,
-      port1: undefined,
-      port2: undefined,
-      promise: undefined,
-    },
-  };
+  this[kState] = createReadableStreamState();
 
   this[kIsClosedPromise] = createDeferredPromise();
 }
@@ -1228,41 +1203,58 @@ ObjectDefineProperties(ReadableByteStreamController.prototype, {
   [SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name),
 });
 
-function TeeReadableStream(start, pull, cancel) {
+function InternalReadableStream(start, pull, cancel, highWaterMark, size) {
   markTransferMode(this, false, true);
   this[kType] = 'ReadableStream';
-  this[kState] = {
-    disturbed: false,
-    state: 'readable',
-    storedError: undefined,
-    stream: undefined,
-    transfer: {
-      writable: undefined,
-      port: undefined,
-      promise: undefined,
-    },
-  };
+  this[kState] = createReadableStreamState();
+  this[kIsClosedPromise] = createDeferredPromise();
+  const controller = new ReadableStreamDefaultController(kSkipThrow);
+  setupReadableStreamDefaultController(
+    this,
+    controller,
+    start,
+    pull,
+    cancel,
+    highWaterMark,
+    size);
+}
+
+ObjectSetPrototypeOf(InternalReadableStream.prototype, ReadableStream.prototype);
+ObjectSetPrototypeOf(InternalReadableStream, ReadableStream);
+
+function createReadableStream(start, pull, cancel, highWaterMark = 1, size = () => 1) {
+  const stream = new InternalReadableStream(start, pull, cancel, highWaterMark, size);
+
+  // For spec compliance the InternalReadableStream must be a ReadableStream
+  stream.constructor = ReadableStream;
+  return stream;
+}
+
+function InternalReadableByteStream(start, pull, cancel) {
+  markTransferMode(this, false, true);
+  this[kType] = 'ReadableStream';
+  this[kState] = createReadableStreamState();
   this[kIsClosedPromise] = createDeferredPromise();
-  setupReadableStreamDefaultControllerFromSource(
+  const controller = new ReadableByteStreamController(kSkipThrow);
+  setupReadableByteStreamController(
     this,
-    ObjectCreate(null, {
-      start: { __proto__: null, value: start },
-      pull: { __proto__: null, value: pull },
-      cancel: { __proto__: null, value: cancel },
-    }),
-    1,
-    () => 1);
+    controller,
+    start,
+    pull,
+    cancel,
+    0,
+    undefined);
 }
 
-ObjectSetPrototypeOf(TeeReadableStream.prototype, ReadableStream.prototype);
-ObjectSetPrototypeOf(TeeReadableStream, ReadableStream);
+ObjectSetPrototypeOf(InternalReadableByteStream.prototype, ReadableStream.prototype);
+ObjectSetPrototypeOf(InternalReadableByteStream, ReadableStream);
 
-function createTeeReadableStream(start, pull, cancel) {
-  const tee = new TeeReadableStream(start, pull, cancel);
+function createReadableByteStream(start, pull, cancel) {
+  const stream = new InternalReadableByteStream(start, pull, cancel);
 
-  // For spec compliance the Tee must be a ReadableStream
-  tee.constructor = ReadableStream;
-  return tee;
+  // For spec compliance the InternalReadableByteStream must be a ReadableStream
+  stream.constructor = ReadableStream;
+  return stream;
 }
 
 const isReadableStream =
@@ -1278,6 +1270,23 @@ const isReadableStreamBYOBReader =
 
 // ---- ReadableStream Implementation
 
+function createReadableStreamState() {
+  return {
+    __proto__: null,
+    disturbed: false,
+    reader: undefined,
+    state: 'readable',
+    storedError: undefined,
+    transfer: {
+      __proto__: null,
+      writable: undefined,
+      port1: undefined,
+      port2: undefined,
+      promise: undefined,
+    },
+  };
+}
+
 function readableStreamFromIterable(iterable) {
   let stream;
   const iteratorRecord = getIterator(iterable, 'async');
@@ -1317,16 +1326,12 @@ function readableStreamFromIterable(iterable) {
     });
   }
 
-  stream = new ReadableStream({
-    start: startAlgorithm,
-    pull: pullAlgorithm,
-    cancel: cancelAlgorithm,
-  }, {
-    size() {
-      return 1;
-    },
-    highWaterMark: 0,
-  });
+  stream = createReadableStream(
+    startAlgorithm,
+    pullAlgorithm,
+    cancelAlgorithm,
+    0,
+  );
 
   return stream;
 }
@@ -1652,9 +1657,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
   }
 
   branch1 =
-    createTeeReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm);
+    createReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm);
   branch2 =
-    createTeeReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm);
+    createReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm);
 
   PromisePrototypeThen(
     reader[kState].close.promise,
@@ -1931,16 +1936,10 @@ function readableByteStreamTee(stream) {
     return cancelDeferred.promise;
   }
 
-  branch1 = new ReadableStream({
-    type: 'bytes',
-    pull: pull1Algorithm,
-    cancel: cancel1Algorithm,
-  });
-  branch2 = new ReadableStream({
-    type: 'bytes',
-    pull: pull2Algorithm,
-    cancel: cancel2Algorithm,
-  });
+  branch1 =
+    createReadableByteStream(nonOpStart, pull1Algorithm, cancel1Algorithm);
+  branch2 =
+    createReadableByteStream(nonOpStart, pull2Algorithm, cancel2Algorithm);
 
   forwardReaderError(reader);
 
@@ -1991,10 +1990,7 @@ function readableStreamCancel(stream, reason) {
   }
 
   return PromisePrototypeThen(
-    ensureIsPromise(
-      stream[kState].controller[kCancel],
-      stream[kState].controller,
-      reason),
+    stream[kState].controller[kCancel](reason),
     () => {});
 }
 
@@ -2359,7 +2355,7 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) {
   assert(!controller[kState].pullAgain);
   controller[kState].pulling = true;
   PromisePrototypeThen(
-    ensureIsPromise(controller[kState].pullAlgorithm, controller),
+    controller[kState].pullAlgorithm(controller),
     () => {
       controller[kState].pulling = false;
       if (controller[kState].pullAgain) {
@@ -2389,12 +2385,9 @@ function readableStreamDefaultControllerError(controller, error) {
 
 function readableStreamDefaultControllerCancelSteps(controller, reason) {
   resetQueue(controller);
-  try {
-    const result = controller[kState].cancelAlgorithm(reason);
-    return result;
-  } finally {
-    readableStreamDefaultControllerClearAlgorithms(controller);
-  }
+  const result = controller[kState].cancelAlgorithm(reason);
+  readableStreamDefaultControllerClearAlgorithms(controller);
+  return result;
 }
 
 function readableStreamDefaultControllerPullSteps(controller, readRequest) {
@@ -2468,11 +2461,10 @@ function setupReadableStreamDefaultControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    FunctionPrototypeBind(pull, source, controller) :
+    createPromiseCallback('source.pull', pull, source) :
     nonOpPull;
-
   const cancelAlgorithm = cancel ?
-    FunctionPrototypeBind(cancel, source) :
+    createPromiseCallback('source.cancel', cancel, source) :
     nonOpCancel;
 
   setupReadableStreamDefaultController(
@@ -3100,7 +3092,7 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
   assert(!controller[kState].pullAgain);
   controller[kState].pulling = true;
   PromisePrototypeThen(
-    ensureIsPromise(controller[kState].pullAlgorithm, controller),
+    controller[kState].pullAlgorithm(controller),
     () => {
       controller[kState].pulling = false;
       if (controller[kState].pullAgain) {
@@ -3267,10 +3259,10 @@ function setupReadableByteStreamControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    FunctionPrototypeBind(pull, source, controller) :
+    createPromiseCallback('source.pull', pull, source, controller) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
-    FunctionPrototypeBind(cancel, source) :
+    createPromiseCallback('source.cancel', cancel, source) :
     nonOpCancel;
 
   if (autoAllocateChunkSize === 0) {
@@ -3367,4 +3359,6 @@ module.exports = {
   readableByteStreamControllerPullSteps,
   setupReadableByteStreamController,
   setupReadableByteStreamControllerFromSource,
+  createReadableStream,
+  createReadableByteStream,
 };