import { _uint8ArrayToBuffer } from "./_utils.ts";import { addAbortSignalNoValidate } from "./add-abort-signal.mjs";import { Buffer } from "../../buffer.ts";import { getDefaultHighWaterMark, getHighWaterMark } from "./state.mjs";import { isUint8Array } from "../util/types.ts";import { Stream } from "./legacy.mjs";import { ERR_INVALID_ARG_TYPE, ERR_METHOD_NOT_IMPLEMENTED, ERR_MULTIPLE_CALLBACK, ERR_STREAM_ALREADY_FINISHED, ERR_STREAM_CANNOT_PIPE, ERR_STREAM_DESTROYED, ERR_STREAM_NULL_VALUES, ERR_STREAM_WRITE_AFTER_END, ERR_UNKNOWN_ENCODING,} from "../errors.ts";import * as process from "../../_process/process.ts";import destroyImpl from "./destroy.mjs";import EE from "../../events.ts";import Readable from "./readable.mjs";
const { errorOrDestroy } = destroyImpl;
function isDuplexStream(maybe_duplex) { const isReadable = Readable.prototype.isPrototypeOf(maybe_duplex);
let prototype = maybe_duplex; let isDuplex = false; while (prototype?.constructor && prototype.constructor.name !== "Object") { if (prototype.constructor.name === "Duplex") { isDuplex = true; break; } prototype = Object.getPrototypeOf(prototype); }
return isReadable && isDuplex;}
Object.setPrototypeOf(Writable.prototype, Stream.prototype);Object.setPrototypeOf(Writable, Stream);
function nop() {}
const kOnFinished = Symbol("kOnFinished");
function WritableState(options, stream, isDuplex) { if (typeof isDuplex !== "boolean") { isDuplex = isDuplexStream(stream); }
this.objectMode = !!(options && options.objectMode);
if (isDuplex) { this.objectMode = this.objectMode || !!(options && options.writableObjectMode); }
this.highWaterMark = options ? getHighWaterMark(this, options, "writableHighWaterMark", isDuplex) : getDefaultHighWaterMark(false);
this.finalCalled = false;
this.needDrain = false; this.ending = false; this.ended = false; this.finished = false;
this.destroyed = false;
const noDecode = !!(options && options.decodeStrings === false); this.decodeStrings = !noDecode;
this.defaultEncoding = (options && options.defaultEncoding) || "utf8";
this.length = 0;
this.writing = false;
this.corked = 0;
this.sync = true;
this.bufferProcessing = false;
this.onwrite = onwrite.bind(undefined, stream);
this.writecb = null;
this.writelen = 0;
this.afterWriteTickInfo = null;
resetBuffer(this);
this.pendingcb = 0;
this.constructed = true;
this.prefinished = false;
this.errorEmitted = false;
this.emitClose = !options || options.emitClose !== false;
this.autoDestroy = !options || options.autoDestroy !== false;
this.errored = null;
this.closed = false;
this.closeEmitted = false;
this[kOnFinished] = [];}
function resetBuffer(state) { state.buffered = []; state.bufferedIndex = 0; state.allBuffers = true; state.allNoop = true;}
WritableState.prototype.getBuffer = function getBuffer() { return this.buffered.slice(this.bufferedIndex);};
Object.defineProperty(WritableState.prototype, "bufferedRequestCount", { get() { return this.buffered.length - this.bufferedIndex; },});
function Writable(options) {
const isDuplex = isDuplexStream(this);
if ( !isDuplex && !Function.prototype[Symbol.hasInstance].call(Writable, this) ) { return new Writable(options); }
this._writableState = new WritableState(options, this, isDuplex);
if (options) { if (typeof options.write === "function") { this._write = options.write; }
if (typeof options.writev === "function") { this._writev = options.writev; }
if (typeof options.destroy === "function") { this._destroy = options.destroy; }
if (typeof options.final === "function") { this._final = options.final; }
if (typeof options.construct === "function") { this._construct = options.construct; } if (options.signal) { addAbortSignalNoValidate(options.signal, this); } }
Stream.call(this, options);
destroyImpl.construct(this, () => { const state = this._writableState;
if (!state.writing) { clearBuffer(this, state); }
finishMaybe(this, state); });}
Object.defineProperty(Writable, Symbol.hasInstance, { value: function (object) { if (Function.prototype[Symbol.hasInstance].call(this, object)) return true; if (this !== Writable) return false;
return object && object._writableState instanceof WritableState; },});
Writable.prototype.pipe = function () { errorOrDestroy(this, new ERR_STREAM_CANNOT_PIPE());};
function _write(stream, chunk, encoding, cb) { const state = stream._writableState;
if (typeof encoding === "function") { cb = encoding; encoding = state.defaultEncoding; } else { if (!encoding) { encoding = state.defaultEncoding; } else if (encoding !== "buffer" && !Buffer.isEncoding(encoding)) { throw new ERR_UNKNOWN_ENCODING(encoding); } if (typeof cb !== "function") { cb = nop; } }
if (chunk === null) { throw new ERR_STREAM_NULL_VALUES(); } else if (!state.objectMode) { if (typeof chunk === "string") { if (state.decodeStrings !== false) { chunk = Buffer.from(chunk, encoding); encoding = "buffer"; } } else if (chunk instanceof Buffer) { encoding = "buffer"; } else if (isUint8Array(chunk)) { chunk = _uint8ArrayToBuffer(chunk); encoding = "buffer"; } else { throw new ERR_INVALID_ARG_TYPE( "chunk", ["string", "Buffer", "Uint8Array"], chunk, ); } }
let err; if (state.ending) { err = new ERR_STREAM_WRITE_AFTER_END(); } else if (state.destroyed) { err = new ERR_STREAM_DESTROYED("write"); }
if (err) { process.nextTick(cb, err); errorOrDestroy(stream, err, true); return err; } state.pendingcb++; return writeOrBuffer(stream, state, chunk, encoding, cb);}
Writable.prototype.write = function (chunk, encoding, cb) { return _write(this, chunk, encoding, cb) === true;};
Writable.prototype.cork = function () { this._writableState.corked++;};
Writable.prototype.uncork = function () { const state = this._writableState;
if (state.corked) { state.corked--;
if (!state.writing) { clearBuffer(this, state); } }};
Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) { if (typeof encoding === "string") { encoding = encoding.toLowerCase(); } if (!Buffer.isEncoding(encoding)) { throw new ERR_UNKNOWN_ENCODING(encoding); } this._writableState.defaultEncoding = encoding; return this;};
function writeOrBuffer(stream, state, chunk, encoding, callback) { const len = state.objectMode ? 1 : chunk.length;
state.length += len;
const ret = state.length < state.highWaterMark; if (!ret) { state.needDrain = true; }
if (state.writing || state.corked || state.errored || !state.constructed) { state.buffered.push({ chunk, encoding, callback }); if (state.allBuffers && encoding !== "buffer") { state.allBuffers = false; } if (state.allNoop && callback !== nop) { state.allNoop = false; } } else { state.writelen = len; state.writecb = callback; state.writing = true; state.sync = true; stream._write(chunk, encoding, state.onwrite); state.sync = false; }
return ret && !state.errored && !state.destroyed;}
function doWrite(stream, state, writev, len, chunk, encoding, cb) { state.writelen = len; state.writecb = cb; state.writing = true; state.sync = true; if (state.destroyed) { state.onwrite(new ERR_STREAM_DESTROYED("write")); } else if (writev) { stream._writev(chunk, state.onwrite); } else { stream._write(chunk, encoding, state.onwrite); } state.sync = false;}
function onwriteError(stream, state, er, cb) { --state.pendingcb;
cb(er); errorBuffer(state); errorOrDestroy(stream, er);}
function onwrite(stream, er) { const state = stream._writableState; const sync = state.sync; const cb = state.writecb;
if (typeof cb !== "function") { errorOrDestroy(stream, new ERR_MULTIPLE_CALLBACK()); return; }
state.writing = false; state.writecb = null; state.length -= state.writelen; state.writelen = 0;
if (er) { er.stack;
if (!state.errored) { state.errored = er; }
if (stream._readableState && !stream._readableState.errored) { stream._readableState.errored = er; }
if (sync) { process.nextTick(onwriteError, stream, state, er, cb); } else { onwriteError(stream, state, er, cb); } } else { if (state.buffered.length > state.bufferedIndex) { clearBuffer(stream, state); }
if (sync) { if ( state.afterWriteTickInfo !== null && state.afterWriteTickInfo.cb === cb ) { state.afterWriteTickInfo.count++; } else { state.afterWriteTickInfo = { count: 1, cb, stream, state }; process.nextTick(afterWriteTick, state.afterWriteTickInfo); } } else { afterWrite(stream, state, 1, cb); } }}
function afterWriteTick({ stream, state, count, cb }) { state.afterWriteTickInfo = null; return afterWrite(stream, state, count, cb);}
function afterWrite(stream, state, count, cb) { const needDrain = !state.ending && !stream.destroyed && state.length === 0 && state.needDrain; if (needDrain) { state.needDrain = false; stream.emit("drain"); }
while (count-- > 0) { state.pendingcb--; cb(); }
if (state.destroyed) { errorBuffer(state); }
finishMaybe(stream, state);}
function errorBuffer(state) { if (state.writing) { return; }
for (let n = state.bufferedIndex; n < state.buffered.length; ++n) { const { chunk, callback } = state.buffered[n]; const len = state.objectMode ? 1 : chunk.length; state.length -= len; callback(new ERR_STREAM_DESTROYED("write")); }
const onfinishCallbacks = state[kOnFinished].splice(0); for (let i = 0; i < onfinishCallbacks.length; i++) { onfinishCallbacks[i](new ERR_STREAM_DESTROYED("end")); }
resetBuffer(state);}
function clearBuffer(stream, state) { if ( state.corked || state.bufferProcessing || state.destroyed || !state.constructed ) { return; }
const { buffered, bufferedIndex, objectMode } = state; const bufferedLength = buffered.length - bufferedIndex;
if (!bufferedLength) { return; }
let i = bufferedIndex;
state.bufferProcessing = true; if (bufferedLength > 1 && stream._writev) { state.pendingcb -= bufferedLength - 1;
const callback = state.allNoop ? nop : (err) => { for (let n = i; n < buffered.length; ++n) { buffered[n].callback(err); } }; const chunks = state.allNoop && i === 0 ? buffered : buffered.slice(i); chunks.allBuffers = state.allBuffers;
doWrite(stream, state, true, state.length, chunks, "", callback);
resetBuffer(state); } else { do { const { chunk, encoding, callback } = buffered[i]; buffered[i++] = null; const len = objectMode ? 1 : chunk.length; doWrite(stream, state, false, len, chunk, encoding, callback); } while (i < buffered.length && !state.writing);
if (i === buffered.length) { resetBuffer(state); } else if (i > 256) { buffered.splice(0, i); state.bufferedIndex = 0; } else { state.bufferedIndex = i; } } state.bufferProcessing = false;}
Writable.prototype._write = function (chunk, encoding, cb) { if (this._writev) { this._writev([{ chunk, encoding }], cb); } else { throw new ERR_METHOD_NOT_IMPLEMENTED("_write()"); }};
Writable.prototype._writev = null;
Writable.prototype.end = function (chunk, encoding, cb) { const state = this._writableState;
if (typeof chunk === "function") { cb = chunk; chunk = null; encoding = null; } else if (typeof encoding === "function") { cb = encoding; encoding = null; }
let err;
if (chunk !== null && chunk !== undefined) { const ret = _write(this, chunk, encoding); if (ret instanceof Error) { err = ret; } }
if (state.corked) { state.corked = 1; this.uncork(); }
if (err) { } else 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) { process.nextTick(cb, err); } else { state[kOnFinished].push(cb); } }
return this;};
function needFinish(state) { return (state.ending && state.constructed && state.length === 0 && !state.errored && state.buffered.length === 0 && !state.finished && !state.writing && !state.errorEmitted && !state.closeEmitted);}
function callFinal(stream, state) { let called = false;
function onFinish(err) { if (called) { errorOrDestroy(stream, err ?? ERR_MULTIPLE_CALLBACK()); return; } called = true;
state.pendingcb--; if (err) { const onfinishCallbacks = state[kOnFinished].splice(0); for (let i = 0; i < onfinishCallbacks.length; i++) { onfinishCallbacks[i](err); } errorOrDestroy(stream, err, state.sync); } else if (needFinish(state)) { state.prefinished = true; stream.emit("prefinish"); state.pendingcb++; process.nextTick(finish, stream, state); } }
state.sync = true; state.pendingcb++;
try { const result = stream._final(onFinish); if (result != null) { const then = result.then; if (typeof then === "function") { then.call( result, function () { process.nextTick(onFinish, null); }, function (err) { process.nextTick(onFinish, err); }, ); } } } catch (err) { onFinish(stream, state, err); }
state.sync = false;}
function prefinish(stream, state) { if (!state.prefinished && !state.finalCalled) { if (typeof stream._final === "function" && !state.destroyed) { state.finalCalled = true; callFinal(stream, state); } else { state.prefinished = true; stream.emit("prefinish"); } }}
function finishMaybe(stream, state, sync) { if (needFinish(state)) { prefinish(stream, state); if (state.pendingcb === 0 && needFinish(state)) { state.pendingcb++; if (sync) { process.nextTick(finish, stream, state); } else { finish(stream, state); } } }}
function finish(stream, state) { state.pendingcb--; state.finished = true;
const onfinishCallbacks = state[kOnFinished].splice(0); for (let i = 0; i < onfinishCallbacks.length; i++) { onfinishCallbacks[i](); }
stream.emit("finish");
if (state.autoDestroy) { const rState = stream._readableState; const autoDestroy = !rState || ( rState.autoDestroy && (rState.endEmitted || rState.readable === false) ); if (autoDestroy) { stream.destroy(); } }}
Object.defineProperties(Writable.prototype, { destroyed: { get() { return this._writableState ? this._writableState.destroyed : false; }, set(value) { if (this._writableState) { this._writableState.destroyed = value; } }, },
writable: { get() { const w = this._writableState; return !!w && w.writable !== false && !w.destroyed && !w.errored && !w.ending && !w.ended; }, set(val) { if (this._writableState) { this._writableState.writable = !!val; } }, },
writableFinished: { get() { return this._writableState ? this._writableState.finished : false; }, },
writableObjectMode: { get() { return this._writableState ? this._writableState.objectMode : false; }, },
writableBuffer: { get() { return this._writableState && this._writableState.getBuffer(); }, },
writableEnded: { get() { return this._writableState ? this._writableState.ending : false; }, },
writableNeedDrain: { get() { const wState = this._writableState; if (!wState) return false; return !wState.destroyed && !wState.ending && wState.needDrain; }, },
writableHighWaterMark: { get() { return this._writableState && this._writableState.highWaterMark; }, },
writableCorked: { get() { return this._writableState ? this._writableState.corked : 0; }, },
writableLength: { get() { return this._writableState && this._writableState.length; }, },});
const destroy = destroyImpl.destroy;Writable.prototype.destroy = function (err, cb) { const state = this._writableState;
if ( !state.destroyed && (state.bufferedIndex < state.buffered.length || state[kOnFinished].length) ) { process.nextTick(errorBuffer, state); }
destroy.call(this, err, cb); return this;};
Writable.prototype._undestroy = destroyImpl.undestroy;Writable.prototype._destroy = function (err, cb) { cb(err);};
Writable.prototype[EE.captureRejectionSymbol] = function (err) { this.destroy(err);};
Writable.WritableState = WritableState;
export default Writable;export { Writable, WritableState };