'use strict';
// This file is derived from the original SonicBoom module
// MIT License
// Copyright (c) 2017 Matteo Collina
const {
ArrayPrototypePush,
ArrayPrototypeUnshift,
MathMax,
Symbol,
SymbolDispose,
} = primordials;
const {
sleep,
kEmptyObject,
} = require('internal/util');
const {
Buffer,
} = require('buffer');
const {
isArrayBufferView,
} = require('internal/util/types');
const fs = require('fs');
const EventEmitter = require('events');
const path = require('path');
const {
clearInterval,
clearTimeout,
setInterval,
setTimeout,
} = require('timers');
const {
codes: {
ERR_INVALID_ARG_TYPE,
ERR_INVALID_ARG_VALUE,
ERR_INVALID_STATE,
ERR_OPERATION_FAILED,
},
} = require('internal/errors');
const {
validateBoolean,
validateFunction,
validateObject,
validateOneOf,
validateString,
validateUint32,
} = require('internal/validators');
const BUSY_WRITE_TIMEOUT = 100;
const kEmptyBuffer = Buffer.allocUnsafe(0);
// 16 KB. Don't write more than docker buffer size.
// https://github.com/moby/moby/blob/513ec73831269947d38a644c278ce3cac36783b2/daemon/logger/copier.go#L13
const kMaxWrite = 16 * 1024;
const kContentModeBuffer = 'buffer';
const kContentModeUtf8 = 'utf8';
const kNullPrototype = { __proto__: null };
const kFlush = Symbol('kFlush');
// States returned by #beginFlushSync().
// No write is in flight.
const kFlushSyncIdle = 0;
// An EAGAIN/EBUSY retry timer was pending and has been cancelled. Nothing is
// in flight, so the remainder of #writingBuf can be written synchronously.
const kFlushSyncRetry = 1;
// Called from a 'write' listener. The completed write has been released and
// nothing is in flight, so the remainder of #writingBuf can be written
// synchronously.
const kFlushSyncInWriteEvent = 2;
// An asynchronous fs.write() of #writingBuf is in flight. Only the queued
// buffers can be written; ordering relative to the in-flight chunk is not
// guaranteed.
const kFlushSyncInFlight = 3;
// fsync() error codes meaning the file descriptor cannot be synchronized
// (e.g. it is a pipe, socket, or TTY) or has already been closed. These are
// not treated as flush failures.
function isIgnorableFsyncError(err) {
switch (err?.code) {
case 'EBADF':
case 'EINVAL':
case 'ENOTSUP':
case 'EOPNOTSUPP':
case 'EROFS':
return true;
default:
return false;
}
}
// Utf8Stream is a port of the original SonicBoom module
// (https://github.com/pinojs/sonic-boom) that provides a fast and efficient
// way to write UTF-8 encoded data to a file or stream.
class Utf8Stream extends EventEmitter {
#len = 0;
#fd = -1;
#bufs = [];
#lens = [];
#writing = false;
#ending = false;
#reopening = false;
#asyncDrainScheduled = false;
#flushPending = 0;
#flushInProgress = 0;
#emittingFlush = false;
#emittingWrite = false;
#retryTimer = null;
#hwm = 16387; // 16 KB
#file = null;
#destroyed = false;
#minLength = 0;
#maxLength = 0;
#maxWrite = kMaxWrite;
#opening = false;
#periodicFlush = 0;
#periodicFlushTimer = undefined;
#sync = false;
#fsync = false;
#append = true;
#mode = 0o666;
#retryEAGAIN = () => true;
#mkdir = false;
#maxWriteRetries = 0;
#writeRetries = 0;
#writingBuf = '';
#write;
#flush;
#flushSync;
#actualWrite;
#fsWriteSync;
#fsWrite;
#fs;
/**
* @typedef {object} Utf8StreamOptions
* @property {string} [dest] - path to the file to write to
* @property {number} [fd] - file descriptor to write to
* @property {number} [minLength] - minimum length of the internal buffer before a write is triggered
* @property {number} [maxLength] - maximum length of the internal buffer before writes are dropped
* @property {number} [maxWrite] - maximum size of a single write operation (default 16384)
* @property {number} [periodicFlush] - interval in ms to flush the stream periodically (default 0, disabled)
* @property {boolean} [sync] - if true, writes are performed synchronously (default false)
* @property {boolean} [fsync] - if true, fsync is called after every write (default false)
* @property {boolean} [append] - if true, data is appended to the file (default true, ignored
* if fd is provided)
* @property {string} [contentMode] - 'utf8' or 'buffer'
* @property {string} [mode] - file mode (permission and sticky bits), default is 0o666
* @property {boolean} [mkdir] - if true, the directory path will be created if it does not exist (default false)
* @property {Function} [retryEAGAIN] - function that receives (err, writingBufLength,
* remainingBufLength) and returns true if EAGAIN/EBUSY should be retried
* @property {object} [fs] - custom fs implementation, must provide write, writeSync, fsync,
* fsyncSync, close, open, mkdir, mkdirSync methods. Mostly useful for testing.
* @param {Utf8StreamOptions} [options]
*/
constructor(options = kNullPrototype) {
validateObject(options, 'options');
let {
fd,
} = options;
const {
dest,
minLength,
maxLength,
maxWrite,
periodicFlush,
sync,
append = true,
mkdir,
retryEAGAIN,
maxWriteRetries,
fsync,
contentMode = kContentModeUtf8,
mode,
// Provides for a custom fs implementation. Mostly useful for testing.
fs: overrideFs = kEmptyObject,
} = options;
super();
fd ??= dest;
validateObject(overrideFs, 'options.fs');
this.#fs = { ...fs, ...overrideFs };
validateFunction(this.#fs.write, 'options.fs.write');
validateFunction(this.#fs.writeSync, 'options.fs.writeSync');
validateFunction(this.#fs.fsync, 'options.fs.fsync');
validateFunction(this.#fs.fsyncSync, 'options.fs.fsyncSync');
validateFunction(this.#fs.close, 'options.fs.close');
validateFunction(this.#fs.open, 'options.fs.open');
validateFunction(this.#fs.mkdir, 'options.fs.mkdir');
validateFunction(this.#fs.mkdirSync, 'options.fs.mkdirSync');
this.#hwm = MathMax(minLength || 0, this.#hwm);
this.#minLength = minLength || 0;
this.#maxLength = maxLength || 0;
this.#maxWrite = maxWrite || kMaxWrite;
this.#periodicFlush = periodicFlush || 0;
this.#sync = sync || false;
this.#fsync = fsync || false;
this.#append = append || false;
this.#mode = mode;
this.#retryEAGAIN = retryEAGAIN || (() => true);
this.#mkdir = mkdir || false;
this.#maxWriteRetries = maxWriteRetries || 0;
validateUint32(this.#hwm, 'options.hwm');
validateUint32(this.#minLength, 'options.minLength');
validateUint32(this.#maxLength, 'options.maxLength');
validateUint32(this.#maxWrite, 'options.maxWrite');
validateUint32(this.#periodicFlush, 'options.periodicFlush');
validateUint32(this.#maxWriteRetries, 'options.maxWriteRetries');
validateBoolean(this.#sync, 'options.sync');
validateBoolean(this.#fsync, 'options.fsync');
validateBoolean(this.#append, 'options.append');
validateBoolean(this.#mkdir, 'options.mkdir');
validateFunction(this.#retryEAGAIN, 'options.retryEAGAIN');
validateOneOf(contentMode, 'options.contentMode', [kContentModeBuffer, kContentModeUtf8]);
if (contentMode === kContentModeBuffer) {
this.#writingBuf = kEmptyBuffer;
this.#write = (...args) => this.#writeBuffer(...args);
this.#flush = (...args) => this.#flushBuffer(...args);
this.#flushSync = (...args) => this.#flushBufferSync(...args);
this.#actualWrite = (...args) => this.#actualWriteBuffer(...args);
this.#fsWriteSync = () => this.#fs.writeSync(this.#fd, this.#writingBuf);
this.#fsWrite = () => this.#fs.write(this.#fd, this.#writingBuf, (...args) => {
return this.#release(...args);
});
} else {
this.#writingBuf = '';
this.#write = (...args) => this.#writeUtf8(...args);
this.#flush = (...args) => this.#flushUtf8(...args);
this.#flushSync = (...args) => this.#flushSyncUtf8(...args);
this.#actualWrite = (...args) => this.#actualWriteUtf8(...args);
this.#fsWriteSync = () => this.#fs.writeSync(this.#fd, this.#writingBuf, 'utf8');
this.#fsWrite = () => this.#fs.write(this.#fd, this.#writingBuf, 'utf8', (...args) => {
return this.#release(...args);
});
}
// TODO(@jasnell): Support passing in an AbortSignal to cancel the stream.
// TODO(@jasnell): Support FileHandle here as well?
// TODO(@jasnell): The `dest` option is a path string. We may consider
// also supporting URL and Buffer types here as well like the rest of
// the fs module APIs do.
if (typeof fd === 'number') {
this.#fd = fd;
process.nextTick(() => this.emit('ready'));
} else if (typeof fd === 'string') {
this.#openFile(fd);
} else {
throw new ERR_INVALID_ARG_TYPE('fd', ['number', 'string'], fd);
}
if (this.#minLength >= this.#maxWrite) {
throw new ERR_INVALID_ARG_VALUE.RangeError('minLength', this.#minLength,
`should be smaller than maxWrite (${this.#maxWrite})`);
}
this.on('newListener', (name) => {
if (name === 'drain') {
this.#asyncDrainScheduled = false;
}
});
if (this.#periodicFlush !== 0) {
this.#periodicFlushTimer = setInterval(() => {
if (this.#flushPending === 0 && this.#flushInProgress === 0) {
this.flush();
}
}, this.#periodicFlush);
this.#periodicFlushTimer.unref();
}
}
/**
* @param {string|Buffer} data
* @returns {boolean}
*/
write(data) {
return this.#write(data);
}
/**
* @callback FlushCallback
* @param {Error} [err] - Error if any
* @returns {void}
*/
/**
* @param {FlushCallback} [cb]
*/
flush(cb = function(_err) {}) { this.#flush(cb); }
flushSync() { return this.#flushSync(); }
/**
* @param {string} [file]
*/
reopen(file) {
if (this.#destroyed) {
throw new ERR_INVALID_STATE('Utf8Stream is destroyed');
}
if (this.#opening) {
this.once('ready', () => {
if (!this.#destroyed) this.reopen(file);
});
return;
}
if (this.#ending) {
return;
}
if (!this.#file) {
throw new ERR_OPERATION_FAILED(
'Unable to reopen a file descriptor, you must pass a file to SonicBoom');
}
if (file) {
this.#file = file;
}
this.#reopening = true;
if (this.#writing) {
return;
}
const fd = this.#fd;
this.once('ready', () => {
if (fd !== this.#fd) {
this.#fs.close(fd, (err) => {
if (err) {
return this.emit('error', err);
}
});
}
});
this.#openFile(this.#file);
}
end() {
if (this.#destroyed) {
throw new ERR_INVALID_STATE('Utf8Stream is destroyed');
}
if (this.#opening) {
this.once('ready', () => {
if (!this.#destroyed) this.end();
});
return;
}
if (this.#ending) {
return;
}
this.#ending = true;
if (this.#writing) {
return;
}
if (this.#len > 0 && this.#fd >= 0) {
this.#actualWrite();
} else {
this.#finishEnding();
}
}
destroy() {
if (this.#destroyed) {
return;
}
const opening = this.#opening;
this.#actualClose();
if (opening && this.#flushPending > 0) {
this.emit('error', new ERR_INVALID_STATE('Utf8Stream is destroyed'));
}
}
/** @type {number} */
get mode() { return this.#mode; }
/** @type {string|undefined} */
get file() { return this.#file; }
/** @type {number} */
get fd() { return this.#fd; }
/** @type {number} */
get minLength() { return this.#minLength; }
/** @type {number} */
get maxLength() { return this.#maxLength; }
/** @type {boolean} */
get writing() { return this.#writing; }
/** @type {boolean} */
get sync() { return this.#sync; }
/** @type {boolean} */
get fsync() { return this.#fsync; }
/** @type {boolean} */
get append() { return this.#append; }
/** @type {number} */
get periodicFlush() { return this.#periodicFlush; }
/** @type {'buffer'|'utf8'} */
get contentMode() {
return this.#writingBuf instanceof Buffer ? kContentModeBuffer : kContentModeUtf8;
}
/** @type {boolean} */
get mkdir() { return this.#mkdir; }
[SymbolDispose]() { this.destroy(); }
#release(err, n) {
if (err) {
const isRetryableErr = (err.code === 'EAGAIN' || err.code === 'EBUSY');
if (isRetryableErr) {
this.#writeRetries++;
}
const retriesExhausted = this.#maxWriteRetries > 0 && this.#writeRetries > this.#maxWriteRetries;
if (isRetryableErr && !retriesExhausted &&
this.#retryEAGAIN(err, this.#writingBuf.length, this.#len - this.#writingBuf.length)) {
if (this.#sync) {
// This error code should not happen in sync mode, because it is
// not using the underlining operating system asynchronous functions.
// However it happens, and so we handle it.
// Ref: https://github.com/pinojs/pino/issues/783
try {
sleep(BUSY_WRITE_TIMEOUT);
this.#release(undefined, 0);
} catch (err) {
this.#release(err);
}
} else {
// Let's give the destination some time to process the chunk.
this.#retryTimer = setTimeout(() => {
this.#retryTimer = null;
this.#fsWrite();
}, BUSY_WRITE_TIMEOUT);
}
} else {
this.#writing = false;
this.emit('error', err);
}
return;
}
// Reset the retry counter only once real forward progress (n > 0) is
// confirmed. Otherwise a stuck destination would never accumulate past
// one retry and `maxWriteRetries` could never trigger.
if (n > 0) {
this.#writeRetries = 0;
}
const releasedBufObj = releaseWritingBuf(this.#writingBuf, this.#len, n);
this.#len = releasedBufObj.len;
this.#writingBuf = releasedBufObj.writingBuf;
// Emit 'write' after the written bytes have been released so that a
// listener calling flushSync() sees a consistent state with no I/O in
// flight. flushSync() may consume the remainder of #writingBuf.
this.#emittingWrite = true;
try {
this.emit('write', n);
} finally {
this.#emittingWrite = false;
}
if (this.#writingBuf.length) {
if (!this.#sync) {
this.#fsWrite();
return;
}
try {
do {
const n = this.#fsWriteSync();
const releasedBufObj = releaseWritingBuf(this.#writingBuf, this.#len, n);
this.#len = releasedBufObj.len;
this.#writingBuf = releasedBufObj.writingBuf;
} while (this.#writingBuf.length);
} catch (err) {
this.#release(err);
return;
}
}
if (this.#destroyed) {
this.#writing = false;
if (this.#flushPending > 0) {
if (this.#len === 0) {
this.#scheduleDrain();
} else {
this.emit('error', new ERR_INVALID_STATE('Utf8Stream is destroyed'));
}
}
return;
}
if (this.#fsync) {
this.#fs.fsyncSync(this.#fd);
}
const len = this.#len;
if (this.#reopening) {
this.#writing = false;
this.#reopening = false;
this.reopen();
} else if (len > 0 &&
(len > this.#minLength || this.#flushPending)) {
this.#actualWrite();
} else if (this.#ending) {
if (len > 0) {
this.#actualWrite();
} else {
this.#writing = false;
this.#finishEnding();
}
} else {
this.#writing = false;
this.#scheduleDrain();
}
}
#openFile(file) {
this.#opening = true;
this.#writing = true;
this.#asyncDrainScheduled = false;
// NOTE: 'error' and 'ready' events emitted below only relevant when sonic.sync===false
// for sync mode, there is no way to add a listener that will receive these
const fileOpened = (err, fd) => {
if (err) {
this.#reopening = false;
this.#writing = false;
this.#opening = false;
if (this.#sync) {
process.nextTick(() => {
if (this.listenerCount('error') > 0) {
this.emit('error', err);
}
});
} else {
this.emit('error', err);
}
return;
}
const reopening = this.#reopening;
this.#fd = fd;
this.#file = file;
this.#reopening = false;
this.#opening = false;
this.#writing = false;
if (this.#sync) {
process.nextTick(() => this.emit('ready'));
} else {
this.emit('ready');
}
if (this.#destroyed) {
return;
}
// start
if (!this.#writing &&
(this.#len > this.#minLength || this.#flushPending)) {
this.#actualWrite();
} else if (reopening && !this.#writing) {
// Do not emit 'drain' if a 'ready' listener started a write:
// #release() will emit the real 'drain' when that write completes.
process.nextTick(() => this.#emitDrain());
}
};
const flags = this.#append ? 'a' : 'w';
const mode = this.#mode;
if (this.#sync) {
try {
if (this.#mkdir) this.#fs.mkdirSync(path.dirname(file), { recursive: true });
const fd = this.#fs.openSync(file, flags, mode);
fileOpened(null, fd);
} catch (err) {
fileOpened(err);
throw err;
}
} else if (this.#mkdir) {
this.#fs.mkdir(path.dirname(file), { recursive: true }, (err) => {
if (err) return fileOpened(err);
this.#fs.open(file, flags, mode, fileOpened);
});
} else {
this.#fs.open(file, flags, mode, fileOpened);
}
}
#scheduleDrain() {
if (this.#sync) {
if (!this.#asyncDrainScheduled) {
this.#asyncDrainScheduled = true;
process.nextTick(() => this.#emitDrain());
}
} else {
this.#emitDrain();
}
}
#emitDrain() {
this.#asyncDrainScheduled = false;
if (this.#writing) return;
this.#emitFlush();
if (this.#writing || this.#destroyed) return;
const hasListeners = this.listenerCount('drain') > 0;
if (!hasListeners) return;
this.emit('drain');
}
#finishEnding() {
if (!this.#ending || this.#writing || this.#destroyed) return;
if (this.#len > 0) {
this.#actualWrite();
return;
}
if (this.#flushPending > 0) this.#emitFlush();
if (this.#flushPending === 0 && this.#flushInProgress === 0 &&
!this.#writing && this.#len === 0) {
this.#actualClose();
}
}
#emitFlush() {
if (this.#emittingFlush) return;
this.#emittingFlush = true;
try {
this.emit(kFlush);
} finally {
this.#emittingFlush = false;
}
}
#actualClose() {
if (this.#destroyed) return;
if (this.#fd === -1) {
this.#destroyed = true;
this.once('ready', () => {
this.#destroyed = false;
this.#actualClose();
});
return;
}
if (this.#periodicFlushTimer !== undefined) {
clearInterval(this.#periodicFlushTimer);
}
this.#destroyed = true;
this.#bufs = [];
this.#lens = [];
const done = (err) => {
if (err) {
this.emit('error', err);
return;
}
if (this.#ending && !this.#writing) {
this.emit('finish');
}
this.emit('close');
};
const closeWrapped = () => {
// We skip errors in fsync
if (this.#fd !== 1 && this.#fd !== 2) {
this.#fs.close(this.#fd, done);
} else {
done();
}
};
try {
this.#fs.fsync(this.#fd, closeWrapped);
} catch {
// Intentionally empty.
}
}
#actualWriteBuffer() {
this.#writing = true;
this.#writingBuf = this.#writingBuf.length ? this.#writingBuf : mergeBuf(this.#bufs.shift(), this.#lens.shift());
if (this.#sync) {
try {
const written = this.#fs.writeSync(this.#fd, this.#writingBuf);
this.#release(null, written);
} catch (err) {
this.#release(err);
}
} else {
// fs.write will need to copy string to buffer anyway so
// we do it here to avoid the overhead of calculating the buffer size
// in releaseWritingBuf.
this.#writingBuf = Buffer.from(this.#writingBuf);
this.#fs.write(this.#fd, this.#writingBuf, (...args) => this.#release(...args));
}
}
#actualWriteUtf8() {
this.#writing = true;
this.#writingBuf ||= this.#bufs.shift() || '';
if (this.#sync) {
try {
const written = this.#fs.writeSync(this.#fd, this.#writingBuf, 'utf8');
this.#release(null, written);
} catch (err) {
this.#release(err);
}
} else {
this.#fs.write(this.#fd, this.#writingBuf, 'utf8', (...args) => this.#release(...args));
}
}
/**
* Validates that flushSync() can run and determines how it must interact
* with any write already in progress.
* @returns {number} One of the kFlushSync* states.
*/
#beginFlushSync() {
if (this.#destroyed) {
throw new ERR_INVALID_STATE('Utf8Stream is destroyed');
}
if (this.#fd < 0) {
throw new ERR_INVALID_STATE('Invalid file descriptor');
}
// While reopening, #writing is set but no write is in flight and #fd still
// refers to the previous file, which is only closed once the new file is
// ready. Flush the buffered data to it.
if (!this.#writing || this.#opening) {
return kFlushSyncIdle;
}
if (this.#retryTimer !== null) {
clearTimeout(this.#retryTimer);
this.#retryTimer = null;
return kFlushSyncRetry;
}
if (this.#emittingWrite) {
return kFlushSyncInWriteEvent;
}
return kFlushSyncInFlight;
}
/**
* Called after flushSync() has taken over a write whose retry timer it
* cancelled. Resumes whatever #release() would have done once that write
* completed.
*/
#endFlushSyncRetry() {
this.#writing = false;
process.nextTick(() => {
if (this.#destroyed || this.#writing) return;
if (this.#reopening) {
this.#reopening = false;
this.reopen();
} else if (this.#ending) {
this.#finishEnding();
} else if (this.#len > 0 &&
(this.#len > this.#minLength || this.#flushPending)) {
this.#actualWrite();
} else {
this.#emitDrain();
}
});
}
#flushBufferSync() {
const state = this.#beginFlushSync();
// Unless an fs.write() of #writingBuf is in flight, its unwritten
// remainder must be written first to preserve ordering.
if (state !== kFlushSyncInFlight && this.#writingBuf.length > 0) {
ArrayPrototypeUnshift(this.#bufs, [this.#writingBuf]);
ArrayPrototypeUnshift(this.#lens, this.#writingBuf.length);
this.#writingBuf = kEmptyBuffer;
}
try {
let buf = kEmptyBuffer;
while (this.#bufs.length || buf.length) {
if (buf.length <= 0) {
buf = mergeBuf(this.#bufs[0], this.#lens[0]);
}
try {
const n = this.#fs.writeSync(this.#fd, buf);
this.#writeRetries = 0;
buf = buf.subarray(n);
this.#len = MathMax(this.#len - n, 0);
if (buf.length <= 0) {
this.#bufs.shift();
this.#lens.shift();
}
} catch (err) {
const shouldRetry = err.code === 'EAGAIN' || err.code === 'EBUSY';
if (shouldRetry) {
this.#writeRetries++;
}
const retriesExhausted = this.#maxWriteRetries > 0 && this.#writeRetries > this.#maxWriteRetries;
if (!shouldRetry || retriesExhausted || !this.#retryEAGAIN(err, buf.length, this.#len - buf.length)) {
throw err;
}
sleep(BUSY_WRITE_TIMEOUT);
}
}
} finally {
if (state === kFlushSyncRetry) this.#endFlushSyncRetry();
}
}
#flushSyncUtf8() {
const state = this.#beginFlushSync();
try {
this.#flushSyncUtf8Bufs(state);
} finally {
if (state === kFlushSyncRetry) this.#endFlushSyncRetry();
}
try {
this.#fs.fsyncSync(this.#fd);
} catch {
// Skip the error. The fd might not support fsync.
}
}
#flushSyncUtf8Bufs(state) {
// Unless an fs.write() of #writingBuf is in flight, its unwritten
// remainder must be written first to preserve ordering.
if (state !== kFlushSyncInFlight && this.#writingBuf.length > 0) {
ArrayPrototypeUnshift(this.#bufs, this.#writingBuf);
this.#writingBuf = '';
}
let buf = '';
while (this.#bufs.length || buf) {
if (buf.length <= 0) {
buf = this.#bufs[0];
}
try {
const n = this.#fs.writeSync(this.#fd, buf, 'utf8');
this.#writeRetries = 0;
const releasedBufObj = releaseWritingBuf(buf, this.#len, n);
buf = releasedBufObj.writingBuf;
this.#len = releasedBufObj.len;
if (buf.length <= 0) {
this.#bufs.shift();
}
} catch (err) {
const shouldRetry = err.code === 'EAGAIN' || err.code === 'EBUSY';
if (shouldRetry) {
this.#writeRetries++;
}
const retriesExhausted = this.#maxWriteRetries > 0 && this.#writeRetries > this.#maxWriteRetries;
if (!shouldRetry || retriesExhausted || !this.#retryEAGAIN(err, buf.length, this.#len - buf.length)) {
throw err;
}
sleep(BUSY_WRITE_TIMEOUT);
}
}
}
#callFlushCallbackOnDrain(cb) {
this.#flushPending++;
let waiting = true;
let completed = false;
let flushing = false;
const stopWaiting = () => {
if (!waiting) return false;
waiting = false;
this.off(kFlush, onDrain);
this.off('error', onError);
return true;
};
const complete = (err, finish = true) => {
if (completed) return;
completed = true;
if (flushing) this.#flushInProgress--;
try {
cb(err);
} finally {
if (finish) this.#finishEnding();
}
};
const onDrain = () => {
if (!stopWaiting()) return;
this.#flushPending--;
this.#flushInProgress++;
flushing = true;
// Only if _fsync is false to avoid double fsync
if (!this.#fsync && !this.#destroyed) {
try {
this.#fs.fsync(this.#fd, (err) => {
// Ignore errors meaning the fd is closed or cannot be synced
// (e.g. stdout/stderr attached to a pipe or TTY). A regular
// file, including a redirected stdout/stderr, is still synced.
if (isIgnorableFsyncError(err)) {
complete();
return;
}
complete(err);
});
} catch (err) {
complete(err);
}
} else {
complete();
}
};
const onError = (err) => {
if (!stopWaiting()) return;
this.#flushPending--;
complete(err, false);
};
this.once(kFlush, onDrain);
this.once('error', onError);
}
#flushBuffer(cb) {
validateFunction(cb, 'cb');
if (this.#destroyed) {
const error = new ERR_INVALID_STATE('Utf8Stream is destroyed');
if (cb) {
cb(error);
return;
}
throw error;
}
if (this.#minLength <= 0 && !this.#writing && this.#len === 0) {
cb?.();
return;
}
if (cb) {
this.#callFlushCallbackOnDrain(cb);
}
if (this.#writing) {
return;
}
if (this.#bufs.length === 0) {
ArrayPrototypePush(this.#bufs, []);
ArrayPrototypePush(this.#lens, 0);
}
this.#actualWrite();
}
#flushUtf8(cb) {
validateFunction(cb, 'cb');
if (this.#destroyed) {
const error = new ERR_INVALID_STATE('Utf8Stream is destroyed');
if (cb) {
cb(error);
return;
}
throw error;
}
if (this.#minLength <= 0 && !this.#writing && this.#len === 0) {
cb?.();
return;
}
if (cb) {
this.#callFlushCallbackOnDrain(cb);
}
if (this.#writing) {
return;
}
if (this.#bufs.length === 0) {
ArrayPrototypePush(this.#bufs, '');
}
this.#actualWrite();
}
#writeBuffer(data) {
if (this.#destroyed) {
throw new ERR_INVALID_STATE('Utf8Stream is destroyed');
}
if (!isArrayBufferView(data)) {
throw new ERR_INVALID_ARG_TYPE('data', ['Buffer', 'TypedArray', 'DataView'], data);
}
if (!Buffer.isBuffer(data)) {
// Reinterpret the view as a byte-oriented Buffer without copying, so that
// `data.length` below (and everywhere else `bufs`/`lens` are consumed)
// reflects the byte length rather than the element count.
data = Buffer.from(data.buffer, data.byteOffset, data.byteLength);
}
const len = this.#len + data.length;
const bufs = this.#bufs;
const lens = this.#lens;
if (this.#maxLength && len > this.#maxLength) {
this.emit('drop', data);
return this.#len < this.#hwm;
}
if (
bufs.length === 0 ||
lens[lens.length - 1] + data.length > this.#maxWrite
) {
ArrayPrototypePush(bufs, [data]);
ArrayPrototypePush(lens, data.length);
} else {
ArrayPrototypePush(bufs[bufs.length - 1], data);
lens[lens.length - 1] += data.length;
}
this.#len = len;
if (!this.#writing && this.#len >= this.#minLength) {
this.#actualWrite();
}
return this.#len < this.#hwm;
}
#writeUtf8(data) {
if (this.#destroyed) {
throw new ERR_INVALID_STATE('Utf8Stream is destroyed');
}
validateString(data, 'data');
const len = this.#len + data.length;
const bufs = this.#bufs;
if (this.#maxLength && len > this.#maxLength) {
this.emit('drop', data);
return this.#len < this.#hwm;
}
if (
bufs.length === 0 ||
bufs[bufs.length - 1].length + data.length > this.#maxWrite
) {
ArrayPrototypePush(bufs, '' + data);
} else {
bufs[bufs.length - 1] += data;
}
this.#len = len;
if (!this.#writing && this.#len >= this.#minLength) {
this.#actualWrite();
}
return this.#len < this.#hwm;
}
}
/**
* Release the writingBuf after fs.write n bytes data
* @param {string | Buffer} writingBuf - currently writing buffer, usually be instance._writingBuf.
* @param {number} len - currently buffer length, usually be instance._len.
* @param {number} n - number of bytes fs already written
* @returns {{writingBuf: string | Buffer, len: number}} released writingBuf and length
*/
function releaseWritingBuf(writingBuf, len, n) {
if (typeof writingBuf === 'string') {
const byteLength = Buffer.byteLength(writingBuf);
// `fs.write` returns the number of bytes written, but `len` is tracked in
// characters and `writingBuf` is sliced by character index below, so `n`
// must be converted from bytes to characters in both cases.
if (byteLength === n) {
// The whole string was written: advance past every character.
n = writingBuf.length;
} else {
// A partial write may split a multi-byte UTF-8 character, so we must back
// up to the start of that character to avoid data corruption.
const buf = Buffer.from(writingBuf);
// Back up from position n to find a valid UTF-8 character boundary.
// UTF-8 continuation bytes have the pattern 10xxxxxx (0x80-0xBF).
// We need to find the start of the character that was split.
while (n > 0 && (buf[n] & 0xC0) === 0x80) {
n--;
}
// Decode the properly-aligned bytes to get the character count.
n = buf.subarray(0, n).toString().length;
}
}
len = MathMax(len - n, 0);
writingBuf = writingBuf.slice(n);
return { writingBuf, len };
}
function mergeBuf(bufs, len) {
if (bufs.length === 0) {
return kEmptyBuffer;
}
if (bufs.length === 1) {
return bufs[0];
}
return Buffer.concat(bufs, len);
}
module.exports = Utf8Stream;