stream: copyedit `webstreams/adapter.js` · nodejs/node@7792f1a

const {

ArrayPrototypeFilter,

ArrayPrototypeMap,

Boolean,

ObjectEntries,

ObjectKeys,

PromisePrototypeThen,

PromiseResolve,

PromiseWithResolvers,

SafePromiseAll,

SafePromiseAllReturnVoid,

SafePromisePrototypeFinally,

SafeSet,

StringPrototypeStartsWith,

const {

kEmptyObject,

normalizeEncoding,

setOwnProperty,

} = require('internal/util');

const {

const { eos } = require('internal/streams/end-of-stream');

const { zlib } = internalBinding('constants');

const { UV_EOF } = internalBinding('uv');

const encoder = new TextEncoder();

const kDestroyOnSyncError = Symbol('kDestroyOnSyncError');

// Collect all negative (error) ZLIB codes and Z_NEED_DICT

const ZLIB_FAILURES = new SafeSet([

...ArrayPrototypeFilter(

ArrayPrototypeMap(

ObjectEntries(internalBinding('constants').zlib),

({ 0: code, 1: value }) => (value < 0 ? code : null),

),

Boolean,

const ZLIB_FAILURES = new SafeSet(

ArrayPrototypeFilter(

ObjectKeys(zlib),

(code) => code === 'Z_NEED_DICT' || zlib[code] < 0,

),

'Z_NEED_DICT',

]);

);

/**

* @param {Error|null} cause

* @returns {Error|null}

*/

function handleKnownInternalErrors(cause) {

const causeCode = cause?.code;

switch (true) {

case cause?.code === 'ERR_STREAM_PREMATURE_CLOSE': {

case causeCode === 'ERR_STREAM_PREMATURE_CLOSE': {

return new AbortError(undefined, { cause });

}

case ZLIB_FAILURES.has(cause?.code):

case ZLIB_FAILURES.has(causeCode):

// Brotli decoder error codes are formatted as 'ERR_' +

// BrotliDecoderErrorString(), where the latter returns strings like

// '_ERROR_FORMAT_...', '_ERROR_ALLOC_...', '_ERROR_UNREACHABLE', etc.

// The resulting JS error codes all start with 'ERR__ERROR_'.

// Falls through

case cause?.code != null &&

StringPrototypeStartsWith(cause.code, 'ERR__ERROR_'): {

case causeCode != null &&

StringPrototypeStartsWith(causeCode, 'ERR__ERROR_'): {

// eslint-disable-next-line no-restricted-syntax

const error = new TypeError(undefined, { cause });

error.code = cause.code;

setOwnProperty(error, 'code', causeCode);

return error;

}

default:

let closed;

function onDrain() {

if (backpressurePromise !== undefined)

backpressurePromise.resolve();

backpressurePromise?.resolve();

}

const cleanup = eos(streamWritable, (error) => {

// that happen to emit an error event again after finished is called.

streamWritable.on('error', () => {});

if (error != null) {

if (backpressurePromise !== undefined)

backpressurePromise.reject(error);

backpressurePromise?.reject(error);

// If closed is not undefined, the error is happening

// after the WritableStream close has already started.

// We need to reject it here.

writer.ready,

() => {

return PromisePrototypeThen(

SafePromiseAll(

SafePromiseAllReturnVoid(

chunks,

(data) => writer.write(data.chunk)),

() => done(),

done,

done);

},

done);

writer.ready,

() => {

return PromisePrototypeThen(

SafePromiseAll(

SafePromiseAllReturnVoid(

chunks,

(data) => writer.write(data.chunk)),

() => done(),

done,

done);

},

done);

if (!writableClosed || !readableClosed) {

PromisePrototypeThen(

SafePromiseAll([

SafePromiseAllReturnVoid([

closeWriter(),

closeReader(),

]),