import { captureRejectionSymbol } from "../events.ts";import Readable, { ReadableState } from "./readable.ts";import Stream from "./stream.ts";import Writable, { WritableState } from "./writable.ts";import { Buffer } from "../buffer.ts";import { ERR_STREAM_ALREADY_FINISHED, ERR_STREAM_DESTROYED, ERR_UNKNOWN_ENCODING,} from "../_errors.ts";import type { Encodings } from "../_utils.ts";import createReadableStreamAsyncIterator from "./async_iterator.ts";import type { ReadableStreamAsyncIterator } from "./async_iterator.ts";import { _destroy, computeNewHighWaterMark, emitReadable, fromList, howMuchToRead, nReadingNextTick, updateReadableListening,} from "./readable_internal.ts";import { kOnFinished, writeV } from "./writable_internal.ts";import { endDuplex, finishMaybe, onwrite, readableAddChunk,} from "./duplex_internal.ts";export { errorOrDestroy } from "./duplex_internal.ts";
export interface DuplexOptions { allowHalfOpen?: boolean; autoDestroy?: boolean; decodeStrings?: boolean; defaultEncoding?: Encodings; destroy?( this: Duplex, error: Error | null, callback: (error: Error | null) => void, ): void; emitClose?: boolean; encoding?: Encodings; final?(this: Duplex, callback: (error?: Error | null) => void): void; highWaterMark?: number; objectMode?: boolean; read?(this: Duplex, size: number): void; readable?: boolean; readableHighWaterMark?: number; readableObjectMode?: boolean; writable?: boolean; writableCorked?: number; writableHighWaterMark?: number; writableObjectMode?: boolean; write?( this: Duplex, chunk: any, encoding: Encodings, callback: (error?: Error | null) => void, ): void; writev?: writeV;}
interface Duplex extends Readable, Writable {}
class Duplex extends Stream { allowHalfOpen = true; _final?: ( callback: (error?: Error | null | undefined) => void, ) => void; _readableState: ReadableState; _writableState: WritableState; _writev?: writeV | null;
constructor(options?: DuplexOptions) { super();
if (options) { if (options.allowHalfOpen === false) { this.allowHalfOpen = false; } if (typeof options.destroy === "function") { this._destroy = options.destroy; } if (typeof options.final === "function") { this._final = options.final; } if (typeof options.read === "function") { this._read = options.read; } if (options.readable === false) { this.readable = false; } if (options.writable === false) { this.writable = false; } if (typeof options.write === "function") { this._write = options.write; } if (typeof options.writev === "function") { this._writev = options.writev; } }
const readableOptions = { autoDestroy: options?.autoDestroy, defaultEncoding: options?.defaultEncoding, destroy: options?.destroy as unknown as ( this: Readable, error: Error | null, callback: (error: Error | null) => void, ) => void, emitClose: options?.emitClose, encoding: options?.encoding, highWaterMark: options?.highWaterMark ?? options?.readableHighWaterMark, objectMode: options?.objectMode ?? options?.readableObjectMode, read: options?.read as unknown as (this: Readable) => void, };
const writableOptions = { autoDestroy: options?.autoDestroy, decodeStrings: options?.decodeStrings, defaultEncoding: options?.defaultEncoding, destroy: options?.destroy as unknown as ( this: Writable, error: Error | null, callback: (error: Error | null) => void, ) => void, emitClose: options?.emitClose, final: options?.final as unknown as ( this: Writable, callback: (error?: Error | null) => void, ) => void, highWaterMark: options?.highWaterMark ?? options?.writableHighWaterMark, objectMode: options?.objectMode ?? options?.writableObjectMode, write: options?.write as unknown as ( this: Writable, chunk: any, encoding: string, callback: (error?: Error | null) => void, ) => void, writev: options?.writev as unknown as ( this: Writable, chunks: Array<{ chunk: any; encoding: Encodings }>, callback: (error?: Error | null) => void, ) => void, };
this._readableState = new ReadableState(readableOptions); this._writableState = new WritableState( writableOptions, this as unknown as Writable, ); this._writableState.onwrite = onwrite.bind(undefined, this); }
[captureRejectionSymbol](err?: Error) { this.destroy(err); }
[Symbol.asyncIterator](): ReadableStreamAsyncIterator { return createReadableStreamAsyncIterator(this); }
_destroy( error: Error | null, callback: (error?: Error | null) => void, ): void { callback(error); }
_read = Readable.prototype._read;
_undestroy = Readable.prototype._undestroy;
destroy(err?: Error | null, cb?: (error?: Error | null) => void) { const r = this._readableState; const w = this._writableState;
if (w.destroyed || r.destroyed) { if (typeof cb === "function") { cb(); }
return this; }
if (err) { err.stack;
if (!w.errored) { w.errored = err; } if (!r.errored) { r.errored = err; } }
w.destroyed = true; r.destroyed = true;
this._destroy(err || null, (err) => { if (err) { err.stack;
if (!w.errored) { w.errored = err; } if (!r.errored) { r.errored = err; } }
w.closed = true; r.closed = true;
if (typeof cb === "function") { cb(err); }
if (err) { queueMicrotask(() => { const r = this._readableState; const w = this._writableState;
if (!w.errorEmitted && !r.errorEmitted) { w.errorEmitted = true; r.errorEmitted = true;
this.emit("error", err); }
r.closeEmitted = true;
if (w.emitClose || r.emitClose) { this.emit("close"); } }); } else { queueMicrotask(() => { const r = this._readableState; const w = this._writableState;
r.closeEmitted = true;
if (w.emitClose || r.emitClose) { this.emit("close"); } }); } });
return this; }
isPaused = Readable.prototype.isPaused;
off = this.removeListener;
on( event: "close" | "end" | "pause" | "readable" | "resume", listener: () => void, ): this; on(event: "data", listener: (chunk: any) => void): this; on(event: "error", listener: (err: Error) => void): this; on(event: string | symbol, listener: (...args: any[]) => void): this; on( ev: string | symbol, fn: | (() => void) | ((chunk: any) => void) | ((err: Error) => void) | ((...args: any[]) => void), ) { const res = super.on.call(this, ev, fn); const state = this._readableState;
if (ev === "data") { state.readableListening = this.listenerCount("readable") > 0;
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; if (state.length) { emitReadable(this); } else if (!state.reading) { queueMicrotask(() => nReadingNextTick(this)); } } }
return res; }
pause = Readable.prototype.pause as () => this;
pipe = Readable.prototype.pipe;
push(chunk: any, encoding?: Encodings): boolean { return readableAddChunk(this, chunk, encoding, false); }
read(n?: number) { if (n === undefined) { n = NaN; } const state = this._readableState; const nOrig = n;
if (n > state.highWaterMark) { state.highWaterMark = computeNewHighWaterMark(n); }
if (n !== 0) { state.emittedReadable = false; }
if ( n === 0 && state.needReadable && ((state.highWaterMark !== 0 ? state.length >= state.highWaterMark : state.length > 0) || state.ended) ) { if (state.length === 0 && state.ended) { endDuplex(this); } else { emitReadable(this); } return null; }
n = howMuchToRead(n, state);
if (n === 0 && state.ended) { if (state.length === 0) { endDuplex(this); } return null; }
let doRead = state.needReadable; if ( state.length === 0 || state.length - (n as number) < state.highWaterMark ) { doRead = true; }
if ( state.ended || state.reading || state.destroyed || state.errored || !state.constructed ) { doRead = false; } else if (doRead) { state.reading = true; state.sync = true; if (state.length === 0) { state.needReadable = true; } this._read(); state.sync = false; if (!state.reading) { n = howMuchToRead(nOrig, state); } }
let ret; if ((n as number) > 0) { ret = fromList((n as number), state); } else { ret = null; }
if (ret === null) { state.needReadable = state.length <= state.highWaterMark; n = 0; } else { state.length -= n as number; if (state.multiAwaitDrain) { (state.awaitDrainWriters as Set<Writable>).clear(); } else { state.awaitDrainWriters = null; } }
if (state.length === 0) { if (!state.ended) { state.needReadable = true; }
if (nOrig !== n && state.ended) { endDuplex(this); } }
if (ret !== null) { this.emit("data", ret); }
return ret; }
removeAllListeners( ev: | "close" | "data" | "end" | "error" | "pause" | "readable" | "resume" | symbol | undefined, ) { const res = super.removeAllListeners(ev);
if (ev === "readable" || ev === undefined) { queueMicrotask(() => updateReadableListening(this)); }
return res; }
removeListener( event: "close" | "end" | "pause" | "readable" | "resume", listener: () => void, ): this; removeListener(event: "data", listener: (chunk: any) => void): this; removeListener(event: "error", listener: (err: Error) => void): this; removeListener( event: string | symbol, listener: (...args: any[]) => void, ): this; removeListener( ev: string | symbol, fn: | (() => void) | ((chunk: any) => void) | ((err: Error) => void) | ((...args: any[]) => void), ) { const res = super.removeListener.call(this, ev, fn);
if (ev === "readable") { queueMicrotask(() => updateReadableListening(this)); }
return res; }
resume = Readable.prototype.resume as () => this;
setEncoding = Readable.prototype.setEncoding as (enc: string) => this;
unshift(chunk: any, encoding?: Encodings): boolean { return readableAddChunk(this, chunk, encoding, true); }
unpipe = Readable.prototype.unpipe as (dest?: Writable | undefined) => this;
wrap = Readable.prototype.wrap as (stream: Stream) => this;
get readable(): boolean { return this._readableState?.readable && !this._readableState?.destroyed && !this._readableState?.errorEmitted && !this._readableState?.endEmitted; } set readable(val: boolean) { if (this._readableState) { this._readableState.readable = val; } }
get readableHighWaterMark(): number { return this._readableState.highWaterMark; }
get readableBuffer() { return this._readableState && this._readableState.buffer; }
get readableFlowing(): boolean | null { return this._readableState.flowing; }
set readableFlowing(state: boolean | null) { if (this._readableState) { this._readableState.flowing = state; } }
get readableLength() { return this._readableState.length; }
get readableObjectMode() { return this._readableState ? this._readableState.objectMode : false; }
get readableEncoding() { return this._readableState ? this._readableState.encoding : null; }
get readableEnded() { return this._readableState ? this._readableState.endEmitted : false; }
_write = Writable.prototype._write;
write = Writable.prototype.write;
cork = Writable.prototype.cork;
uncork = Writable.prototype.uncork;
setDefaultEncoding(encoding: string) { if (typeof encoding === "string") { encoding = encoding.toLowerCase(); } if (!Buffer.isEncoding(encoding)) { throw new ERR_UNKNOWN_ENCODING(encoding); } this._writableState.defaultEncoding = encoding as Encodings; return this; }
end(cb?: () => void): void; end(chunk: any, cb?: () => void): void; end(chunk: any, encoding: Encodings, cb?: () => void): void;
end( x?: any | (() => void), y?: Encodings | (() => void), z?: () => void, ) { const state = this._writableState; let chunk: any | null; let encoding: Encodings | null; let cb: undefined | ((error?: Error) => void);
if (typeof x === "function") { chunk = null; encoding = null; cb = x; } else if (typeof y === "function") { chunk = x; encoding = null; cb = y; } else { chunk = x; encoding = y as Encodings; cb = z; }
if (chunk !== null && chunk !== undefined) { this.write(chunk, encoding); }
if (state.corked) { state.corked = 1; this.uncork(); }
let err: Error | undefined; if (!state.errored && !state.ending) { state.ending = true; finishMaybe(this, state, true); state.ended = true; } else if (state.finished) { err = new ERR_STREAM_ALREADY_FINISHED("end"); } else if (state.destroyed) { err = new ERR_STREAM_DESTROYED("end"); }
if (typeof cb === "function") { if (err || state.finished) { queueMicrotask(() => { (cb as (error?: Error | undefined) => void)(err); }); } else { state[kOnFinished].push(cb); } }
return this; }
get destroyed() { if ( this._readableState === undefined || this._writableState === undefined ) { return false; } return this._readableState.destroyed && this._writableState.destroyed; }
set destroyed(value: boolean) { if (this._readableState && this._writableState) { this._readableState.destroyed = value; this._writableState.destroyed = value; } }
get writable() { const w = this._writableState; return !w.destroyed && !w.errored && !w.ending && !w.ended; }
set writable(val) { if (this._writableState) { this._writableState.writable = !!val; } }
get writableFinished() { return this._writableState ? this._writableState.finished : false; }
get writableObjectMode() { return this._writableState ? this._writableState.objectMode : false; }
get writableBuffer() { return this._writableState && this._writableState.getBuffer(); }
get writableEnded() { return this._writableState ? this._writableState.ending : false; }
get writableHighWaterMark() { return this._writableState && this._writableState.highWaterMark; }
get writableCorked() { return this._writableState ? this._writableState.corked : 0; }
get writableLength() { return this._writableState && this._writableState.length; }}
export default Duplex;