You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
422 lines
9.8 KiB
422 lines
9.8 KiB
'use strict' |
|
const MiniPass = require('minipass') |
|
const EE = require('events').EventEmitter |
|
const fs = require('fs') |
|
|
|
let writev = fs.writev |
|
/* istanbul ignore next */ |
|
if (!writev) { |
|
// This entire block can be removed if support for earlier than Node.js |
|
// 12.9.0 is not needed. |
|
const binding = process.binding('fs') |
|
const FSReqWrap = binding.FSReqWrap || binding.FSReqCallback |
|
|
|
writev = (fd, iovec, pos, cb) => { |
|
const done = (er, bw) => cb(er, bw, iovec) |
|
const req = new FSReqWrap() |
|
req.oncomplete = done |
|
binding.writeBuffers(fd, iovec, pos, req) |
|
} |
|
} |
|
|
|
const _autoClose = Symbol('_autoClose') |
|
const _close = Symbol('_close') |
|
const _ended = Symbol('_ended') |
|
const _fd = Symbol('_fd') |
|
const _finished = Symbol('_finished') |
|
const _flags = Symbol('_flags') |
|
const _flush = Symbol('_flush') |
|
const _handleChunk = Symbol('_handleChunk') |
|
const _makeBuf = Symbol('_makeBuf') |
|
const _mode = Symbol('_mode') |
|
const _needDrain = Symbol('_needDrain') |
|
const _onerror = Symbol('_onerror') |
|
const _onopen = Symbol('_onopen') |
|
const _onread = Symbol('_onread') |
|
const _onwrite = Symbol('_onwrite') |
|
const _open = Symbol('_open') |
|
const _path = Symbol('_path') |
|
const _pos = Symbol('_pos') |
|
const _queue = Symbol('_queue') |
|
const _read = Symbol('_read') |
|
const _readSize = Symbol('_readSize') |
|
const _reading = Symbol('_reading') |
|
const _remain = Symbol('_remain') |
|
const _size = Symbol('_size') |
|
const _write = Symbol('_write') |
|
const _writing = Symbol('_writing') |
|
const _defaultFlag = Symbol('_defaultFlag') |
|
const _errored = Symbol('_errored') |
|
|
|
class ReadStream extends MiniPass { |
|
constructor (path, opt) { |
|
opt = opt || {} |
|
super(opt) |
|
|
|
this.readable = true |
|
this.writable = false |
|
|
|
if (typeof path !== 'string') |
|
throw new TypeError('path must be a string') |
|
|
|
this[_errored] = false |
|
this[_fd] = typeof opt.fd === 'number' ? opt.fd : null |
|
this[_path] = path |
|
this[_readSize] = opt.readSize || 16*1024*1024 |
|
this[_reading] = false |
|
this[_size] = typeof opt.size === 'number' ? opt.size : Infinity |
|
this[_remain] = this[_size] |
|
this[_autoClose] = typeof opt.autoClose === 'boolean' ? |
|
opt.autoClose : true |
|
|
|
if (typeof this[_fd] === 'number') |
|
this[_read]() |
|
else |
|
this[_open]() |
|
} |
|
|
|
get fd () { return this[_fd] } |
|
get path () { return this[_path] } |
|
|
|
write () { |
|
throw new TypeError('this is a readable stream') |
|
} |
|
|
|
end () { |
|
throw new TypeError('this is a readable stream') |
|
} |
|
|
|
[_open] () { |
|
fs.open(this[_path], 'r', (er, fd) => this[_onopen](er, fd)) |
|
} |
|
|
|
[_onopen] (er, fd) { |
|
if (er) |
|
this[_onerror](er) |
|
else { |
|
this[_fd] = fd |
|
this.emit('open', fd) |
|
this[_read]() |
|
} |
|
} |
|
|
|
[_makeBuf] () { |
|
return Buffer.allocUnsafe(Math.min(this[_readSize], this[_remain])) |
|
} |
|
|
|
[_read] () { |
|
if (!this[_reading]) { |
|
this[_reading] = true |
|
const buf = this[_makeBuf]() |
|
/* istanbul ignore if */ |
|
if (buf.length === 0) |
|
return process.nextTick(() => this[_onread](null, 0, buf)) |
|
fs.read(this[_fd], buf, 0, buf.length, null, (er, br, buf) => |
|
this[_onread](er, br, buf)) |
|
} |
|
} |
|
|
|
[_onread] (er, br, buf) { |
|
this[_reading] = false |
|
if (er) |
|
this[_onerror](er) |
|
else if (this[_handleChunk](br, buf)) |
|
this[_read]() |
|
} |
|
|
|
[_close] () { |
|
if (this[_autoClose] && typeof this[_fd] === 'number') { |
|
const fd = this[_fd] |
|
this[_fd] = null |
|
fs.close(fd, er => er ? this.emit('error', er) : this.emit('close')) |
|
} |
|
} |
|
|
|
[_onerror] (er) { |
|
this[_reading] = true |
|
this[_close]() |
|
this.emit('error', er) |
|
} |
|
|
|
[_handleChunk] (br, buf) { |
|
let ret = false |
|
// no effect if infinite |
|
this[_remain] -= br |
|
if (br > 0) |
|
ret = super.write(br < buf.length ? buf.slice(0, br) : buf) |
|
|
|
if (br === 0 || this[_remain] <= 0) { |
|
ret = false |
|
this[_close]() |
|
super.end() |
|
} |
|
|
|
return ret |
|
} |
|
|
|
emit (ev, data) { |
|
switch (ev) { |
|
case 'prefinish': |
|
case 'finish': |
|
break |
|
|
|
case 'drain': |
|
if (typeof this[_fd] === 'number') |
|
this[_read]() |
|
break |
|
|
|
case 'error': |
|
if (this[_errored]) |
|
return |
|
this[_errored] = true |
|
return super.emit(ev, data) |
|
|
|
default: |
|
return super.emit(ev, data) |
|
} |
|
} |
|
} |
|
|
|
class ReadStreamSync extends ReadStream { |
|
[_open] () { |
|
let threw = true |
|
try { |
|
this[_onopen](null, fs.openSync(this[_path], 'r')) |
|
threw = false |
|
} finally { |
|
if (threw) |
|
this[_close]() |
|
} |
|
} |
|
|
|
[_read] () { |
|
let threw = true |
|
try { |
|
if (!this[_reading]) { |
|
this[_reading] = true |
|
do { |
|
const buf = this[_makeBuf]() |
|
/* istanbul ignore next */ |
|
const br = buf.length === 0 ? 0 |
|
: fs.readSync(this[_fd], buf, 0, buf.length, null) |
|
if (!this[_handleChunk](br, buf)) |
|
break |
|
} while (true) |
|
this[_reading] = false |
|
} |
|
threw = false |
|
} finally { |
|
if (threw) |
|
this[_close]() |
|
} |
|
} |
|
|
|
[_close] () { |
|
if (this[_autoClose] && typeof this[_fd] === 'number') { |
|
const fd = this[_fd] |
|
this[_fd] = null |
|
fs.closeSync(fd) |
|
this.emit('close') |
|
} |
|
} |
|
} |
|
|
|
class WriteStream extends EE { |
|
constructor (path, opt) { |
|
opt = opt || {} |
|
super(opt) |
|
this.readable = false |
|
this.writable = true |
|
this[_errored] = false |
|
this[_writing] = false |
|
this[_ended] = false |
|
this[_needDrain] = false |
|
this[_queue] = [] |
|
this[_path] = path |
|
this[_fd] = typeof opt.fd === 'number' ? opt.fd : null |
|
this[_mode] = opt.mode === undefined ? 0o666 : opt.mode |
|
this[_pos] = typeof opt.start === 'number' ? opt.start : null |
|
this[_autoClose] = typeof opt.autoClose === 'boolean' ? |
|
opt.autoClose : true |
|
|
|
// truncating makes no sense when writing into the middle |
|
const defaultFlag = this[_pos] !== null ? 'r+' : 'w' |
|
this[_defaultFlag] = opt.flags === undefined |
|
this[_flags] = this[_defaultFlag] ? defaultFlag : opt.flags |
|
|
|
if (this[_fd] === null) |
|
this[_open]() |
|
} |
|
|
|
emit (ev, data) { |
|
if (ev === 'error') { |
|
if (this[_errored]) |
|
return |
|
this[_errored] = true |
|
} |
|
return super.emit(ev, data) |
|
} |
|
|
|
|
|
get fd () { return this[_fd] } |
|
get path () { return this[_path] } |
|
|
|
[_onerror] (er) { |
|
this[_close]() |
|
this[_writing] = true |
|
this.emit('error', er) |
|
} |
|
|
|
[_open] () { |
|
fs.open(this[_path], this[_flags], this[_mode], |
|
(er, fd) => this[_onopen](er, fd)) |
|
} |
|
|
|
[_onopen] (er, fd) { |
|
if (this[_defaultFlag] && |
|
this[_flags] === 'r+' && |
|
er && er.code === 'ENOENT') { |
|
this[_flags] = 'w' |
|
this[_open]() |
|
} else if (er) |
|
this[_onerror](er) |
|
else { |
|
this[_fd] = fd |
|
this.emit('open', fd) |
|
this[_flush]() |
|
} |
|
} |
|
|
|
end (buf, enc) { |
|
if (buf) |
|
this.write(buf, enc) |
|
|
|
this[_ended] = true |
|
|
|
// synthetic after-write logic, where drain/finish live |
|
if (!this[_writing] && !this[_queue].length && |
|
typeof this[_fd] === 'number') |
|
this[_onwrite](null, 0) |
|
return this |
|
} |
|
|
|
write (buf, enc) { |
|
if (typeof buf === 'string') |
|
buf = Buffer.from(buf, enc) |
|
|
|
if (this[_ended]) { |
|
this.emit('error', new Error('write() after end()')) |
|
return false |
|
} |
|
|
|
if (this[_fd] === null || this[_writing] || this[_queue].length) { |
|
this[_queue].push(buf) |
|
this[_needDrain] = true |
|
return false |
|
} |
|
|
|
this[_writing] = true |
|
this[_write](buf) |
|
return true |
|
} |
|
|
|
[_write] (buf) { |
|
fs.write(this[_fd], buf, 0, buf.length, this[_pos], (er, bw) => |
|
this[_onwrite](er, bw)) |
|
} |
|
|
|
[_onwrite] (er, bw) { |
|
if (er) |
|
this[_onerror](er) |
|
else { |
|
if (this[_pos] !== null) |
|
this[_pos] += bw |
|
if (this[_queue].length) |
|
this[_flush]() |
|
else { |
|
this[_writing] = false |
|
|
|
if (this[_ended] && !this[_finished]) { |
|
this[_finished] = true |
|
this[_close]() |
|
this.emit('finish') |
|
} else if (this[_needDrain]) { |
|
this[_needDrain] = false |
|
this.emit('drain') |
|
} |
|
} |
|
} |
|
} |
|
|
|
[_flush] () { |
|
if (this[_queue].length === 0) { |
|
if (this[_ended]) |
|
this[_onwrite](null, 0) |
|
} else if (this[_queue].length === 1) |
|
this[_write](this[_queue].pop()) |
|
else { |
|
const iovec = this[_queue] |
|
this[_queue] = [] |
|
writev(this[_fd], iovec, this[_pos], |
|
(er, bw) => this[_onwrite](er, bw)) |
|
} |
|
} |
|
|
|
[_close] () { |
|
if (this[_autoClose] && typeof this[_fd] === 'number') { |
|
const fd = this[_fd] |
|
this[_fd] = null |
|
fs.close(fd, er => er ? this.emit('error', er) : this.emit('close')) |
|
} |
|
} |
|
} |
|
|
|
class WriteStreamSync extends WriteStream { |
|
[_open] () { |
|
let fd |
|
// only wrap in a try{} block if we know we'll retry, to avoid |
|
// the rethrow obscuring the error's source frame in most cases. |
|
if (this[_defaultFlag] && this[_flags] === 'r+') { |
|
try { |
|
fd = fs.openSync(this[_path], this[_flags], this[_mode]) |
|
} catch (er) { |
|
if (er.code === 'ENOENT') { |
|
this[_flags] = 'w' |
|
return this[_open]() |
|
} else |
|
throw er |
|
} |
|
} else |
|
fd = fs.openSync(this[_path], this[_flags], this[_mode]) |
|
|
|
this[_onopen](null, fd) |
|
} |
|
|
|
[_close] () { |
|
if (this[_autoClose] && typeof this[_fd] === 'number') { |
|
const fd = this[_fd] |
|
this[_fd] = null |
|
fs.closeSync(fd) |
|
this.emit('close') |
|
} |
|
} |
|
|
|
[_write] (buf) { |
|
// throw the original, but try to close if it fails |
|
let threw = true |
|
try { |
|
this[_onwrite](null, |
|
fs.writeSync(this[_fd], buf, 0, buf.length, this[_pos])) |
|
threw = false |
|
} finally { |
|
if (threw) |
|
try { this[_close]() } catch (_) {} |
|
} |
|
} |
|
} |
|
|
|
exports.ReadStream = ReadStream |
|
exports.ReadStreamSync = ReadStreamSync |
|
|
|
exports.WriteStream = WriteStream |
|
exports.WriteStreamSync = WriteStreamSync
|
|
|