strong-tie/inbound-calls
0
1'use strict'2 3const fs = require('fs')4const EventEmitter = require('events')5const inherits = require('util').inherits6const path = require('path')7const sleep = require('atomic-sleep')8const assert = require('assert')9 10const BUSY_WRITE_TIMEOUT = 10011const kEmptyBuffer = Buffer.allocUnsafe(0)12 13// 16 KB. Don't write more than docker buffer size.14// https://github.com/moby/moby/blob/513ec73831269947d38a644c278ce3cac36783b2/daemon/logger/copier.go#L1315const MAX_WRITE = 16 * 102416 17const kContentModeBuffer = 'buffer'18const kContentModeUtf8 = 'utf8'19 20const [major, minor] = (process.versions.node || '0.0').split('.').map(Number)21const kCopyBuffer = major >= 22 && minor >= 722 23function openFile (file, sonic) {24 sonic._opening = true25 sonic._writing = true26 sonic._asyncDrainScheduled = false27 28 // NOTE: 'error' and 'ready' events emitted below only relevant when sonic.sync===false29 // for sync mode, there is no way to add a listener that will receive these30 31 function fileOpened (err, fd) {32 if (err) {33 sonic._reopening = false34 sonic._writing = false35 sonic._opening = false36 37 if (sonic.sync) {38 process.nextTick(() => {39 if (sonic.listenerCount('error') > 0) {40 sonic.emit('error', err)41 }42 })43 } else {44 sonic.emit('error', err)45 }46 return47 }48 49 const reopening = sonic._reopening50 51 sonic.fd = fd52 sonic.file = file53 sonic._reopening = false54 sonic._opening = false55 sonic._writing = false56 57 if (sonic.sync) {58 process.nextTick(() => sonic.emit('ready'))59 } else {60 sonic.emit('ready')61 }62 63 if (sonic.destroyed) {64 return65 }66 67 // start68 if ((!sonic._writing && sonic._len > sonic.minLength) || sonic._flushPending) {69 sonic._actualWrite()70 } else if (reopening) {71 process.nextTick(() => sonic.emit('drain'))72 }73 }74 75 const flags = sonic.append ? 'a' : 'w'76 const mode = sonic.mode77 78 if (sonic.sync) {79 try {80 if (sonic.mkdir) fs.mkdirSync(path.dirname(file), { recursive: true })81 const fd = fs.openSync(file, flags, mode)82 fileOpened(null, fd)83 } catch (err) {84 fileOpened(err)85 throw err86 }87 } else if (sonic.mkdir) {88 fs.mkdir(path.dirname(file), { recursive: true }, (err) => {89 if (err) return fileOpened(err)90 fs.open(file, flags, mode, fileOpened)91 })92 } else {93 fs.open(file, flags, mode, fileOpened)94 }95}96 97function SonicBoom (opts) {98 if (!(this instanceof SonicBoom)) {99 return new SonicBoom(opts)100 }101 102 let { fd, dest, minLength, maxLength, maxWrite, periodicFlush, sync, append = true, mkdir, retryEAGAIN, fsync, contentMode, mode } = opts || {}103 104 fd = fd || dest105 106 this._len = 0107 this.fd = -1108 this._bufs = []109 this._lens = []110 this._writing = false111 this._ending = false112 this._reopening = false113 this._asyncDrainScheduled = false114 this._flushPending = false115 this._hwm = Math.max(minLength || 0, 16387)116 this.file = null117 this.destroyed = false118 this.minLength = minLength || 0119 this.maxLength = maxLength || 0120 this.maxWrite = maxWrite || MAX_WRITE121 this._periodicFlush = periodicFlush || 0122 this._periodicFlushTimer = undefined123 this.sync = sync || false124 this.writable = true125 this._fsync = fsync || false126 this.append = append || false127 this.mode = mode128 this.retryEAGAIN = retryEAGAIN || (() => true)129 this.mkdir = mkdir || false130 131 let fsWriteSync132 let fsWrite133 if (contentMode === kContentModeBuffer) {134 this._writingBuf = kEmptyBuffer135 this.write = writeBuffer136 this.flush = flushBuffer137 this.flushSync = flushBufferSync138 this._actualWrite = actualWriteBuffer139 fsWriteSync = () => fs.writeSync(this.fd, this._writingBuf)140 fsWrite = () => fs.write(this.fd, this._writingBuf, this.release)141 } else if (contentMode === undefined || contentMode === kContentModeUtf8) {142 this._writingBuf = ''143 this.write = write144 this.flush = flush145 this.flushSync = flushSync146 this._actualWrite = actualWrite147 fsWriteSync = () => fs.writeSync(this.fd, this._writingBuf, 'utf8')148 fsWrite = () => fs.write(this.fd, this._writingBuf, 'utf8', this.release)149 } else {150 throw new Error(`SonicBoom supports "${kContentModeUtf8}" and "${kContentModeBuffer}", but passed ${contentMode}`)151 }152 153 if (typeof fd === 'number') {154 this.fd = fd155 process.nextTick(() => this.emit('ready'))156 } else if (typeof fd === 'string') {157 openFile(fd, this)158 } else {159 throw new Error('SonicBoom supports only file descriptors and files')160 }161 if (this.minLength >= this.maxWrite) {162 throw new Error(`minLength should be smaller than maxWrite (${this.maxWrite})`)163 }164 165 this.release = (err, n) => {166 if (err) {167 if ((err.code === 'EAGAIN' || err.code === 'EBUSY') && this.retryEAGAIN(err, this._writingBuf.length, this._len - this._writingBuf.length)) {168 if (this.sync) {169 // This error code should not happen in sync mode, because it is170 // not using the underlining operating system asynchronous functions.171 // However it happens, and so we handle it.172 // Ref: https://github.com/pinojs/pino/issues/783173 try {174 sleep(BUSY_WRITE_TIMEOUT)175 this.release(undefined, 0)176 } catch (err) {177 this.release(err)178 }179 } else {180 // Let's give the destination some time to process the chunk.181 setTimeout(fsWrite, BUSY_WRITE_TIMEOUT)182 }183 } else {184 this._writing = false185 186 this.emit('error', err)187 }188 return189 }190 191 this.emit('write', n)192 const releasedBufObj = releaseWritingBuf(this._writingBuf, this._len, n)193 this._len = releasedBufObj.len194 this._writingBuf = releasedBufObj.writingBuf195 196 if (this._writingBuf.length) {197 if (!this.sync) {198 fsWrite()199 return200 }201 202 try {203 do {204 const n = fsWriteSync()205 const releasedBufObj = releaseWritingBuf(this._writingBuf, this._len, n)206 this._len = releasedBufObj.len207 this._writingBuf = releasedBufObj.writingBuf208 } while (this._writingBuf.length)209 } catch (err) {210 this.release(err)211 return212 }213 }214 215 if (this._fsync) {216 fs.fsyncSync(this.fd)217 }218 219 const len = this._len220 if (this._reopening) {221 this._writing = false222 this._reopening = false223 this.reopen()224 } else if (len > this.minLength) {225 this._actualWrite()226 } else if (this._ending) {227 if (len > 0) {228 this._actualWrite()229 } else {230 this._writing = false231 actualClose(this)232 }233 } else {234 this._writing = false235 if (this.sync) {236 if (!this._asyncDrainScheduled) {237 this._asyncDrainScheduled = true238 process.nextTick(emitDrain, this)239 }240 } else {241 this.emit('drain')242 }243 }244 }245 246 this.on('newListener', function (name) {247 if (name === 'drain') {248 this._asyncDrainScheduled = false249 }250 })251 252 if (this._periodicFlush !== 0) {253 this._periodicFlushTimer = setInterval(() => this.flush(null), this._periodicFlush)254 this._periodicFlushTimer.unref()255 }256}257 258/**259 * Release the writingBuf after fs.write n bytes data260 * @param {string | Buffer} writingBuf - currently writing buffer, usually be instance._writingBuf.261 * @param {number} len - currently buffer length, usually be instance._len.262 * @param {number} n - number of bytes fs already written263 * @returns {{writingBuf: string | Buffer, len: number}} released writingBuf and length264 */265function releaseWritingBuf (writingBuf, len, n) {266 // if Buffer.byteLength is equal to n, that means writingBuf contains no multi-byte character267 if (typeof writingBuf === 'string' && Buffer.byteLength(writingBuf) !== n) {268 // Since the fs.write callback parameter `n` means how many bytes the passed of string269 // We calculate the original string length for avoiding the multi-byte character issue270 n = Buffer.from(writingBuf).subarray(0, n).toString().length271 }272 len = Math.max(len - n, 0)273 writingBuf = writingBuf.slice(n)274 return { writingBuf, len }275}276 277function emitDrain (sonic) {278 const hasListeners = sonic.listenerCount('drain') > 0279 if (!hasListeners) return280 sonic._asyncDrainScheduled = false281 sonic.emit('drain')282}283 284inherits(SonicBoom, EventEmitter)285 286function mergeBuf (bufs, len) {287 if (bufs.length === 0) {288 return kEmptyBuffer289 }290 291 if (bufs.length === 1) {292 return bufs[0]293 }294 295 return Buffer.concat(bufs, len)296}297 298function write (data) {299 if (this.destroyed) {300 throw new Error('SonicBoom destroyed')301 }302 303 const len = this._len + data.length304 const bufs = this._bufs305 306 if (this.maxLength && len > this.maxLength) {307 this.emit('drop', data)308 return this._len < this._hwm309 }310 311 if (312 bufs.length === 0 ||313 bufs[bufs.length - 1].length + data.length > this.maxWrite314 ) {315 bufs.push('' + data)316 } else {317 bufs[bufs.length - 1] += data318 }319 320 this._len = len321 322 if (!this._writing && this._len >= this.minLength) {323 this._actualWrite()324 }325 326 return this._len < this._hwm327}328 329function writeBuffer (data) {330 if (this.destroyed) {331 throw new Error('SonicBoom destroyed')332 }333 334 const len = this._len + data.length335 const bufs = this._bufs336 const lens = this._lens337 338 if (this.maxLength && len > this.maxLength) {339 this.emit('drop', data)340 return this._len < this._hwm341 }342 343 if (344 bufs.length === 0 ||345 lens[lens.length - 1] + data.length > this.maxWrite346 ) {347 bufs.push([data])348 lens.push(data.length)349 } else {350 bufs[bufs.length - 1].push(data)351 lens[lens.length - 1] += data.length352 }353 354 this._len = len355 356 if (!this._writing && this._len >= this.minLength) {357 this._actualWrite()358 }359 360 return this._len < this._hwm361}362 363function callFlushCallbackOnDrain (cb) {364 this._flushPending = true365 const onDrain = () => {366 // only if _fsync is false to avoid double fsync367 if (!this._fsync) {368 try {369 fs.fsync(this.fd, (err) => {370 this._flushPending = false371 cb(err)372 })373 } catch (err) {374 cb(err)375 }376 } else {377 this._flushPending = false378 cb()379 }380 this.off('error', onError)381 }382 const onError = (err) => {383 this._flushPending = false384 cb(err)385 this.off('drain', onDrain)386 }387 388 this.once('drain', onDrain)389 this.once('error', onError)390}391 392function flush (cb) {393 if (cb != null && typeof cb !== 'function') {394 throw new Error('flush cb must be a function')395 }396 397 if (this.destroyed) {398 const error = new Error('SonicBoom destroyed')399 if (cb) {400 cb(error)401 return402 }403 404 throw error405 }406 407 if (this.minLength <= 0) {408 cb?.()409 return410 }411 412 if (cb) {413 callFlushCallbackOnDrain.call(this, cb)414 }415 416 if (this._writing) {417 return418 }419 420 if (this._bufs.length === 0) {421 this._bufs.push('')422 }423 424 this._actualWrite()425}426 427function flushBuffer (cb) {428 if (cb != null && typeof cb !== 'function') {429 throw new Error('flush cb must be a function')430 }431 432 if (this.destroyed) {433 const error = new Error('SonicBoom destroyed')434 if (cb) {435 cb(error)436 return437 }438 439 throw error440 }441 442 if (this.minLength <= 0) {443 cb?.()444 return445 }446 447 if (cb) {448 callFlushCallbackOnDrain.call(this, cb)449 }450 451 if (this._writing) {452 return453 }454 455 if (this._bufs.length === 0) {456 this._bufs.push([])457 this._lens.push(0)458 }459 460 this._actualWrite()461}462 463SonicBoom.prototype.reopen = function (file) {464 if (this.destroyed) {465 throw new Error('SonicBoom destroyed')466 }467 468 if (this._opening) {469 this.once('ready', () => {470 this.reopen(file)471 })472 return473 }474 475 if (this._ending) {476 return477 }478 479 if (!this.file) {480 throw new Error('Unable to reopen a file descriptor, you must pass a file to SonicBoom')481 }482 483 if (file) {484 this.file = file485 }486 this._reopening = true487 488 if (this._writing) {489 return490 }491 492 const fd = this.fd493 this.once('ready', () => {494 if (fd !== this.fd) {495 fs.close(fd, (err) => {496 if (err) {497 return this.emit('error', err)498 }499 })500 }501 })502 503 openFile(this.file, this)504}505 506SonicBoom.prototype.end = function () {507 if (this.destroyed) {508 throw new Error('SonicBoom destroyed')509 }510 511 if (this._opening) {512 this.once('ready', () => {513 this.end()514 })515 return516 }517 518 if (this._ending) {519 return520 }521 522 this._ending = true523 524 if (this._writing) {525 return526 }527 528 if (this._len > 0 && this.fd >= 0) {529 this._actualWrite()530 } else {531 actualClose(this)532 }533}534 535function flushSync () {536 if (this.destroyed) {537 throw new Error('SonicBoom destroyed')538 }539 540 if (this.fd < 0) {541 throw new Error('sonic boom is not ready yet')542 }543 544 if (!this._writing && this._writingBuf.length > 0) {545 this._bufs.unshift(this._writingBuf)546 this._writingBuf = ''547 }548 549 let buf = ''550 while (this._bufs.length || buf) {551 if (buf.length <= 0) {552 buf = this._bufs[0]553 }554 try {555 const n = fs.writeSync(this.fd, buf, 'utf8')556 const releasedBufObj = releaseWritingBuf(buf, this._len, n)557 buf = releasedBufObj.writingBuf558 this._len = releasedBufObj.len559 if (buf.length <= 0) {560 this._bufs.shift()561 }562 } catch (err) {563 const shouldRetry = err.code === 'EAGAIN' || err.code === 'EBUSY'564 if (shouldRetry && !this.retryEAGAIN(err, buf.length, this._len - buf.length)) {565 throw err566 }567 568 sleep(BUSY_WRITE_TIMEOUT)569 }570 }571 572 try {573 fs.fsyncSync(this.fd)574 } catch {575 // Skip the error. The fd might not support fsync.576 }577}578 579function flushBufferSync () {580 if (this.destroyed) {581 throw new Error('SonicBoom destroyed')582 }583 584 if (this.fd < 0) {585 throw new Error('sonic boom is not ready yet')586 }587 588 if (!this._writing && this._writingBuf.length > 0) {589 this._bufs.unshift([this._writingBuf])590 this._writingBuf = kEmptyBuffer591 }592 593 let buf = kEmptyBuffer594 while (this._bufs.length || buf.length) {595 if (buf.length <= 0) {596 buf = mergeBuf(this._bufs[0], this._lens[0])597 }598 try {599 const n = fs.writeSync(this.fd, buf)600 buf = buf.subarray(n)601 this._len = Math.max(this._len - n, 0)602 if (buf.length <= 0) {603 this._bufs.shift()604 this._lens.shift()605 }606 } catch (err) {607 const shouldRetry = err.code === 'EAGAIN' || err.code === 'EBUSY'608 if (shouldRetry && !this.retryEAGAIN(err, buf.length, this._len - buf.length)) {609 throw err610 }611 612 sleep(BUSY_WRITE_TIMEOUT)613 }614 }615}616 617SonicBoom.prototype.destroy = function () {618 if (this.destroyed) {619 return620 }621 actualClose(this)622}623 624function actualWrite () {625 const release = this.release626 this._writing = true627 this._writingBuf = this._writingBuf || this._bufs.shift() || ''628 629 if (this.sync) {630 try {631 const written = fs.writeSync(this.fd, this._writingBuf, 'utf8')632 release(null, written)633 } catch (err) {634 release(err)635 }636 } else {637 fs.write(this.fd, this._writingBuf, 'utf8', release)638 }639}640 641function actualWriteBuffer () {642 const release = this.release643 this._writing = true644 this._writingBuf = this._writingBuf.length ? this._writingBuf : mergeBuf(this._bufs.shift(), this._lens.shift())645 646 if (this.sync) {647 try {648 const written = fs.writeSync(this.fd, this._writingBuf)649 release(null, written)650 } catch (err) {651 release(err)652 }653 } else {654 // fs.write will need to copy string to buffer anyway so655 // we do it here to avoid the overhead of calculating the buffer size656 // in releaseWritingBuf.657 if (kCopyBuffer) {658 this._writingBuf = Buffer.from(this._writingBuf)659 }660 fs.write(this.fd, this._writingBuf, release)661 }662}663 664function actualClose (sonic) {665 if (sonic.fd === -1) {666 sonic.once('ready', actualClose.bind(null, sonic))667 return668 }669 670 if (sonic._periodicFlushTimer !== undefined) {671 clearInterval(sonic._periodicFlushTimer)672 }673 674 sonic.destroyed = true675 sonic._bufs = []676 sonic._lens = []677 678 assert(typeof sonic.fd === 'number', `sonic.fd must be a number, got ${typeof sonic.fd}`)679 try {680 fs.fsync(sonic.fd, closeWrapped)681 } catch {682 }683 684 function closeWrapped () {685 // We skip errors in fsync686 687 if (sonic.fd !== 1 && sonic.fd !== 2) {688 fs.close(sonic.fd, done)689 } else {690 done()691 }692 }693 694 function done (err) {695 if (err) {696 sonic.emit('error', err)697 return698 }699 700 if (sonic._ending && !sonic._writing) {701 sonic.emit('finish')702 }703 sonic.emit('close')704 }705}706 707/**708 * These export configurations enable JS and TS developers709 * to consumer SonicBoom in whatever way best suits their needs.710 * Some examples of supported import syntax includes:711 * - `const SonicBoom = require('SonicBoom')`712 * - `const { SonicBoom } = require('SonicBoom')`713 * - `import * as SonicBoom from 'SonicBoom'`714 * - `import { SonicBoom } from 'SonicBoom'`715 * - `import SonicBoom from 'SonicBoom'`716 */717SonicBoom.SonicBoom = SonicBoom718SonicBoom.default = SonicBoom719module.exports = SonicBoom720