123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439
// Copyright Node.js contributors. All rights reserved. MIT License.import { Buffer } from "../buffer.ts";import type Duplex from "./duplex.ts";import type EventEmitter from "../events.ts";import type Readable from "./readable.ts";import type Writable from "./writable.ts";import type { ReadableState } from "./readable.ts";import { kPaused } from "./symbols.ts";import {  ERR_STREAM_PUSH_AFTER_EOF,  ERR_STREAM_UNSHIFT_AFTER_END_EVENT,} from "../_errors.ts";
export function _destroy(  self: Readable,  err?: Error | null,  cb?: (error?: Error | null) => void,) {  self._destroy(err || null, (err) => {    const r = (self as Readable)._readableState;
    if (err) {      // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364      err.stack;
      if (!r.errored) {        r.errored = err;      }    }
    r.closed = true;
    if (typeof cb === "function") {      cb(err);    }
    if (err) {      queueMicrotask(() => {        if (!r.errorEmitted) {          r.errorEmitted = true;          self.emit("error", err);        }        r.closeEmitted = true;        if (r.emitClose) {          self.emit("close");        }      });    } else {      queueMicrotask(() => {        r.closeEmitted = true;        if (r.emitClose) {          self.emit("close");        }      });    }  });}
export function addChunk(  stream: Duplex | Readable,  state: ReadableState,  chunk: string | Buffer | Uint8Array,  addToFront: boolean,) {  if (state.flowing && state.length === 0 && !state.sync) {    if (state.multiAwaitDrain) {      (state.awaitDrainWriters as Set<Writable>).clear();    } else {      state.awaitDrainWriters = null;    }    stream.emit("data", chunk);  } else {    // Update the buffer info.    state.length += state.objectMode ? 1 : chunk.length;    if (addToFront) {      state.buffer.unshift(chunk);    } else {      state.buffer.push(chunk);    }
    if (state.needReadable) {      emitReadable(stream);    }  }  maybeReadMore(stream, state);}
// Don't raise the hwm > 1GB.const MAX_HWM = 0x40000000;export function computeNewHighWaterMark(n: number) {  if (n >= MAX_HWM) {    n = MAX_HWM;  } else {    n--;    n |= n >>> 1;    n |= n >>> 2;    n |= n >>> 4;    n |= n >>> 8;    n |= n >>> 16;    n++;  }  return n;}
export function emitReadable(stream: Duplex | Readable) {  const state = stream._readableState;  state.needReadable = false;  if (!state.emittedReadable) {    state.emittedReadable = true;    queueMicrotask(() => emitReadable_(stream));  }}
function emitReadable_(stream: Duplex | Readable) {  const state = stream._readableState;  if (!state.destroyed && !state.errored && (state.length || state.ended)) {    stream.emit("readable");    state.emittedReadable = false;  }
  state.needReadable = !state.flowing &&    !state.ended &&    state.length <= state.highWaterMark;  flow(stream);}
export function endReadable(stream: Readable) {  const state = stream._readableState;
  if (!state.endEmitted) {    state.ended = true;    queueMicrotask(() => endReadableNT(state, stream));  }}
function endReadableNT(state: ReadableState, stream: Readable) {  if (    !state.errorEmitted && !state.closeEmitted &&    !state.endEmitted && state.length === 0  ) {    state.endEmitted = true;    stream.emit("end");
    if (state.autoDestroy) {      stream.destroy();    }  }}
export function errorOrDestroy(  stream: Duplex | Readable,  err: Error,  sync = false,) {  const r = stream._readableState;
  if (r.destroyed) {    return stream;  }
  if (r.autoDestroy) {    stream.destroy(err);  } else if (err) {    // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364    err.stack;
    if (!r.errored) {      r.errored = err;    }    if (sync) {      queueMicrotask(() => {        if (!r.errorEmitted) {          r.errorEmitted = true;          stream.emit("error", err);        }      });    } else if (!r.errorEmitted) {      r.errorEmitted = true;      stream.emit("error", err);    }  }}
function flow(stream: Duplex | Readable) {  const state = stream._readableState;  while (state.flowing && stream.read() !== null);}
/** Pluck off n bytes from an array of buffers.* Length is the combined lengths of all the buffers in the list.* This function is designed to be inlinable, so please take care when making* changes to the function body.*/export function fromList(n: number, state: ReadableState) {  // nothing buffered.  if (state.length === 0) {    return null;  }
  let ret;  if (state.objectMode) {    ret = state.buffer.shift();  } else if (!n || n >= state.length) {    if (state.decoder) {      ret = state.buffer.join("");    } else if (state.buffer.length === 1) {      ret = state.buffer.first();    } else {      ret = state.buffer.concat(state.length);    }    state.buffer.clear();  } else {    ret = state.buffer.consume(n, !!state.decoder);  }
  return ret;}
export function howMuchToRead(n: number, state: ReadableState) {  if (n <= 0 || (state.length === 0 && state.ended)) {    return 0;  }  if (state.objectMode) {    return 1;  }  if (Number.isNaN(n)) {    // Only flow one buffer at a time.    if (state.flowing && state.length) {      return state.buffer.first().length;    }    return state.length;  }  if (n <= state.length) {    return n;  }  return state.ended ? state.length : 0;}
export function maybeReadMore(stream: Readable, state: ReadableState) {  if (!state.readingMore && state.constructed) {    state.readingMore = true;    queueMicrotask(() => maybeReadMore_(stream, state));  }}
function maybeReadMore_(stream: Readable, state: ReadableState) {  while (    !state.reading && !state.ended &&    (state.length < state.highWaterMark ||      (state.flowing && state.length === 0))  ) {    const len = state.length;    stream.read(0);    if (len === state.length) {      // Didn't get any data, stop spinning.      break;    }  }  state.readingMore = false;}
export function nReadingNextTick(self: Duplex | Readable) {  self.read(0);}
export function onEofChunk(stream: Duplex | Readable, state: ReadableState) {  if (state.ended) return;  if (state.decoder) {    const chunk = state.decoder.end();    if (chunk && chunk.length) {      state.buffer.push(chunk);      state.length += state.objectMode ? 1 : chunk.length;    }  }  state.ended = true;
  if (state.sync) {    emitReadable(stream);  } else {    state.needReadable = false;    state.emittedReadable = true;    emitReadable_(stream);  }}
export function pipeOnDrain(src: Duplex | Readable, dest: Duplex | Writable) {  return function pipeOnDrainFunctionResult() {    const state = src._readableState;
    if (state.awaitDrainWriters === dest) {      state.awaitDrainWriters = null;    } else if (state.multiAwaitDrain) {      (state.awaitDrainWriters as Set<Duplex | Writable>).delete(dest);    }
    if (      (!state.awaitDrainWriters ||        (state.awaitDrainWriters as Set<Writable>).size === 0) &&      src.listenerCount("data")    ) {      state.flowing = true;      flow(src);    }  };}
export function prependListener(  emitter: EventEmitter,  event: string,  // deno-lint-ignore no-explicit-any  fn: (...args: any[]) => any,) {  if (typeof emitter.prependListener === "function") {    return emitter.prependListener(event, fn);  }
  // This is a hack to make sure that our error handler is attached before any  // userland ones.  NEVER DO THIS. This is here only because this code needs  // to continue to work with older versions of Node.js that do not include  //the prependListener() method. The goal is to eventually remove this hack.  // TODO(Soremwar)  // Burn it with fire  // deno-lint-ignore ban-ts-comment  //@ts-ignore  if (emitter._events.get(event)?.length) {    // deno-lint-ignore ban-ts-comment    //@ts-ignore    const listeners = [fn, ...emitter._events.get(event)];    // deno-lint-ignore ban-ts-comment    //@ts-ignore    emitter._events.set(event, listeners);  } else {    emitter.on(event, fn);  }}
export function readableAddChunk(  stream: Duplex | Readable,  chunk: string | Buffer | Uint8Array | null,  encoding: undefined | string = undefined,  addToFront: boolean,) {  const state = stream._readableState;  let usedEncoding = encoding;
  let err;  if (!state.objectMode) {    if (typeof chunk === "string") {      usedEncoding = encoding || state.defaultEncoding;      if (state.encoding !== usedEncoding) {        if (addToFront && state.encoding) {          chunk = Buffer.from(chunk, usedEncoding).toString(state.encoding);        } else {          chunk = Buffer.from(chunk, usedEncoding);          usedEncoding = "";        }      }    } else if (chunk instanceof Uint8Array) {      chunk = Buffer.from(chunk);    }  }
  if (err) {    errorOrDestroy(stream, err);  } else if (chunk === null) {    state.reading = false;    onEofChunk(stream, state);  } else if (state.objectMode || (chunk.length > 0)) {    if (addToFront) {      if (state.endEmitted) {        errorOrDestroy(stream, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());      } else {        addChunk(stream, state, chunk, true);      }    } else if (state.ended) {      errorOrDestroy(stream, new ERR_STREAM_PUSH_AFTER_EOF());    } else if (state.destroyed || state.errored) {      return false;    } else {      state.reading = false;      if (state.decoder && !usedEncoding) {        //TODO(Soremwar)        //I don't think this cast is right        chunk = state.decoder.write(Buffer.from(chunk as Uint8Array));        if (state.objectMode || chunk.length !== 0) {          addChunk(stream, state, chunk, false);        } else {          maybeReadMore(stream, state);        }      } else {        addChunk(stream, state, chunk, false);      }    }  } else if (!addToFront) {    state.reading = false;    maybeReadMore(stream, state);  }
  return !state.ended &&    (state.length < state.highWaterMark || state.length === 0);}
export function resume(stream: Duplex | Readable, state: ReadableState) {  if (!state.resumeScheduled) {    state.resumeScheduled = true;    queueMicrotask(() => resume_(stream, state));  }}
function resume_(stream: Duplex | Readable, state: ReadableState) {  if (!state.reading) {    stream.read(0);  }
  state.resumeScheduled = false;  stream.emit("resume");  flow(stream);  if (state.flowing && !state.reading) {    stream.read(0);  }}
export function updateReadableListening(self: Duplex | Readable) {  const state = self._readableState;  state.readableListening = self.listenerCount("readable") > 0;
  if (state.resumeScheduled && state[kPaused] === false) {    // Flowing needs to be set to true now, otherwise    // the upcoming resume will not flow.    state.flowing = true;
    // Crude way to check if we should resume.  } else if (self.listenerCount("data") > 0) {    self.resume();  } else if (!state.readableListening) {    state.flowing = null;  }}