Add ReadableStreamBYOBReader.prototype.read(view, { min })

etc/web-streams-polyfill.api.md

@@ -88,10 +88,16 @@ export class ReadableStreamBYOBReader {
     constructor(stream: ReadableStream<Uint8Array>);
     cancel(reason?: any): Promise<void>;
     get closed(): Promise<undefined>;
-    read<T extends ArrayBufferView>(view: T): Promise<ReadableStreamBYOBReadResult<T>>;
+    read<T extends ArrayBufferView>(view: T, options?: ReadableStreamBYOBReaderReadOptions): Promise<ReadableStreamBYOBReadResult<T>>;
     releaseLock(): void;
 }
 
+// @public
+export interface ReadableStreamBYOBReaderReadOptions {
+    // (undocumented)
+    min?: number;
+}
+
 // @public
 export type ReadableStreamBYOBReadResult<T extends ArrayBufferView> = {
     done: false;

src/lib/helpers/array-buffer-view.ts

@@ -27,6 +27,10 @@ function isDataViewConstructor(ctor: Function): ctor is DataViewConstructor {
   return ctor === DataView;
 }
 
+export function isDataView(view: ArrayBufferView): view is DataView {
+  return isDataViewConstructor(view.constructor);
+}
+
 export function arrayBufferViewElementSize<T extends ArrayBufferView>(ctor: ArrayBufferViewConstructor<T>): number {
   if (isDataViewConstructor(ctor)) {
     return 1;

src/lib/readable-stream.ts

@@ -56,7 +56,10 @@ import { assertObject, assertRequiredArgument } from './validators/basic';
 import { convertQueuingStrategy } from './validators/queuing-strategy';
 import { ExtractHighWaterMark, ExtractSizeAlgorithm } from './abstract-ops/queuing-strategy';
 import { convertUnderlyingDefaultOrByteSource } from './validators/underlying-source';
-import type { ReadableStreamGetReaderOptions } from './readable-stream/reader-options';
+import type {
+  ReadableStreamBYOBReaderReadOptions,
+  ReadableStreamGetReaderOptions
+} from './readable-stream/reader-options';
 import { convertReaderOptions } from './validators/reader-options';
 import type { StreamPipeOptions, ValidatedStreamPipeOptions } from './readable-stream/pipe-options';
 import type { ReadableStreamIteratorOptions } from './readable-stream/iterator-options';
@@ -371,6 +374,7 @@ export type {
   ReadableStreamAsyncIterator,
   ReadableStreamDefaultReadResult,
   ReadableStreamBYOBReadResult,
+  ReadableStreamBYOBReaderReadOptions,
   UnderlyingByteSource,
   UnderlyingSource,
   UnderlyingSourceStartCallback,

src/lib/readable-stream/byob-reader.ts

@@ -17,6 +17,12 @@ import { newPromise, promiseRejectedWith } from '../helpers/webidl';
 import { assertRequiredArgument } from '../validators/basic';
 import { assertReadableStream } from '../validators/readable-stream';
 import { IsDetachedBuffer } from '../abstract-ops/ecmascript';
+import type {
+  ReadableStreamBYOBReaderReadOptions,
+  ValidatedReadableStreamBYOBReaderReadOptions
+} from './reader-options';
+import { convertByobReadOptions } from '../validators/reader-options';
+import { isDataView, type TypedArray } from '../helpers/array-buffer-view';
 
 /**
  * A result returned by {@link ReadableStreamBYOBReader.read}.
@@ -157,7 +163,14 @@ export class ReadableStreamBYOBReader {
    *
    * If reading a chunk causes the queue to become empty, more data will be pulled from the underlying source.
    */
-  read<T extends ArrayBufferView>(view: T): Promise<ReadableStreamBYOBReadResult<T>> {
+  read<T extends ArrayBufferView>(
+    view: T,
+    options?: ReadableStreamBYOBReaderReadOptions
+  ): Promise<ReadableStreamBYOBReadResult<T>>;
+  read<T extends ArrayBufferView>(
+    view: T,
+    rawOptions: ReadableStreamBYOBReaderReadOptions | null | undefined = {}
+  ): Promise<ReadableStreamBYOBReadResult<T>> {
     if (!IsReadableStreamBYOBReader(this)) {
       return promiseRejectedWith(byobReaderBrandCheckException('read'));
     }
@@ -175,6 +188,24 @@ export class ReadableStreamBYOBReader {
       return promiseRejectedWith(new TypeError('view\'s buffer has been detached'));
     }
 
+    let options: ValidatedReadableStreamBYOBReaderReadOptions;
+    try {
+      options = convertByobReadOptions(rawOptions, 'options');
+    } catch (e) {
+      return promiseRejectedWith(e);
+    }
+    const min = options.min;
+    if (min === 0) {
+      return promiseRejectedWith(new TypeError('options.min must be greater than 0'));
+    }
+    if (!isDataView(view)) {
+      if (min > (view as unknown as TypedArray).length) {
+        return promiseRejectedWith(new RangeError('options.min must be less than or equal to view\'s length'));
+      }
+    } else if (min > view.byteLength) {
+      return promiseRejectedWith(new RangeError('options.min must be less than or equal to view\'s byteLength'));
+    }
+
     if (this._ownerReadableStream === undefined) {
       return promiseRejectedWith(readerLockException('read from'));
     }
@@ -190,7 +221,7 @@ export class ReadableStreamBYOBReader {
       _closeSteps: chunk => resolvePromise({ value: chunk, done: true }),
       _errorSteps: e => rejectPromise(e)
     };
-    ReadableStreamBYOBReaderRead(this, view, readIntoRequest);
+    ReadableStreamBYOBReaderRead(this, view, min, readIntoRequest);
     return promise;
   }
 
@@ -249,6 +280,7 @@ export function IsReadableStreamBYOBReader(x: any): x is ReadableStreamBYOBReade
 export function ReadableStreamBYOBReaderRead<T extends ArrayBufferView>(
   reader: ReadableStreamBYOBReader,
   view: T,
+  min: number,
   readIntoRequest: ReadIntoRequest<T>
 ): void {
   const stream = reader._ownerReadableStream;
@@ -263,6 +295,7 @@ export function ReadableStreamBYOBReaderRead<T extends ArrayBufferView>(
     ReadableByteStreamControllerPullInto(
       stream._readableStreamController as ReadableByteStreamController,
       view,
+      min,
       readIntoRequest
     );
   }

src/lib/readable-stream/byte-stream-controller.ts

@@ -155,6 +155,7 @@ interface DefaultPullIntoDescriptor {
   byteOffset: number;
   byteLength: number;
   bytesFilled: number;
+  minimumFill: number;
   elementSize: number;
   viewConstructor: TypedArrayConstructor<Uint8Array>;
   readerType: 'default' | 'none';
@@ -166,6 +167,7 @@ interface BYOBPullIntoDescriptor<T extends ArrayBufferView = ArrayBufferView> {
   byteOffset: number;
   byteLength: number;
   bytesFilled: number;
+  minimumFill: number;
   elementSize: number;
   viewConstructor: ArrayBufferViewConstructor<T>;
   readerType: 'byob' | 'none';
@@ -335,6 +337,7 @@ export class ReadableByteStreamController {
         byteOffset: 0,
         byteLength: autoAllocateChunkSize,
         bytesFilled: 0,
+        minimumFill: 1,
         elementSize: 1,
         viewConstructor: Uint8Array,
         readerType: 'default'
@@ -452,7 +455,7 @@ function ReadableByteStreamControllerCommitPullIntoDescriptor<T extends ArrayBuf
 
   let done = false;
   if (stream._state === 'closed') {
-    assert(pullIntoDescriptor.bytesFilled === 0);
+    assert(pullIntoDescriptor.bytesFilled % pullIntoDescriptor.elementSize === 0);
     done = true;
   }
 
@@ -516,18 +519,18 @@ function ReadableByteStreamControllerEnqueueDetachedPullIntoToQueue(controller:
 
 function ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller: ReadableByteStreamController,
                                                                      pullIntoDescriptor: PullIntoDescriptor) {
-  const elementSize = pullIntoDescriptor.elementSize;
-
-  const currentAlignedBytes = pullIntoDescriptor.bytesFilled - pullIntoDescriptor.bytesFilled % elementSize;
-
   const maxBytesToCopy = Math.min(controller._queueTotalSize,
                                   pullIntoDescriptor.byteLength - pullIntoDescriptor.bytesFilled);
   const maxBytesFilled = pullIntoDescriptor.bytesFilled + maxBytesToCopy;
-  const maxAlignedBytes = maxBytesFilled - maxBytesFilled % elementSize;
 
   let totalBytesToCopyRemaining = maxBytesToCopy;
   let ready = false;
-  if (maxAlignedBytes > currentAlignedBytes) {
+  assert(pullIntoDescriptor.bytesFilled < pullIntoDescriptor.minimumFill);
+  const remainderBytes = maxBytesFilled % pullIntoDescriptor.elementSize;
+  const maxAlignedBytes = maxBytesFilled - remainderBytes;
+  // A descriptor for a read() request that is not yet filled up to its minimum length will stay at the head
+  // of the queue, so the underlying source can keep filling it.
+  if (maxAlignedBytes >= pullIntoDescriptor.minimumFill) {
     totalBytesToCopyRemaining = maxAlignedBytes - pullIntoDescriptor.bytesFilled;
     ready = true;
   }
@@ -558,7 +561,7 @@ function ReadableByteStreamControllerFillPullIntoDescriptorFromQueue(controller:
   if (!ready) {
     assert(controller._queueTotalSize === 0);
     assert(pullIntoDescriptor.bytesFilled > 0);
-    assert(pullIntoDescriptor.bytesFilled < pullIntoDescriptor.elementSize);
+    assert(pullIntoDescriptor.bytesFilled < pullIntoDescriptor.minimumFill);
   }
 
   return ready;
@@ -630,13 +633,18 @@ function ReadableByteStreamControllerProcessReadRequestsUsingQueue(controller: R
 export function ReadableByteStreamControllerPullInto<T extends ArrayBufferView>(
   controller: ReadableByteStreamController,
   view: T,
+  min: number,
   readIntoRequest: ReadIntoRequest<T>
 ): void {
   const stream = controller._controlledReadableByteStream;
 
   const ctor = view.constructor as ArrayBufferViewConstructor<T>;
   const elementSize = arrayBufferViewElementSize(ctor);
 
+  const minimumFill = min * elementSize;
+  assert(minimumFill >= elementSize && minimumFill <= view.byteLength);
+  assert(minimumFill % elementSize === 0);
+
   // try {
   const buffer = TransferArrayBuffer(view.buffer);
   // } catch (e) {
@@ -650,6 +658,7 @@ export function ReadableByteStreamControllerPullInto<T extends ArrayBufferView>(
     byteOffset: view.byteOffset,
     byteLength: view.byteLength,
     bytesFilled: 0,
+    minimumFill,
     elementSize,
     viewConstructor: ctor,
     readerType: 'byob'
@@ -699,7 +708,7 @@ export function ReadableByteStreamControllerPullInto<T extends ArrayBufferView>(
 
 function ReadableByteStreamControllerRespondInClosedState(controller: ReadableByteStreamController,
                                                           firstDescriptor: PullIntoDescriptor) {
-  assert(firstDescriptor.bytesFilled === 0);
+  assert(firstDescriptor.bytesFilled % firstDescriptor.elementSize === 0);
 
   if (firstDescriptor.readerType === 'none') {
     ReadableByteStreamControllerShiftPendingPullInto(controller);
@@ -727,7 +736,9 @@ function ReadableByteStreamControllerRespondInReadableState(controller: Readable
     return;
   }
 
-  if (pullIntoDescriptor.bytesFilled < pullIntoDescriptor.elementSize) {
+  if (pullIntoDescriptor.bytesFilled < pullIntoDescriptor.minimumFill) {
+    // A descriptor for a read() request that is not yet filled up to its minimum length will stay at the head
+    // of the queue, so the underlying source can keep filling it.
     return;
   }
 
@@ -831,7 +842,7 @@ export function ReadableByteStreamControllerClose(controller: ReadableByteStream
 
   if (controller._pendingPullIntos.length > 0) {
     const firstPendingPullInto = controller._pendingPullIntos.peek();
-    if (firstPendingPullInto.bytesFilled > 0) {
+    if (firstPendingPullInto.bytesFilled % firstPendingPullInto.elementSize !== 0) {
       const e = new TypeError('Insufficient bytes to fill elements in the given buffer');
       ReadableByteStreamControllerError(controller, e);
 

src/lib/readable-stream/reader-options.ts

@@ -1,3 +1,15 @@
 export interface ReadableStreamGetReaderOptions {
   mode?: 'byob';
 }
+
+/**
+ * Options for {@link ReadableStreamBYOBReader.read | reading} a stream
+ * with a {@link ReadableStreamBYOBReader | BYOB reader}.
+ *
+ * @public
+ */
+export interface ReadableStreamBYOBReaderReadOptions {
+  min?: number;
+}
+
+export type ValidatedReadableStreamBYOBReaderReadOptions = Required<ReadableStreamBYOBReaderReadOptions>;

src/lib/readable-stream/tee.ts

@@ -365,7 +365,7 @@ export function ReadableByteStreamTee(stream: ReadableByteStream): [ReadableByte
         reading = false;
       }
     };
-    ReadableStreamBYOBReaderRead(reader, view, readIntoRequest);
+    ReadableStreamBYOBReaderRead(reader, view, 1, readIntoRequest);
   }
 
   function pull1Algorithm(): Promise<void> {

src/lib/validators/reader-options.ts

@@ -1,5 +1,9 @@
-import { assertDictionary } from './basic';
-import type { ReadableStreamGetReaderOptions } from '../readable-stream/reader-options';
+import { assertDictionary, convertUnsignedLongLongWithEnforceRange } from './basic';
+import type {
+  ReadableStreamBYOBReaderReadOptions,
+  ReadableStreamGetReaderOptions,
+  ValidatedReadableStreamBYOBReaderReadOptions
+} from '../readable-stream/reader-options';
 
 export function convertReaderOptions(options: ReadableStreamGetReaderOptions | null | undefined,
                                      context: string): ReadableStreamGetReaderOptions {
@@ -17,3 +21,17 @@ function convertReadableStreamReaderMode(mode: string, context: string): 'byob'
   }
   return mode;
 }
+
+export function convertByobReadOptions(
+  options: ReadableStreamBYOBReaderReadOptions | null | undefined,
+  context: string
+): ValidatedReadableStreamBYOBReaderReadOptions {
+  assertDictionary(options, context);
+  const min = options?.min ?? 1;
+  return {
+    min: convertUnsignedLongLongWithEnforceRange(
+      min,
+      `${context} has member 'min' that`
+    )
+  };
+}

src/ponyfill.ts

@@ -3,6 +3,7 @@ import {
   ReadableStream,
   type ReadableStreamAsyncIterator,
   ReadableStreamBYOBReader,
+  type ReadableStreamBYOBReaderReadOptions,
   type ReadableStreamBYOBReadResult,
   ReadableStreamBYOBRequest,
   ReadableStreamDefaultController,
@@ -56,6 +57,7 @@ export {
   type StreamPipeOptions,
   type ReadableStreamDefaultReadResult,
   type ReadableStreamBYOBReadResult,
+  type ReadableStreamBYOBReaderReadOptions,
   ReadableStreamDefaultController,
   ReadableByteStreamController,
   ReadableStreamBYOBRequest,

test/wpt/shared/exclusions.js

@@ -47,7 +47,7 @@ const ignoredFailuresMinified = {
   'idlharness.any.html': [
     // Terser turns `(a = undefined) => {}` into `(a) => {}`, changing the function's length property
     // Therefore we cannot correctly implement methods with optional arguments
-    /interface: operation (abort|cancel|enqueue|error|getReader|write)/,
+    /interface: operation (abort|cancel|enqueue|error|getReader|read|write)/,
     // Same thing for ReadableStream.values(), which is tested as part of the async iterable declaration
     'ReadableStream interface: async iterable<any>'
   ]