stream: implement `min` option for `ReadableStreamBYOBReader.read` · nodejs/node@4a3ecbf

SymbolAsyncIterator,

SymbolDispose,

SymbolToStringTag,

TypedArrayPrototypeGetLength,

Uint8Array,

} = primordials;

ERR_INVALID_ARG_TYPE,

ERR_INVALID_STATE,

ERR_INVALID_THIS,

ERR_OUT_OF_RANGE,

},

} = require('internal/errors');

validateAbortSignal,

validateBuffer,

validateObject,

kValidateObjectAllowNullable,

kValidateObjectAllowFunction,

kValidateObjectAllowObjects,

kValidateObjectAllowObjectsAndNull,

} = require('internal/validators');

const {

* @param {UnderlyingSource} [source]

* @param {QueuingStrategy} [strategy]

*/

constructor(source = {}, strategy = kEmptyObject) {

if (source === null)

throw new ERR_INVALID_ARG_VALUE('source', 'Object', source);

constructor(source = kEmptyObject, strategy = kEmptyObject) {

validateObject(source, 'source', kValidateObjectAllowObjects);

validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull);

this[kState] = createReadableStreamState();

this[kIsClosedPromise] = createDeferredPromise();

getReader(options = kEmptyObject) {

if (!isReadableStream(this))

throw new ERR_INVALID_THIS('ReadableStream');

validateObject(options, 'options', kValidateObjectAllowNullable | kValidateObjectAllowFunction);

validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);

const mode = options?.mode;

if (mode === undefined)

// The web platform tests require that these be handled one at a

// time and in a specific order. options can be null or undefined.

validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);

const preventAbort = options?.preventAbort;

const preventCancel = options?.preventCancel;

const preventClose = options?.preventClose;

destination);

}

validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);

const preventAbort = options?.preventAbort;

const preventCancel = options?.preventCancel;

const preventClose = options?.preventClose;

values(options = kEmptyObject) {

if (!isReadableStream(this))

throw new ERR_INVALID_THIS('ReadableStream');

validateObject(options, 'options');

const {

preventCancel = false,

} = options;

validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);

const preventCancel = !!(options?.preventCancel);

// eslint-disable-next-line no-use-before-define

const reader = new ReadableStreamDefaultReader(this);

/**

* @param {ArrayBufferView} view

* @param {{

* min? : number

* }} [options]

* @returns {Promise<{

* view : ArrayBufferView,

* value : ArrayBufferView,

* done : boolean,

* }>}

*/

read(view) {

async read(view, options = kEmptyObject) {

if (!isReadableStreamBYOBReader(this))

return PromiseReject(new ERR_INVALID_THIS('ReadableStreamBYOBReader'));

throw new ERR_INVALID_THIS('ReadableStreamBYOBReader');

if (!isArrayBufferView(view)) {

return PromiseReject(

new ERR_INVALID_ARG_TYPE(

'view',

[

'Buffer',

'TypedArray',

'DataView',

],

view));

throw new ERR_INVALID_ARG_TYPE(

'view',

[

'Buffer',

'TypedArray',

'DataView',

],

view,

);

}

validateObject(options, 'options', kValidateObjectAllowObjectsAndNull);

const viewByteLength = ArrayBufferViewGetByteLength(view);

const viewBuffer = ArrayBufferViewGetBuffer(view);

const viewBufferByteLength = ArrayBufferPrototypeGetByteLength(viewBuffer);

if (viewByteLength === 0 || viewBufferByteLength === 0) {

return PromiseReject(

new ERR_INVALID_STATE.TypeError(

'View or Viewed ArrayBuffer is zero-length or detached',

),

);

throw new ERR_INVALID_STATE.TypeError(

'View or Viewed ArrayBuffer is zero-length or detached');

}

// Supposed to assert here that the view's buffer is not

// detached, but there's no API available to use to check that.

const min = options?.min ?? 1;

if (typeof min !== 'number')

throw new ERR_INVALID_ARG_TYPE('options.min', 'number', min);

if (!NumberIsInteger(min))

throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be an integer');

if (min <= 0)

throw new ERR_INVALID_ARG_VALUE('options.min', min, 'must be greater than 0');

if (!isDataView(view)) {

if (min > TypedArrayPrototypeGetLength(view)) {

throw new ERR_OUT_OF_RANGE('options.min', '<= view.length', min);

}

} else if (min > viewByteLength) {

throw new ERR_OUT_OF_RANGE('options.min', '<= view.byteLength', min);

}

if (this[kState].stream === undefined) {

return PromiseReject(

new ERR_INVALID_STATE.TypeError(

'The reader is not attached to a stream'));

throw new ERR_INVALID_STATE.TypeError('The reader is not attached to a stream');

}

const readIntoRequest = new ReadIntoRequest();

readableStreamBYOBReaderRead(this, view, readIntoRequest);

readableStreamBYOBReaderRead(this, view, min, readIntoRequest);

return readIntoRequest.promise;

}

reading = false;

},

};

readableStreamBYOBReaderRead(reader, view, readIntoRequest);

readableStreamBYOBReaderRead(reader, view, 1, readIntoRequest);

}

function pull1Algorithm() {

reader[kState].stream = undefined;

}

function readableStreamBYOBReaderRead(reader, view, readIntoRequest) {

function readableStreamBYOBReaderRead(reader, view, min, readIntoRequest) {

const {

stream,

} = reader[kState];

readableByteStreamControllerPullInto(

stream[kState].controller,

view,

min,

readIntoRequest);

}

if (pendingPullIntos.length) {

const firstPendingPullInto = pendingPullIntos[0];

if (firstPendingPullInto.bytesFilled > 0) {

if (firstPendingPullInto.bytesFilled % firstPendingPullInto.elementSize !== 0) {

const error = new ERR_INVALID_STATE.TypeError('Partial read');

readableByteStreamControllerError(controller, error);

throw error;

let done = false;

if (stream[kState].state === 'closed') {

desc.bytesFilled = 0;

assert(desc.bytesFilled % desc.elementSize === 0);

done = true;

}

function readableByteStreamControllerPullInto(

controller,

view,

min,

readIntoRequest) {

const {

closeRequested,

elementSize = view.constructor.BYTES_PER_ELEMENT;

ctor = view.constructor;

}

const minimumFill = min * elementSize;

assert(minimumFill >= elementSize && minimumFill <= view.byteLength);

assert(minimumFill % elementSize === 0);

const buffer = ArrayBufferViewGetBuffer(view);

const byteOffset = ArrayBufferViewGetByteOffset(view);

const byteLength = ArrayBufferViewGetByteLength(view);

byteOffset,

byteLength,

bytesFilled: 0,

minimumFill,

elementSize,

ctor,

type: 'byob',

}

function readableByteStreamControllerRespondInClosedState(controller, desc) {

assert(!desc.bytesFilled);

assert(desc.bytesFilled % desc.elementSize === 0);

if (desc.type === 'none') {

readableByteStreamControllerShiftPendingPullInto(controller);

}

byteLength,

byteOffset,

bytesFilled,

minimumFill,

elementSize,

} = desc;

const currentAlignedBytes = bytesFilled - (bytesFilled % elementSize);

const maxBytesToCopy = MathMin(

controller[kState].queueTotalSize,

byteLength - bytesFilled);

const maxBytesFilled = bytesFilled + maxBytesToCopy;

const maxAlignedBytes = maxBytesFilled - (maxBytesFilled % elementSize);

let totalBytesToCopyRemaining = maxBytesToCopy;

let ready = false;

if (maxAlignedBytes > currentAlignedBytes) {

assert(bytesFilled < minimumFill);

if (maxAlignedBytes >= minimumFill) {

totalBytesToCopyRemaining = maxAlignedBytes - bytesFilled;

ready = true;

}

if (!ready) {

assert(!controller[kState].queueTotalSize);

assert(desc.bytesFilled > 0);

assert(desc.bytesFilled < elementSize);

assert(desc.bytesFilled < minimumFill);

}

return ready;

}

return;

}

if (desc.bytesFilled < desc.elementSize)

if (desc.bytesFilled < desc.minimumFill)

return;

readableByteStreamControllerShiftPendingPullInto(controller);

byteOffset: 0,

byteLength: autoAllocateChunkSize,

bytesFilled: 0,

minimumFill: 1,

elementSize: 1,

ctor: Uint8Array,

type: 'default',