quic: fix a handful of bugs and missing functionality · nodejs/node@7f3a85e

// There really should only be one read at a time so using an

// array here is purely defensive.

this.pendingPulls = [];

// Register a wakeup callback that the C++ side can invoke

// when new data is available after a STATUS_BLOCK.

reader.setWakeup(() => {

if (this.pendingPulls.length > 0) {

this.readNext(c);

}

});

},

pull(c) {

const { promise, resolve, reject } = PromiseWithResolvers();

this.pendingPulls.push({ resolve, reject });

const readNext = () => {

reader.pull((status, buffer) => {

// If pendingPulls is empty here, the stream had to have

// been canceled, and we don't really care about the result.

// We can simply exit.

if (this.pendingPulls.length === 0) {

return;

}

if (status === 0) {

// EOS

c.close();

// This is to signal the end for byob readers

// see https://streams.spec.whatwg.org/#example-rbs-pull

c.byobRequest?.respond(0);

const pending = this.pendingPulls.shift();

pending.resolve();

return;

} else if (status < 0) {

// The read could fail for many different reasons when reading

// from a non-memory resident blob part (e.g. file-backed blob).

// The error details the system error code.

const error = lazyDOMException('The blob could not be read', 'NotReadableError');

const pending = this.pendingPulls.shift();

c.error(error);

pending.reject(error);

this.readNext(c);

return promise;

},

readNext(c) {

reader.pull((status, buffer) => {

// If pendingPulls is empty here, the stream had to have

// been canceled, and we don't really care about the result.

// We can simply exit.

if (this.pendingPulls.length === 0) {

return;

}

if (status === 0) {

// EOS

c.close();

// This is to signal the end for byob readers

// see https://streams.spec.whatwg.org/#example-rbs-pull

c.byobRequest?.respond(0);

const pending = this.pendingPulls.shift();

pending.resolve();

return;

} else if (status < 0) {

// The read could fail for many different reasons when reading

// from a non-memory resident blob part (e.g. file-backed blob).

// The error details the system error code.

const error =

lazyDOMException('The blob could not be read',

'NotReadableError');

const pending = this.pendingPulls.shift();

c.error(error);

pending.reject(error);

return;

} else if (status === 2) {

// STATUS_BLOCK: No data available yet. The wakeup callback

// registered in start() will re-invoke readNext when data

// arrives.

return;

}

// ReadableByteStreamController.enqueue errors if we submit a

// 0-length buffer. We need to check for that here.

if (buffer !== undefined && buffer.byteLength !== 0) {

c.enqueue(new Uint8Array(buffer));

}

// We keep reading until we either reach EOS, some error, or

// we hit the flow rate of the stream (c.desiredSize).

// We use setImmediate here because we have to allow the event

// loop to turn in order to process any pending i/o. Using

// queueMicrotask won't allow the event loop to turn.

setImmediate(() => {

if (c.desiredSize < 0) {

// A manual backpressure check.

if (this.pendingPulls.length !== 0) {

const pending = this.pendingPulls.shift();

pending.resolve();

}

return;

}

// ReadableByteStreamController.enqueue errors if we submit a 0-length

// buffer. We need to check for that here.

if (buffer !== undefined && buffer.byteLength !== 0) {

c.enqueue(new Uint8Array(buffer));

}

// We keep reading until we either reach EOS, some error, or we

// hit the flow rate of the stream (c.desiredSize).

// We use set immediate here because we have to allow the event

// loop to turn in order to process any pending i/o. Using

// queueMicrotask won't allow the event loop to turn.

setImmediate(() => {

if (c.desiredSize < 0) {

// A manual backpressure check.

if (this.pendingPulls.length !== 0) {

// A case of waiting pull finished (= not yet canceled)

const pending = this.pendingPulls.shift();

pending.resolve();

}

return;

}

readNext();

});

this.readNext(c);

});

};

readNext();

return promise;

});

},

cancel(reason) {

// Reject any currently pending pulls here.