1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492
// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.// Copyright Joyent and Node contributors. All rights reserved. MIT license.// deno-lint-ignore-file
import { addAbortSignalNoValidate } from "./add-abort-signal.mjs";import { Buffer } from "../../buffer.ts";import { debuglog } from "../util/debuglog.ts";import { getDefaultHighWaterMark, getHighWaterMark } from "./state.mjs";import { prependListener, Stream } from "./legacy.mjs";import { StringDecoder } from "../../string_decoder.ts";import { validateBoolean, validateObject } from "../validators.mjs";import {  ERR_INVALID_ARG_TYPE,  ERR_INVALID_ARG_VALUE,  ERR_METHOD_NOT_IMPLEMENTED,  ERR_STREAM_PUSH_AFTER_EOF,  ERR_STREAM_UNSHIFT_AFTER_END_EVENT,} from "../errors.ts";import { nextTick } from "../../_next_tick.ts";import { stdio } from "../../_process/stdio.mjs";import _from from "./from.mjs";import BufferList from "./buffer_list.mjs";import destroyImpl from "./destroy.mjs";import EE from "../../events.ts";import { isReadableEnded } from "./utils.mjs";
let debug = debuglog("stream", (fn) => {  debug = fn;});
const kPaused = Symbol("kPaused");
Object.setPrototypeOf(Readable.prototype, Stream.prototype);Object.setPrototypeOf(Readable, Stream);const nop = () => {};
const { errorOrDestroy } = destroyImpl;
function ReadableState(options, stream, isDuplex) {  // Duplex streams are both readable and writable, but share  // the same options object.  // However, some cases require setting options to different  // values for the readable and the writable sides of the duplex stream.  // These options can be provided separately as readableXXX and writableXXX.  if (typeof isDuplex !== "boolean") {    isDuplex = stream instanceof Stream.Duplex;  }
  // Object stream flag. Used to make read(n) ignore n and to  // make all the buffer merging and length checks go away.  this.objectMode = !!(options && options.objectMode);
  if (isDuplex) {    this.objectMode = this.objectMode ||      !!(options && options.readableObjectMode);  }
  // The point at which it stops calling _read() to fill the buffer  // Note: 0 is a valid value, means "don't call _read preemptively ever"  this.highWaterMark = options    ? getHighWaterMark(this, options, "readableHighWaterMark", isDuplex)    : getDefaultHighWaterMark(false);
  // A linked list is used to store data chunks instead of an array because the  // linked list can remove elements from the beginning faster than  // array.shift().  this.buffer = new BufferList();  this.length = 0;  this.pipes = [];  this.flowing = null;  this.ended = false;  this.endEmitted = false;  this.reading = false;
  // Stream is still being constructed and cannot be  // destroyed until construction finished or failed.  // Async construction is opt in, therefore we start as  // constructed.  this.constructed = true;
  // A flag to be able to tell if the event 'readable'/'data' is emitted  // immediately, or on a later tick.  We set this to true at first, because  // any actions that shouldn't happen until "later" should generally also  // not happen before the first read call.  this.sync = true;
  // Whenever we return null, then we set a flag to say  // that we're awaiting a 'readable' event emission.  this.needReadable = false;  this.emittedReadable = false;  this.readableListening = false;  this.resumeScheduled = false;  this[kPaused] = null;
  // True if the error was already emitted and should not be thrown again.  this.errorEmitted = false;
  // Should close be emitted on destroy. Defaults to true.  this.emitClose = !options || options.emitClose !== false;
  // Should .destroy() be called after 'end' (and potentially 'finish').  this.autoDestroy = !options || options.autoDestroy !== false;
  // Has it been destroyed.  this.destroyed = false;
  // Indicates whether the stream has errored. When true no further  // _read calls, 'data' or 'readable' events should occur. This is needed  // since when autoDestroy is disabled we need a way to tell whether the  // stream has failed.  this.errored = null;
  // Indicates whether the stream has finished destroying.  this.closed = false;
  // True if close has been emitted or would have been emitted  // depending on emitClose.  this.closeEmitted = false;
  // Crypto is kind of old and crusty.  Historically, its default string  // encoding is 'binary' so we have to make this configurable.  // Everything else in the universe uses 'utf8', though.  this.defaultEncoding = (options && options.defaultEncoding) || "utf8";
  // Ref the piped dest which we need a drain event on it  // type: null | Writable | Set<Writable>.  this.awaitDrainWriters = null;  this.multiAwaitDrain = false;
  // If true, a maybeReadMore has been scheduled.  this.readingMore = false;
  this.dataEmitted = false;
  this.decoder = null;  this.encoding = null;  if (options && options.encoding) {    this.decoder = new StringDecoder(options.encoding);    this.encoding = options.encoding;  }}
function Readable(options) {  if (!(this instanceof Readable)) {    return new Readable(options);  }
  // Checking for a Stream.Duplex instance is faster here instead of inside  // the ReadableState constructor, at least with V8 6.5.  const isDuplex = this instanceof Stream.Duplex;
  this._readableState = new ReadableState(options, this, isDuplex);
  if (options) {    if (typeof options.read === "function") {      this._read = options.read;    }
    if (typeof options.destroy === "function") {      this._destroy = options.destroy;    }
    if (typeof options.construct === "function") {      this._construct = options.construct;    }    if (options.signal && !isDuplex) {      addAbortSignalNoValidate(options.signal, this);    }  }
  Stream.call(this, options);
  destroyImpl.construct(this, () => {    if (this._readableState.needReadable) {      maybeReadMore(this, this._readableState);    }  });}
Readable.prototype.destroy = destroyImpl.destroy;Readable.prototype._undestroy = destroyImpl.undestroy;Readable.prototype._destroy = function (err, cb) {  cb(err);};
Readable.prototype[EE.captureRejectionSymbol] = function (err) {  this.destroy(err);};
// Manually shove something into the read() buffer.// This returns true if the highWaterMark has not been hit yet,// similar to how Writable.write() returns true if you should// write() some more.Readable.prototype.push = function (chunk, encoding) {  return readableAddChunk(this, chunk, encoding, false);};
// Unshift should *always* be something directly out of read().Readable.prototype.unshift = function (chunk, encoding) {  return readableAddChunk(this, chunk, encoding, true);};
function readableAddChunk(stream, chunk, encoding, addToFront) {  debug("readableAddChunk", chunk);  const state = stream._readableState;
  let err;  if (!state.objectMode) {    if (typeof chunk === "string") {      encoding = encoding || state.defaultEncoding;      if (state.encoding !== encoding) {        if (addToFront && state.encoding) {          // When unshifting, if state.encoding is set, we have to save          // the string in the BufferList with the state encoding.          chunk = Buffer.from(chunk, encoding).toString(state.encoding);        } else {          chunk = Buffer.from(chunk, encoding);          encoding = "";        }      }    } else if (chunk instanceof Buffer) {      encoding = "";    } else if (Stream._isUint8Array(chunk)) {      chunk = Stream._uint8ArrayToBuffer(chunk);      encoding = "";    } else if (chunk != null) {      err = new ERR_INVALID_ARG_TYPE(        "chunk",        ["string", "Buffer", "Uint8Array"],        chunk,      );    }  }
  if (err) {    errorOrDestroy(stream, err);  } else if (chunk === null) {    state.reading = false;    onEofChunk(stream, state);  } else if (state.objectMode || (chunk && 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 && !encoding) {        chunk = state.decoder.write(chunk);        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);  }
  // We can push more data if we are below the highWaterMark.  // Also, if we have no data yet, we can stand some more bytes.  // This is to work around cases where hwm=0, such as the repl.  return !state.ended &&    (state.length < state.highWaterMark || state.length === 0);}
function addChunk(stream, state, chunk, addToFront) {  if (    state.flowing && state.length === 0 && !state.sync &&    stream.listenerCount("data") > 0  ) {    // Use the guard to avoid creating `Set()` repeatedly    // when we have multiple pipes.    if (state.multiAwaitDrain) {      state.awaitDrainWriters.clear();    } else {      state.awaitDrainWriters = null;    }    state.dataEmitted = true;    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);}
Readable.prototype.isPaused = function () {  const state = this._readableState;  return state[kPaused] === true || state.flowing === false;};
// Backwards compatibility.Readable.prototype.setEncoding = function (enc) {  const decoder = new StringDecoder(enc);  this._readableState.decoder = decoder;  // If setEncoding(null), decoder.encoding equals utf8.  this._readableState.encoding = this._readableState.decoder.encoding;
  const buffer = this._readableState.buffer;  // Iterate over current buffer to convert already stored Buffers:  let content = "";  for (const data of buffer) {    content += decoder.write(data);  }  buffer.clear();  if (content !== "") {    buffer.push(content);  }  this._readableState.length = content.length;  return this;};
// Don't raise the hwm > 1GB.const MAX_HWM = 0x40000000;function computeNewHighWaterMark(n) {  if (n >= MAX_HWM) {    // TODO(ronag): Throw ERR_VALUE_OUT_OF_RANGE.    n = MAX_HWM;  } else {    // Get the next highest power of 2 to prevent increasing hwm excessively in    // tiny amounts.    n--;    n |= n >>> 1;    n |= n >>> 2;    n |= n >>> 4;    n |= n >>> 8;    n |= n >>> 16;    n++;  }  return n;}
// This function is designed to be inlinable, so please take care when making// changes to the function body.function howMuchToRead(n, state) {  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;}
// You can override either this method, or the async _read(n) below.Readable.prototype.read = function (n) {  debug("read", n);  // Same as parseInt(undefined, 10), however V8 7.3 performance regressed  // in this scenario, so we are doing it manually.  if (n === undefined) {    n = NaN;  } else if (!Number.isInteger(n)) {    n = Number.parseInt(n, 10);  }  const state = this._readableState;  const nOrig = n;
  // If we're asking for more than the current hwm, then raise the hwm.  if (n > state.highWaterMark) {    state.highWaterMark = computeNewHighWaterMark(n);  }
  if (n !== 0) {    state.emittedReadable = false;  }
  // If we're doing read(0) to trigger a readable event, but we  // already have a bunch of data in the buffer, then just trigger  // the 'readable' event and move on.  if (    n === 0 &&    state.needReadable &&    ((state.highWaterMark !== 0      ? state.length >= state.highWaterMark      : state.length > 0) ||      state.ended)  ) {    debug("read: emitReadable", state.length, state.ended);    if (state.length === 0 && state.ended) {      endReadable(this);    } else {      emitReadable(this);    }    return null;  }
  n = howMuchToRead(n, state);
  // If we've ended, and we're now clear, then finish it up.  if (n === 0 && state.ended) {    if (state.length === 0) {      endReadable(this);    }    return null;  }
  // All the actual chunk generation logic needs to be  // *below* the call to _read.  The reason is that in certain  // synthetic stream cases, such as passthrough streams, _read  // may be a completely synchronous operation which may change  // the state of the read buffer, providing enough data when  // before there was *not* enough.  //  // So, the steps are:  // 1. Figure out what the state of things will be after we do  // a read from the buffer.  //  // 2. If that resulting state will trigger a _read, then call _read.  // Note that this may be asynchronous, or synchronous.  Yes, it is  // deeply ugly to write APIs this way, but that still doesn't mean  // that the Readable class should behave improperly, as streams are  // designed to be sync/async agnostic.  // Take note if the _read call is sync or async (ie, if the read call  // has returned yet), so that we know whether or not it's safe to emit  // 'readable' etc.  //  // 3. Actually pull the requested chunks out of the buffer and return.
  // if we need a readable event, then we need to do some reading.  let doRead = state.needReadable;  debug("need readable", doRead);
  // If we currently have less than the highWaterMark, then also read some.  if (state.length === 0 || state.length - n < state.highWaterMark) {    doRead = true;    debug("length less than watermark", doRead);  }
  // However, if we've ended, then there's no point, if we're already  // reading, then it's unnecessary, if we're constructing we have to wait,  // and if we're destroyed or errored, then it's not allowed,  if (    state.ended || state.reading || state.destroyed || state.errored ||    !state.constructed  ) {    doRead = false;    debug("reading, ended or constructing", doRead);  } else if (doRead) {    debug("do read");    state.reading = true;    state.sync = true;    // If the length is currently zero, then we *need* a readable event.    if (state.length === 0) {      state.needReadable = true;    }
    // Call internal read method    this._read(state.highWaterMark);
    state.sync = false;    // If _read pushed data synchronously, then `reading` will be false,    // and we need to re-evaluate how much data we can return to the user.    if (!state.reading) {      n = howMuchToRead(nOrig, state);    }  }
  let ret;  if (n > 0) {    ret = fromList(n, state);  } else {    ret = null;  }
  if (ret === null) {    state.needReadable = state.length <= state.highWaterMark;    n = 0;  } else {    state.length -= n;    if (state.multiAwaitDrain) {      state.awaitDrainWriters.clear();    } else {      state.awaitDrainWriters = null;    }  }
  if (state.length === 0) {    // If we have nothing in the buffer, then we want to know    // as soon as we *do* get something into the buffer.    if (!state.ended) {      state.needReadable = true;    }
    // If we tried to read() past the EOF, then emit end on the next tick.    if (nOrig !== n && state.ended) {      endReadable(this);    }  }
  if (ret !== null) {    state.dataEmitted = true;    this.emit("data", ret);  }
  return ret;};
function onEofChunk(stream, state) {  debug("onEofChunk");  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) {    // If we are sync, wait until next tick to emit the data.    // Otherwise we risk emitting data in the flow()    // the readable code triggers during a read() call.    emitReadable(stream);  } else {    // Emit 'readable' now to make sure it gets picked up.    state.needReadable = false;    state.emittedReadable = true;    // We have to emit readable now that we are EOF. Modules    // in the ecosystem (e.g. dicer) rely on this event being sync.    emitReadable_(stream);  }}
// Don't emit readable right away in sync mode, because this can trigger// another read() call => stack overflow.  This way, it might trigger// a nextTick recursion warning, but that's not so bad.function emitReadable(stream) {  const state = stream._readableState;  debug("emitReadable", state.needReadable, state.emittedReadable);  state.needReadable = false;  if (!state.emittedReadable) {    debug("emitReadable", state.flowing);    state.emittedReadable = true;    nextTick(emitReadable_, stream);  }}
function emitReadable_(stream) {  const state = stream._readableState;  debug("emitReadable_", state.destroyed, state.length, state.ended);  if (!state.destroyed && !state.errored && (state.length || state.ended)) {    stream.emit("readable");    state.emittedReadable = false;  }
  // The stream needs another readable event if:  // 1. It is not flowing, as the flow mechanism will take  //    care of it.  // 2. It is not ended.  // 3. It is below the highWaterMark, so we can schedule  //    another readable later.  state.needReadable = !state.flowing &&    !state.ended &&    state.length <= state.highWaterMark;  flow(stream);}
// At this point, the user has presumably seen the 'readable' event,// and called read() to consume some data.  that may have triggered// in turn another _read(n) call, in which case reading = true if// it's in progress.// However, if we're not ended, or reading, and the length < hwm,// then go ahead and try to read some more preemptively.function maybeReadMore(stream, state) {  if (!state.readingMore && state.constructed) {    state.readingMore = true;    nextTick(maybeReadMore_, stream, state);  }}
function maybeReadMore_(stream, state) {  // Attempt to read more data if we should.  //  // The conditions for reading more data are (one of):  // - Not enough data buffered (state.length < state.highWaterMark). The loop  //   is responsible for filling the buffer with enough data if such data  //   is available. If highWaterMark is 0 and we are not in the flowing mode  //   we should _not_ attempt to buffer any extra data. We'll get more data  //   when the stream consumer calls read() instead.  // - No data in the buffer, and the stream is in flowing mode. In this mode  //   the loop below is responsible for ensuring read() is called. Failing to  //   call read here would abort the flow and there's no other mechanism for  //   continuing the flow if the stream consumer has just subscribed to the  //   'data' event.  //  // In addition to the above conditions to keep reading data, the following  // conditions prevent the data from being read:  // - The stream has ended (state.ended).  // - There is already a pending 'read' operation (state.reading). This is a  //   case where the stream has called the implementation defined _read()  //   method, but they are processing the call asynchronously and have _not_  //   called push() with new data. In this case we skip performing more  //   read()s. The execution ends in this method again after the _read() ends  //   up calling push() with more data.  while (    !state.reading && !state.ended &&    (state.length < state.highWaterMark ||      (state.flowing && state.length === 0))  ) {    const len = state.length;    debug("maybeReadMore read 0");    stream.read(0);    if (len === state.length) {      // Didn't get any data, stop spinning.      break;    }  }  state.readingMore = false;}
// Abstract method.  to be overridden in specific implementation classes.// call cb(er, data) where data is <= n in length.// for virtual (non-string, non-buffer) streams, "length" is somewhat// arbitrary, and perhaps not very meaningful.Readable.prototype._read = function (n) {  throw new ERR_METHOD_NOT_IMPLEMENTED("_read()");};
Readable.prototype.pipe = function (dest, pipeOpts) {  const src = this;  const state = this._readableState;
  if (state.pipes.length === 1) {    if (!state.multiAwaitDrain) {      state.multiAwaitDrain = true;      state.awaitDrainWriters = new Set(        state.awaitDrainWriters ? [state.awaitDrainWriters] : [],      );    }  }
  state.pipes.push(dest);  debug("pipe count=%d opts=%j", state.pipes.length, pipeOpts);
  const doEnd = (!pipeOpts || pipeOpts.end !== false) &&    dest !== stdio.stdout &&    dest !== stdio.stderr;
  const endFn = doEnd ? onend : unpipe;  if (state.endEmitted) {    nextTick(endFn);  } else {    src.once("end", endFn);  }
  dest.on("unpipe", onunpipe);  function onunpipe(readable, unpipeInfo) {    debug("onunpipe");    if (readable === src) {      if (unpipeInfo && unpipeInfo.hasUnpiped === false) {        unpipeInfo.hasUnpiped = true;        cleanup();      }    }  }
  function onend() {    debug("onend");    dest.end();  }
  let ondrain;
  let cleanedUp = false;  function cleanup() {    debug("cleanup");    // Cleanup event handlers once the pipe is broken.    dest.removeListener("close", onclose);    dest.removeListener("finish", onfinish);    if (ondrain) {      dest.removeListener("drain", ondrain);    }    dest.removeListener("error", onerror);    dest.removeListener("unpipe", onunpipe);    src.removeListener("end", onend);    src.removeListener("end", unpipe);    src.removeListener("data", ondata);
    cleanedUp = true;
    // If the reader is waiting for a drain event from this    // specific writer, then it would cause it to never start    // flowing again.    // So, if this is awaiting a drain, then we just call it now.    // If we don't know, then assume that we are waiting for one.    if (      ondrain && state.awaitDrainWriters &&      (!dest._writableState || dest._writableState.needDrain)    ) {      ondrain();    }  }
  function pause() {    // If the user unpiped during `dest.write()`, it is possible    // to get stuck in a permanently paused state if that write    // also returned false.    // => Check whether `dest` is still a piping destination.    if (!cleanedUp) {      if (state.pipes.length === 1 && state.pipes[0] === dest) {        debug("false write response, pause", 0);        state.awaitDrainWriters = dest;        state.multiAwaitDrain = false;      } else if (state.pipes.length > 1 && state.pipes.includes(dest)) {        debug("false write response, pause", state.awaitDrainWriters.size);        state.awaitDrainWriters.add(dest);      }      src.pause();    }    if (!ondrain) {      // When the dest drains, it reduces the awaitDrain counter      // on the source.  This would be more elegant with a .once()      // handler in flow(), but adding and removing repeatedly is      // too slow.      ondrain = pipeOnDrain(src, dest);      dest.on("drain", ondrain);    }  }
  src.on("data", ondata);  function ondata(chunk) {    debug("ondata");    const ret = dest.write(chunk);    debug("dest.write", ret);    if (ret === false) {      pause();    }  }
  // If the dest has an error, then stop piping into it.  // However, don't suppress the throwing behavior for this.  function onerror(er) {    debug("onerror", er);    unpipe();    dest.removeListener("error", onerror);    if (EE.listenerCount(dest, "error") === 0) {      const s = dest._writableState || dest._readableState;      if (s && !s.errorEmitted) {        // User incorrectly emitted 'error' directly on the stream.        errorOrDestroy(dest, er);      } else {        dest.emit("error", er);      }    }  }
  // Make sure our error handler is attached before userland ones.  prependListener(dest, "error", onerror);
  // Both close and finish should trigger unpipe, but only once.  function onclose() {    dest.removeListener("finish", onfinish);    unpipe();  }  dest.once("close", onclose);  function onfinish() {    debug("onfinish");    dest.removeListener("close", onclose);    unpipe();  }  dest.once("finish", onfinish);
  function unpipe() {    debug("unpipe");    src.unpipe(dest);  }
  // Tell the dest that it's being piped to.  dest.emit("pipe", src);
  // Start the flow if it hasn't been started already.
  if (dest.writableNeedDrain === true) {    if (state.flowing) {      pause();    }  } else if (!state.flowing) {    debug("pipe resume");    src.resume();  }
  return dest;};
function pipeOnDrain(src, dest) {  return function pipeOnDrainFunctionResult() {    const state = src._readableState;
    // `ondrain` will call directly,    // `this` maybe not a reference to dest,    // so we use the real dest here.    if (state.awaitDrainWriters === dest) {      debug("pipeOnDrain", 1);      state.awaitDrainWriters = null;    } else if (state.multiAwaitDrain) {      debug("pipeOnDrain", state.awaitDrainWriters.size);      state.awaitDrainWriters.delete(dest);    }
    if (      (!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) &&      EE.listenerCount(src, "data")    ) {      state.flowing = true;      flow(src);    }  };}
Readable.prototype.unpipe = function (dest) {  const state = this._readableState;  const unpipeInfo = { hasUnpiped: false };
  // If we're not piping anywhere, then do nothing.  if (state.pipes.length === 0) {    return this;  }
  if (!dest) {    // remove all.    const dests = state.pipes;    state.pipes = [];    this.pause();
    for (let i = 0; i < dests.length; i++) {      dests[i].emit("unpipe", this, { hasUnpiped: false });    }    return this;  }
  // Try to find the right one.  const index = state.pipes.indexOf(dest);  if (index === -1) {    return this;  }
  state.pipes.splice(index, 1);  if (state.pipes.length === 0) {    this.pause();  }
  dest.emit("unpipe", this, unpipeInfo);
  return this;};
// Set up data events if they are asked for// Ensure readable listeners eventually get something.Readable.prototype.on = function (ev, fn) {  const res = Stream.prototype.on.call(this, ev, fn);  const state = this._readableState;
  if (ev === "data") {    // Update readableListening so that resume() may be a no-op    // a few lines down. This is needed to support once('readable').    state.readableListening = this.listenerCount("readable") > 0;
    // Try start flowing on next tick if stream isn't explicitly paused.    if (state.flowing !== false) {      this.resume();    }  } else if (ev === "readable") {    if (!state.endEmitted && !state.readableListening) {      state.readableListening = state.needReadable = true;      state.flowing = false;      state.emittedReadable = false;      debug("on readable", state.length, state.reading);      if (state.length) {        emitReadable(this);      } else if (!state.reading) {        nextTick(nReadingNextTick, this);      }    }  }
  return res;};Readable.prototype.addListener = Readable.prototype.on;
Readable.prototype.removeListener = function (ev, fn) {  const res = Stream.prototype.removeListener.call(this, ev, fn);
  if (ev === "readable") {    // We need to check if there is someone still listening to    // readable and reset the state. However this needs to happen    // after readable has been emitted but before I/O (nextTick) to    // support once('readable', fn) cycles. This means that calling    // resume within the same tick will have no    // effect.    nextTick(updateReadableListening, this);  }
  return res;};Readable.prototype.off = Readable.prototype.removeListener;
Readable.prototype.removeAllListeners = function (ev) {  const res = Stream.prototype.removeAllListeners.apply(this, arguments);
  if (ev === "readable" || ev === undefined) {    // We need to check if there is someone still listening to    // readable and reset the state. However this needs to happen    // after readable has been emitted but before I/O (nextTick) to    // support once('readable', fn) cycles. This means that calling    // resume within the same tick will have no    // effect.    nextTick(updateReadableListening, this);  }
  return res;};
function updateReadableListening(self) {  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;  }}
function nReadingNextTick(self) {  debug("readable nexttick read 0");  self.read(0);}
// pause() and resume() are remnants of the legacy readable stream API// If the user uses them, then switch into old mode.Readable.prototype.resume = function () {  const state = this._readableState;  if (!state.flowing) {    debug("resume");    // We flow only if there is no one listening    // for readable, but we still have to call    // resume().    state.flowing = !state.readableListening;    resume(this, state);  }  state[kPaused] = false;  return this;};
function resume(stream, state) {  if (!state.resumeScheduled) {    state.resumeScheduled = true;    nextTick(resume_, stream, state);  }}
function resume_(stream, state) {  debug("resume", state.reading);  if (!state.reading) {    stream.read(0);  }
  state.resumeScheduled = false;  stream.emit("resume");  flow(stream);  if (state.flowing && !state.reading) {    stream.read(0);  }}
Readable.prototype.pause = function () {  debug("call pause flowing=%j", this._readableState.flowing);  if (this._readableState.flowing !== false) {    debug("pause");    this._readableState.flowing = false;    this.emit("pause");  }  this._readableState[kPaused] = true;  return this;};
function flow(stream) {  const state = stream._readableState;  debug("flow", state.flowing);  while (state.flowing && stream.read() !== null);}
// Wrap an old-style stream as the async data source.// This is *not* part of the readable stream interface.// It is an ugly unfortunate mess of history.Readable.prototype.wrap = function (stream) {  let paused = false;
  // TODO (ronag): Should this.destroy(err) emit  // 'error' on the wrapped stream? Would require  // a static factory method, e.g. Readable.wrap(stream).
  stream.on("data", (chunk) => {    if (!this.push(chunk) && stream.pause) {      paused = true;      stream.pause();    }  });
  stream.on("end", () => {    this.push(null);  });
  stream.on("error", (err) => {    errorOrDestroy(this, err);  });
  stream.on("close", () => {    this.destroy();  });
  stream.on("destroy", () => {    this.destroy();  });
  this._read = () => {    if (paused && stream.resume) {      paused = false;      stream.resume();    }  };
  // Proxy all the other methods. Important when wrapping filters and duplexes.  const streamKeys = Object.keys(stream);  for (let j = 1; j < streamKeys.length; j++) {    const i = streamKeys[j];    if (this[i] === undefined && typeof stream[i] === "function") {      this[i] = stream[i].bind(stream);    }  }
  return this;};
Readable.prototype[Symbol.asyncIterator] = function () {  return streamToAsyncIterator(this);};
Readable.prototype.iterator = function (options) {  if (options !== undefined) {    validateObject(options, "options");  }  return streamToAsyncIterator(this, options);};
function streamToAsyncIterator(stream, options) {  if (typeof stream.read !== "function") {    stream = Readable.wrap(stream, { objectMode: true });  }
  const iter = createAsyncIterator(stream, options);  iter.stream = stream;  return iter;}
async function* createAsyncIterator(stream, options) {  let callback = nop;
  const opts = {    destroyOnReturn: true,    destroyOnError: true,    ...options,  };
  function next(resolve) {    if (this === stream) {      callback();      callback = nop;    } else {      callback = resolve;    }  }
  const state = stream._readableState;
  let error = state.errored;  let errorEmitted = state.errorEmitted;  let endEmitted = state.endEmitted;  let closeEmitted = state.closeEmitted;
  stream    .on("readable", next)    .on("error", function (err) {      error = err;      errorEmitted = true;      next.call(this);    })    .on("end", function () {      endEmitted = true;      next.call(this);    })    .on("close", function () {      closeEmitted = true;      next.call(this);    });
  let errorThrown = false;  try {    while (true) {      const chunk = stream.destroyed ? null : stream.read();      if (chunk !== null) {        yield chunk;      } else if (errorEmitted) {        throw error;      } else if (endEmitted) {        break;      } else if (closeEmitted) {        break;      } else {        await new Promise(next);      }    }  } catch (err) {    if (opts.destroyOnError) {      destroyImpl.destroyer(stream, err);    }    errorThrown = true;    throw err;  } finally {    if (!errorThrown && opts.destroyOnReturn) {      if (state.autoDestroy || !endEmitted) {        // TODO(ronag): ERR_PREMATURE_CLOSE?        destroyImpl.destroyer(stream, null);      }    }  }}
// Making it explicit these properties are not enumerable// because otherwise some prototype manipulation in// userland will fail.Object.defineProperties(Readable.prototype, {  readable: {    get() {      const r = this._readableState;      // r.readable === false means that this is part of a Duplex stream      // where the readable side was disabled upon construction.      // Compat. The user might manually disable readable side through      // deprecated setter.      return !!r && r.readable !== false && !r.destroyed && !r.errorEmitted &&        !r.endEmitted;    },    set(val) {      // Backwards compat.      if (this._readableState) {        this._readableState.readable = !!val;      }    },  },
  readableDidRead: {    enumerable: false,    get: function () {      return this._readableState.dataEmitted;    },  },
  readableAborted: {    enumerable: false,    get: function () {      return !!(this._readableState.destroyed || this._readableState.errored) &&        !this._readableState.endEmitted;    },  },
  readableHighWaterMark: {    enumerable: false,    get: function () {      return this._readableState.highWaterMark;    },  },
  readableBuffer: {    enumerable: false,    get: function () {      return this._readableState && this._readableState.buffer;    },  },
  readableFlowing: {    enumerable: false,    get: function () {      return this._readableState.flowing;    },    set: function (state) {      if (this._readableState) {        this._readableState.flowing = state;      }    },  },
  readableLength: {    enumerable: false,    get() {      return this._readableState.length;    },  },
  readableObjectMode: {    enumerable: false,    get() {      return this._readableState ? this._readableState.objectMode : false;    },  },
  readableEncoding: {    enumerable: false,    get() {      return this._readableState ? this._readableState.encoding : null;    },  },
  destroyed: {    enumerable: false,    get() {      if (this._readableState === undefined) {        return false;      }      return this._readableState.destroyed;    },    set(value) {      // We ignore the value if the stream      // has not been initialized yet.      if (!this._readableState) {        return;      }
      // Backward compatibility, the user is explicitly      // managing destroyed.      this._readableState.destroyed = value;    },  },
  readableEnded: {    enumerable: false,    get() {      return this._readableState ? this._readableState.endEmitted : false;    },  },});
Object.defineProperties(ReadableState.prototype, {  // Legacy getter for `pipesCount`.  pipesCount: {    get() {      return this.pipes.length;    },  },
  // Legacy property for `paused`.  paused: {    get() {      return this[kPaused] !== false;    },    set(value) {      this[kPaused] = !!value;    },  },});
// 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.function fromList(n, state) {  // nothing buffered.  if (state.length === 0) {    return null;  }
  let ret;  if (state.objectMode) {    ret = state.buffer.shift();  } else if (!n || n >= state.length) {    // Read it all, truncate the list.    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 {    // read part of list.    ret = state.buffer.consume(n, state.decoder);  }
  return ret;}
function endReadable(stream) {  const state = stream._readableState;
  debug("endReadable", state.endEmitted);  if (!state.endEmitted) {    state.ended = true;    nextTick(endReadableNT, state, stream);  }}
function endReadableNT(state, stream) {  debug("endReadableNT", state.endEmitted, state.length);
  // Check that we didn't get one last unshift.  if (    !state.errorEmitted && !state.closeEmitted &&    !state.endEmitted && state.length === 0  ) {    state.endEmitted = true;    stream.emit("end");
    if (stream.writable && stream.allowHalfOpen === false) {      nextTick(endWritableNT, stream);    } else if (state.autoDestroy) {      // In case of duplex streams we need a way to detect      // if the writable side is ready for autoDestroy as well.      const wState = stream._writableState;      const autoDestroy = !wState || (        wState.autoDestroy &&        // We don't expect the writable to ever 'finish'        // if writable is explicitly set to false.        (wState.finished || wState.writable === false)      );
      if (autoDestroy) {        stream.destroy();      }    }  }}
function endWritableNT(stream) {  const writable = stream.writable && !stream.writableEnded &&    !stream.destroyed;  if (writable) {    stream.end();  }}
function readableFrom(iterable, opts) {  return _from(Readable, iterable, opts);}
function isReadableStream(object) {  return object instanceof ReadableStream;}
export const fromWeb = Readable.fromWeb = function (  readableStream,  options = {},) {  if (!isReadableStream(readableStream)) {    throw new ERR_INVALID_ARG_TYPE(      "readableStream",      "ReadableStream",      readableStream,    );  }
  validateObject(options, "options");  const {    highWaterMark,    encoding,    objectMode = false,    signal,  } = options;
  if (encoding !== undefined && !Buffer.isEncoding(encoding)) {    throw new ERR_INVALID_ARG_VALUE(encoding, "options.encoding");  }  validateBoolean(objectMode, "options.objectMode");
  const reader = readableStream.getReader();  let closed = false;
  const readable = new Readable({    objectMode,    highWaterMark,    encoding,    signal,
    read() {      reader.read().then(        (chunk) => {          if (chunk.done) {            readable.push(null);          } else {            readable.push(chunk.value);          }        },        (error) => destroy.call(readable, error),      );    },
    destroy(error, callback) {      function done() {        try {          callback(error);        } catch (error) {          // In a next tick because this is happening within          // a promise context, and if there are any errors          // thrown we don't want those to cause an unhandled          // rejection. Let's just escape the promise and          // handle it separately.          process.nextTick(() => {            throw error;          });        }      }
      if (!closed) {        reader.cancel(error).then(done, done);        return;      }
      done();    },  });
  reader.closed.then(    () => {      closed = true;      if (!isReadableEnded(readable)) {        readable.push(null);      }    },    (error) => {      closed = true;      destroy.call(readable, error);    },  );
  return readable;};
function wrap(src, options) {  return new Readable({    objectMode: src.readableObjectMode ?? src.objectMode ?? true,    ...options,    destroy(err, callback) {      destroyImpl.destroyer(src, err);      callback(err);    },  }).wrap(src);}
// Exposed for testing purposes only.Readable._fromList = fromList;Readable.ReadableState = ReadableState;Readable.from = readableFrom;Readable.wrap = wrap;
export default Readable;export { fromList as _fromList, readableFrom as from, ReadableState, wrap };