deno.land / std@0.91.0 / node / _stream / readable_internal.ts
نووسراو ببینە
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439// Copyright Node.js contributors. All rights reserved. MIT License.import { Buffer } from "../buffer.ts";import type Duplex from "./duplex.ts";import { 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; }}
std
Version Info
3 years ago
External Dependencies
No external dependencies 🎉
