diff options
author | Florian Dold <florian.dold@gmail.com> | 2017-08-14 05:01:11 +0200 |
---|---|---|
committer | Florian Dold <florian.dold@gmail.com> | 2017-08-14 05:02:09 +0200 |
commit | 363723fc84f7b8477592e0105aeb331ec9a017af (patch) | |
tree | 29f92724f34131bac64d6a318dd7e30612e631c7 /node_modules/readable-stream/lib | |
parent | 5634e77ad96bfe1818f6b6ee70b7379652e5487f (diff) | |
download | wallet-core-363723fc84f7b8477592e0105aeb331ec9a017af.tar.xz |
node_modules
Diffstat (limited to 'node_modules/readable-stream/lib')
7 files changed, 559 insertions, 183 deletions
diff --git a/node_modules/readable-stream/lib/_stream_duplex.js b/node_modules/readable-stream/lib/_stream_duplex.js index 736693b84..c599463dd 100644 --- a/node_modules/readable-stream/lib/_stream_duplex.js +++ b/node_modules/readable-stream/lib/_stream_duplex.js @@ -1,3 +1,24 @@ +// Copyright Joyent, Inc. and other Node contributors. +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the +// "Software"), to deal in the Software without restriction, including +// without limitation the rights to use, copy, modify, merge, publish, +// distribute, sublicense, and/or sell copies of the Software, and to permit +// persons to whom the Software is furnished to do so, subject to the +// following conditions: +// +// The above copyright notice and this permission notice shall be included +// in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN +// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE +// USE OR OTHER DEALINGS IN THE SOFTWARE. + // a duplex stream is just a stream that is both readable and writable. // Since JS doesn't have multiple prototypal inheritance, this class // prototypally inherits from Readable, and then parasitically from @@ -7,6 +28,10 @@ /*<replacement>*/ +var processNextTick = require('process-nextick-args'); +/*</replacement>*/ + +/*<replacement>*/ var objectKeys = Object.keys || function (obj) { var keys = []; for (var key in obj) { @@ -18,10 +43,6 @@ var objectKeys = Object.keys || function (obj) { module.exports = Duplex; /*<replacement>*/ -var processNextTick = require('process-nextick-args'); -/*</replacement>*/ - -/*<replacement>*/ var util = require('core-util-is'); util.inherits = require('inherits'); /*</replacement>*/ @@ -68,6 +89,34 @@ function onEndNT(self) { self.end(); } +Object.defineProperty(Duplex.prototype, 'destroyed', { + get: function () { + if (this._readableState === undefined || this._writableState === undefined) { + return false; + } + return this._readableState.destroyed && this._writableState.destroyed; + }, + set: function (value) { + // we ignore the value if the stream + // has not been initialized yet + if (this._readableState === undefined || this._writableState === undefined) { + return; + } + + // backward compatibility, the user is explicitly + // managing destroyed + this._readableState.destroyed = value; + this._writableState.destroyed = value; + } +}); + +Duplex.prototype._destroy = function (err, cb) { + this.push(null); + this.end(); + + processNextTick(cb, err); +}; + function forEach(xs, f) { for (var i = 0, l = xs.length; i < l; i++) { f(xs[i], i); diff --git a/node_modules/readable-stream/lib/_stream_passthrough.js b/node_modules/readable-stream/lib/_stream_passthrough.js index d06f71f18..a9c835884 100644 --- a/node_modules/readable-stream/lib/_stream_passthrough.js +++ b/node_modules/readable-stream/lib/_stream_passthrough.js @@ -1,3 +1,24 @@ +// Copyright Joyent, Inc. and other Node contributors. +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the +// "Software"), to deal in the Software without restriction, including +// without limitation the rights to use, copy, modify, merge, publish, +// distribute, sublicense, and/or sell copies of the Software, and to permit +// persons to whom the Software is furnished to do so, subject to the +// following conditions: +// +// The above copyright notice and this permission notice shall be included +// in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN +// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE +// USE OR OTHER DEALINGS IN THE SOFTWARE. + // a passthrough stream. // basically just the most minimal sort of Transform stream. // Every written chunk gets output as-is. diff --git a/node_modules/readable-stream/lib/_stream_readable.js b/node_modules/readable-stream/lib/_stream_readable.js index b19b2088b..ee9001cdf 100644 --- a/node_modules/readable-stream/lib/_stream_readable.js +++ b/node_modules/readable-stream/lib/_stream_readable.js @@ -1,11 +1,33 @@ -'use strict'; +// Copyright Joyent, Inc. and other Node contributors. +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the +// "Software"), to deal in the Software without restriction, including +// without limitation the rights to use, copy, modify, merge, publish, +// distribute, sublicense, and/or sell copies of the Software, and to permit +// persons to whom the Software is furnished to do so, subject to the +// following conditions: +// +// The above copyright notice and this permission notice shall be included +// in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN +// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE +// USE OR OTHER DEALINGS IN THE SOFTWARE. -module.exports = Readable; +'use strict'; /*<replacement>*/ + var processNextTick = require('process-nextick-args'); /*</replacement>*/ +module.exports = Readable; + /*<replacement>*/ var isArray = require('isarray'); /*</replacement>*/ @@ -28,9 +50,17 @@ var EElistenerCount = function (emitter, type) { var Stream = require('./internal/streams/stream'); /*</replacement>*/ -var Buffer = require('buffer').Buffer; +// TODO(bmeurer): Change this back to const once hole checks are +// properly optimized away early in Ignition+TurboFan. /*<replacement>*/ -var bufferShim = require('buffer-shims'); +var Buffer = require('safe-buffer').Buffer; +var OurUint8Array = global.Uint8Array || function () {}; +function _uint8ArrayToBuffer(chunk) { + return Buffer.from(chunk); +} +function _isUint8Array(obj) { + return Buffer.isBuffer(obj) || obj instanceof OurUint8Array; +} /*</replacement>*/ /*<replacement>*/ @@ -49,6 +79,7 @@ if (debugUtil && debugUtil.debuglog) { /*</replacement>*/ var BufferList = require('./internal/streams/BufferList'); +var destroyImpl = require('./internal/streams/destroy'); var StringDecoder; util.inherits(Readable, Stream); @@ -87,7 +118,7 @@ function ReadableState(options, stream) { this.highWaterMark = hwm || hwm === 0 ? hwm : defaultHwm; // cast to ints. - this.highWaterMark = ~~this.highWaterMark; + this.highWaterMark = Math.floor(this.highWaterMark); // A linked list is used to store data chunks instead of an array because the // linked list can remove elements from the beginning faster than @@ -101,10 +132,10 @@ function ReadableState(options, stream) { this.endEmitted = false; this.reading = false; - // a flag to be able to tell if the onwrite cb is called immediately, - // or on a later tick. We set this to true at first, because any - // actions that shouldn't happen until "later" should generally also - // not happen before the first write call. + // a flag to be able to tell if the event 'readable'/'data' is emitted + // immediately, or on a later tick. We set this to true at first, because + // any actions that shouldn't happen until "later" should generally also + // not happen before the first read call. this.sync = true; // whenever we return null, then we set a flag to say @@ -114,15 +145,14 @@ function ReadableState(options, stream) { this.readableListening = false; this.resumeScheduled = false; + // has it been destroyed + this.destroyed = false; + // Crypto is kind of old and crusty. Historically, its default string // encoding is 'binary' so we have to make this configurable. // Everything else in the universe uses 'utf8', though. this.defaultEncoding = options.defaultEncoding || 'utf8'; - // when piping, we only care about 'readable' events that happen - // after read()ing all the bytes and not getting any pushback. - this.ranOut = false; - // the number of writers that are awaiting a drain event in .pipe()s this.awaitDrain = 0; @@ -148,87 +178,129 @@ function Readable(options) { // legacy this.readable = true; - if (options && typeof options.read === 'function') this._read = options.read; + if (options) { + if (typeof options.read === 'function') this._read = options.read; + + if (typeof options.destroy === 'function') this._destroy = options.destroy; + } Stream.call(this); } +Object.defineProperty(Readable.prototype, 'destroyed', { + get: function () { + if (this._readableState === undefined) { + return false; + } + return this._readableState.destroyed; + }, + set: function (value) { + // we ignore the value if the stream + // has not been initialized yet + if (!this._readableState) { + return; + } + + // backward compatibility, the user is explicitly + // managing destroyed + this._readableState.destroyed = value; + } +}); + +Readable.prototype.destroy = destroyImpl.destroy; +Readable.prototype._undestroy = destroyImpl.undestroy; +Readable.prototype._destroy = function (err, cb) { + this.push(null); + cb(err); +}; + // Manually shove something into the read() buffer. // This returns true if the highWaterMark has not been hit yet, // similar to how Writable.write() returns true if you should // write() some more. Readable.prototype.push = function (chunk, encoding) { var state = this._readableState; - - if (!state.objectMode && typeof chunk === 'string') { - encoding = encoding || state.defaultEncoding; - if (encoding !== state.encoding) { - chunk = bufferShim.from(chunk, encoding); - encoding = ''; + var skipChunkCheck; + + if (!state.objectMode) { + if (typeof chunk === 'string') { + encoding = encoding || state.defaultEncoding; + if (encoding !== state.encoding) { + chunk = Buffer.from(chunk, encoding); + encoding = ''; + } + skipChunkCheck = true; } + } else { + skipChunkCheck = true; } - return readableAddChunk(this, state, chunk, encoding, false); + return readableAddChunk(this, chunk, encoding, false, skipChunkCheck); }; // Unshift should *always* be something directly out of read() Readable.prototype.unshift = function (chunk) { - var state = this._readableState; - return readableAddChunk(this, state, chunk, '', true); + return readableAddChunk(this, chunk, null, true, false); }; -Readable.prototype.isPaused = function () { - return this._readableState.flowing === false; -}; - -function readableAddChunk(stream, state, chunk, encoding, addToFront) { - var er = chunkInvalid(state, chunk); - if (er) { - stream.emit('error', er); - } else if (chunk === null) { +function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) { + var state = stream._readableState; + if (chunk === null) { state.reading = false; onEofChunk(stream, state); - } else if (state.objectMode || chunk && chunk.length > 0) { - if (state.ended && !addToFront) { - var e = new Error('stream.push() after EOF'); - stream.emit('error', e); - } else if (state.endEmitted && addToFront) { - var _e = new Error('stream.unshift() after end event'); - stream.emit('error', _e); - } else { - var skipAdd; - if (state.decoder && !addToFront && !encoding) { - chunk = state.decoder.write(chunk); - skipAdd = !state.objectMode && chunk.length === 0; + } else { + var er; + if (!skipChunkCheck) er = chunkInvalid(state, chunk); + if (er) { + stream.emit('error', er); + } else if (state.objectMode || chunk && chunk.length > 0) { + if (typeof chunk !== 'string' && !state.objectMode && Object.getPrototypeOf(chunk) !== Buffer.prototype) { + chunk = _uint8ArrayToBuffer(chunk); } - if (!addToFront) state.reading = false; - - // Don't add to the buffer if we've decoded to an empty string chunk and - // we're not in object mode - if (!skipAdd) { - // if we want the data now, just emit it. - if (state.flowing && state.length === 0 && !state.sync) { - stream.emit('data', chunk); - stream.read(0); + if (addToFront) { + if (state.endEmitted) stream.emit('error', new Error('stream.unshift() after end event'));else addChunk(stream, state, chunk, true); + } else if (state.ended) { + stream.emit('error', new Error('stream.push() after EOF')); + } else { + state.reading = false; + if (state.decoder && !encoding) { + chunk = state.decoder.write(chunk); + if (state.objectMode || chunk.length !== 0) addChunk(stream, state, chunk, false);else maybeReadMore(stream, state); } else { - // update the buffer info. - state.length += state.objectMode ? 1 : chunk.length; - if (addToFront) state.buffer.unshift(chunk);else state.buffer.push(chunk); - - if (state.needReadable) emitReadable(stream); + addChunk(stream, state, chunk, false); } } - - maybeReadMore(stream, state); + } else if (!addToFront) { + state.reading = false; } - } else if (!addToFront) { - state.reading = false; } return needMoreData(state); } +function addChunk(stream, state, chunk, addToFront) { + if (state.flowing && state.length === 0 && !state.sync) { + stream.emit('data', chunk); + stream.read(0); + } else { + // update the buffer info. + state.length += state.objectMode ? 1 : chunk.length; + if (addToFront) state.buffer.unshift(chunk);else state.buffer.push(chunk); + + if (state.needReadable) emitReadable(stream); + } + maybeReadMore(stream, state); +} + +function chunkInvalid(state, chunk) { + var er; + if (!_isUint8Array(chunk) && typeof chunk !== 'string' && chunk !== undefined && !state.objectMode) { + er = new TypeError('Invalid non-string/buffer chunk'); + } + return er; +} + // if it's past the high water mark, we can push in some more. // Also, if we have no data yet, we can stand some // more bytes. This is to work around cases where hwm=0, @@ -240,6 +312,10 @@ function needMoreData(state) { return !state.ended && (state.needReadable || state.length < state.highWaterMark || state.length === 0); } +Readable.prototype.isPaused = function () { + return this._readableState.flowing === false; +}; + // backwards compatibility. Readable.prototype.setEncoding = function (enc) { if (!StringDecoder) StringDecoder = require('string_decoder/').StringDecoder; @@ -388,14 +464,6 @@ Readable.prototype.read = function (n) { return ret; }; -function chunkInvalid(state, chunk) { - var er = null; - if (!Buffer.isBuffer(chunk) && typeof chunk !== 'string' && chunk !== null && chunk !== undefined && !state.objectMode) { - er = new TypeError('Invalid non-string/buffer chunk'); - } - return er; -} - function onEofChunk(stream, state) { if (state.ended) return; if (state.decoder) { @@ -483,14 +551,17 @@ Readable.prototype.pipe = function (dest, pipeOpts) { var doEnd = (!pipeOpts || pipeOpts.end !== false) && dest !== process.stdout && dest !== process.stderr; - var endFn = doEnd ? onend : cleanup; + var endFn = doEnd ? onend : unpipe; if (state.endEmitted) processNextTick(endFn);else src.once('end', endFn); dest.on('unpipe', onunpipe); - function onunpipe(readable) { + function onunpipe(readable, unpipeInfo) { debug('onunpipe'); if (readable === src) { - cleanup(); + if (unpipeInfo && unpipeInfo.hasUnpiped === false) { + unpipeInfo.hasUnpiped = true; + cleanup(); + } } } @@ -516,7 +587,7 @@ Readable.prototype.pipe = function (dest, pipeOpts) { dest.removeListener('error', onerror); dest.removeListener('unpipe', onunpipe); src.removeListener('end', onend); - src.removeListener('end', cleanup); + src.removeListener('end', unpipe); src.removeListener('data', ondata); cleanedUp = true; @@ -609,6 +680,7 @@ function pipeOnDrain(src) { Readable.prototype.unpipe = function (dest) { var state = this._readableState; + var unpipeInfo = { hasUnpiped: false }; // if we're not piping anywhere, then do nothing. if (state.pipesCount === 0) return this; @@ -624,7 +696,7 @@ Readable.prototype.unpipe = function (dest) { state.pipes = null; state.pipesCount = 0; state.flowing = false; - if (dest) dest.emit('unpipe', this); + if (dest) dest.emit('unpipe', this, unpipeInfo); return this; } @@ -639,7 +711,7 @@ Readable.prototype.unpipe = function (dest) { state.flowing = false; for (var i = 0; i < len; i++) { - dests[i].emit('unpipe', this); + dests[i].emit('unpipe', this, unpipeInfo); }return this; } @@ -651,7 +723,7 @@ Readable.prototype.unpipe = function (dest) { state.pipesCount -= 1; if (state.pipesCount === 1) state.pipes = state.pipes[0]; - dest.emit('unpipe', this); + dest.emit('unpipe', this, unpipeInfo); return this; }; @@ -672,7 +744,7 @@ Readable.prototype.on = function (ev, fn) { if (!state.reading) { processNextTick(nReadingNextTick, this); } else if (state.length) { - emitReadable(this, state); + emitReadable(this); } } } @@ -873,7 +945,7 @@ function copyFromBufferString(n, list) { // This function is designed to be inlinable, so please take care when making // changes to the function body. function copyFromBuffer(n, list) { - var ret = bufferShim.allocUnsafe(n); + var ret = Buffer.allocUnsafe(n); var p = list.head; var c = 1; p.data.copy(ret); diff --git a/node_modules/readable-stream/lib/_stream_transform.js b/node_modules/readable-stream/lib/_stream_transform.js index cd2583207..a0c23173d 100644 --- a/node_modules/readable-stream/lib/_stream_transform.js +++ b/node_modules/readable-stream/lib/_stream_transform.js @@ -1,3 +1,24 @@ +// Copyright Joyent, Inc. and other Node contributors. +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the +// "Software"), to deal in the Software without restriction, including +// without limitation the rights to use, copy, modify, merge, publish, +// distribute, sublicense, and/or sell copies of the Software, and to permit +// persons to whom the Software is furnished to do so, subject to the +// following conditions: +// +// The above copyright notice and this permission notice shall be included +// in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN +// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE +// USE OR OTHER DEALINGS IN THE SOFTWARE. + // a transform stream is a readable/writable stream where you do // something with the data. Sometimes it's called a "filter", // but that's not a great name for it, since that implies a thing where @@ -71,7 +92,9 @@ function afterTransform(stream, er, data) { var cb = ts.writecb; - if (!cb) return stream.emit('error', new Error('no writecb in Transform class')); + if (!cb) { + return stream.emit('error', new Error('write callback called multiple times')); + } ts.writechunk = null; ts.writecb = null; @@ -164,6 +187,15 @@ Transform.prototype._read = function (n) { } }; +Transform.prototype._destroy = function (err, cb) { + var _this = this; + + Duplex.prototype._destroy.call(this, err, function (err2) { + cb(err2); + _this.emit('close'); + }); +}; + function done(stream, er, data) { if (er) return stream.emit('error', er); diff --git a/node_modules/readable-stream/lib/_stream_writable.js b/node_modules/readable-stream/lib/_stream_writable.js index 15db03868..d20da88c7 100644 --- a/node_modules/readable-stream/lib/_stream_writable.js +++ b/node_modules/readable-stream/lib/_stream_writable.js @@ -1,15 +1,58 @@ +// Copyright Joyent, Inc. and other Node contributors. +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the +// "Software"), to deal in the Software without restriction, including +// without limitation the rights to use, copy, modify, merge, publish, +// distribute, sublicense, and/or sell copies of the Software, and to permit +// persons to whom the Software is furnished to do so, subject to the +// following conditions: +// +// The above copyright notice and this permission notice shall be included +// in all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN +// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, +// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE +// USE OR OTHER DEALINGS IN THE SOFTWARE. + // A bit simpler than readable streams. // Implement an async ._write(chunk, encoding, cb), and it'll handle all // the drain event emission and buffering. 'use strict'; -module.exports = Writable; - /*<replacement>*/ + var processNextTick = require('process-nextick-args'); /*</replacement>*/ +module.exports = Writable; + +/* <replacement> */ +function WriteReq(chunk, encoding, cb) { + this.chunk = chunk; + this.encoding = encoding; + this.callback = cb; + this.next = null; +} + +// It seems a linked list but it is not +// there will be only 2 of these for each stream +function CorkedRequest(state) { + var _this = this; + + this.next = null; + this.entry = null; + this.finish = function () { + onCorkedFinish(_this, state); + }; +} +/* </replacement> */ + /*<replacement>*/ var asyncWrite = !process.browser && ['v0.10', 'v0.9.'].indexOf(process.version.slice(0, 5)) > -1 ? setImmediate : processNextTick; /*</replacement>*/ @@ -35,22 +78,23 @@ var internalUtil = { var Stream = require('./internal/streams/stream'); /*</replacement>*/ -var Buffer = require('buffer').Buffer; /*<replacement>*/ -var bufferShim = require('buffer-shims'); +var Buffer = require('safe-buffer').Buffer; +var OurUint8Array = global.Uint8Array || function () {}; +function _uint8ArrayToBuffer(chunk) { + return Buffer.from(chunk); +} +function _isUint8Array(obj) { + return Buffer.isBuffer(obj) || obj instanceof OurUint8Array; +} /*</replacement>*/ +var destroyImpl = require('./internal/streams/destroy'); + util.inherits(Writable, Stream); function nop() {} -function WriteReq(chunk, encoding, cb) { - this.chunk = chunk; - this.encoding = encoding; - this.callback = cb; - this.next = null; -} - function WritableState(options, stream) { Duplex = Duplex || require('./_stream_duplex'); @@ -70,7 +114,10 @@ function WritableState(options, stream) { this.highWaterMark = hwm || hwm === 0 ? hwm : defaultHwm; // cast to ints. - this.highWaterMark = ~~this.highWaterMark; + this.highWaterMark = Math.floor(this.highWaterMark); + + // if _final has been called + this.finalCalled = false; // drain event flag. this.needDrain = false; @@ -81,6 +128,9 @@ function WritableState(options, stream) { // when 'finish' is emitted this.finished = false; + // has it been destroyed + this.destroyed = false; + // should we decode strings into buffers before passing to _write? // this is here so that some node-core streams can optimize string // handling at a lower level. @@ -162,7 +212,7 @@ WritableState.prototype.getBuffer = function getBuffer() { Object.defineProperty(WritableState.prototype, 'buffer', { get: internalUtil.deprecate(function () { return this.getBuffer(); - }, '_writableState.buffer is deprecated. Use _writableState.getBuffer ' + 'instead.') + }, '_writableState.buffer is deprecated. Use _writableState.getBuffer ' + 'instead.', 'DEP0003') }); } catch (_) {} })(); @@ -208,6 +258,10 @@ function Writable(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; } Stream.call(this); @@ -248,7 +302,11 @@ function validChunk(stream, state, chunk, cb) { Writable.prototype.write = function (chunk, encoding, cb) { var state = this._writableState; var ret = false; - var isBuf = Buffer.isBuffer(chunk); + var isBuf = _isUint8Array(chunk) && !state.objectMode; + + if (isBuf && !Buffer.isBuffer(chunk)) { + chunk = _uint8ArrayToBuffer(chunk); + } if (typeof encoding === 'function') { cb = encoding; @@ -293,7 +351,7 @@ Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) { function decodeChunk(state, chunk, encoding) { if (!state.objectMode && state.decodeStrings !== false && typeof chunk === 'string') { - chunk = bufferShim.from(chunk, encoding); + chunk = Buffer.from(chunk, encoding); } return chunk; } @@ -303,8 +361,12 @@ function decodeChunk(state, chunk, encoding) { // If we return false, then we need a drain event, so set that flag. function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) { if (!isBuf) { - chunk = decodeChunk(state, chunk, encoding); - if (Buffer.isBuffer(chunk)) encoding = 'buffer'; + var newChunk = decodeChunk(state, chunk, encoding); + if (chunk !== newChunk) { + isBuf = true; + encoding = 'buffer'; + chunk = newChunk; + } } var len = state.objectMode ? 1 : chunk.length; @@ -316,7 +378,13 @@ function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) { if (state.writing || state.corked) { var last = state.lastBufferedRequest; - state.lastBufferedRequest = new WriteReq(chunk, encoding, cb); + state.lastBufferedRequest = { + chunk: chunk, + encoding: encoding, + isBuf: isBuf, + callback: cb, + next: null + }; if (last) { last.next = state.lastBufferedRequest; } else { @@ -341,10 +409,26 @@ function doWrite(stream, state, writev, len, chunk, encoding, cb) { function onwriteError(stream, state, sync, er, cb) { --state.pendingcb; - if (sync) processNextTick(cb, er);else cb(er); - stream._writableState.errorEmitted = true; - stream.emit('error', er); + if (sync) { + // defer the callback if we are being called synchronously + // to avoid piling up things on the stack + processNextTick(cb, er); + // this can emit finish, and it will always happen + // after error + processNextTick(finishMaybe, stream, state); + stream._writableState.errorEmitted = true; + stream.emit('error', er); + } else { + // the caller expect this to happen before if + // it is async + cb(er); + stream._writableState.errorEmitted = true; + stream.emit('error', er); + // this can emit finish, but finish must + // always follow error + finishMaybe(stream, state); + } } function onwriteStateUpdate(state) { @@ -409,11 +493,14 @@ function clearBuffer(stream, state) { holder.entry = entry; var count = 0; + var allBuffers = true; while (entry) { buffer[count] = entry; + if (!entry.isBuf) allBuffers = false; entry = entry.next; count += 1; } + buffer.allBuffers = allBuffers; doWrite(stream, state, true, state.length, buffer, '', holder.finish); @@ -487,23 +574,37 @@ Writable.prototype.end = function (chunk, encoding, cb) { function needFinish(state) { return state.ending && state.length === 0 && state.bufferedRequest === null && !state.finished && !state.writing; } - -function prefinish(stream, state) { - if (!state.prefinished) { +function callFinal(stream, state) { + stream._final(function (err) { + state.pendingcb--; + if (err) { + stream.emit('error', err); + } state.prefinished = true; stream.emit('prefinish'); + finishMaybe(stream, state); + }); +} +function prefinish(stream, state) { + if (!state.prefinished && !state.finalCalled) { + if (typeof stream._final === 'function') { + state.pendingcb++; + state.finalCalled = true; + processNextTick(callFinal, stream, state); + } else { + state.prefinished = true; + stream.emit('prefinish'); + } } } function finishMaybe(stream, state) { var need = needFinish(state); if (need) { + prefinish(stream, state); if (state.pendingcb === 0) { - prefinish(stream, state); state.finished = true; stream.emit('finish'); - } else { - prefinish(stream, state); } } return need; @@ -519,26 +620,45 @@ function endWritable(stream, state, cb) { stream.writable = false; } -// It seems a linked list but it is not -// there will be only 2 of these for each stream -function CorkedRequest(state) { - var _this = this; +function onCorkedFinish(corkReq, state, err) { + var entry = corkReq.entry; + corkReq.entry = null; + while (entry) { + var cb = entry.callback; + state.pendingcb--; + cb(err); + entry = entry.next; + } + if (state.corkedRequestsFree) { + state.corkedRequestsFree.next = corkReq; + } else { + state.corkedRequestsFree = corkReq; + } +} - this.next = null; - this.entry = null; - this.finish = function (err) { - var entry = _this.entry; - _this.entry = null; - while (entry) { - var cb = entry.callback; - state.pendingcb--; - cb(err); - entry = entry.next; +Object.defineProperty(Writable.prototype, 'destroyed', { + get: function () { + if (this._writableState === undefined) { + return false; } - if (state.corkedRequestsFree) { - state.corkedRequestsFree.next = _this; - } else { - state.corkedRequestsFree = _this; + return this._writableState.destroyed; + }, + set: function (value) { + // we ignore the value if the stream + // has not been initialized yet + if (!this._writableState) { + return; } - }; -}
\ No newline at end of file + + // backward compatibility, the user is explicitly + // managing destroyed + this._writableState.destroyed = value; + } +}); + +Writable.prototype.destroy = destroyImpl.destroy; +Writable.prototype._undestroy = destroyImpl.undestroy; +Writable.prototype._destroy = function (err, cb) { + this.end(); + cb(err); +};
\ No newline at end of file diff --git a/node_modules/readable-stream/lib/internal/streams/BufferList.js b/node_modules/readable-stream/lib/internal/streams/BufferList.js index e4bfcf02d..d46761597 100644 --- a/node_modules/readable-stream/lib/internal/streams/BufferList.js +++ b/node_modules/readable-stream/lib/internal/streams/BufferList.js @@ -1,64 +1,74 @@ 'use strict'; -var Buffer = require('buffer').Buffer; /*<replacement>*/ -var bufferShim = require('buffer-shims'); -/*</replacement>*/ -module.exports = BufferList; +function _classCallCheck(instance, Constructor) { if (!(instance instanceof Constructor)) { throw new TypeError("Cannot call a class as a function"); } } + +var Buffer = require('safe-buffer').Buffer; +/*</replacement>*/ -function BufferList() { - this.head = null; - this.tail = null; - this.length = 0; +function copyBuffer(src, target, offset) { + src.copy(target, offset); } -BufferList.prototype.push = function (v) { - var entry = { data: v, next: null }; - if (this.length > 0) this.tail.next = entry;else this.head = entry; - this.tail = entry; - ++this.length; -}; +module.exports = function () { + function BufferList() { + _classCallCheck(this, BufferList); + + this.head = null; + this.tail = null; + this.length = 0; + } -BufferList.prototype.unshift = function (v) { - var entry = { data: v, next: this.head }; - if (this.length === 0) this.tail = entry; - this.head = entry; - ++this.length; -}; + BufferList.prototype.push = function push(v) { + var entry = { data: v, next: null }; + if (this.length > 0) this.tail.next = entry;else this.head = entry; + this.tail = entry; + ++this.length; + }; -BufferList.prototype.shift = function () { - if (this.length === 0) return; - var ret = this.head.data; - if (this.length === 1) this.head = this.tail = null;else this.head = this.head.next; - --this.length; - return ret; -}; + BufferList.prototype.unshift = function unshift(v) { + var entry = { data: v, next: this.head }; + if (this.length === 0) this.tail = entry; + this.head = entry; + ++this.length; + }; -BufferList.prototype.clear = function () { - this.head = this.tail = null; - this.length = 0; -}; + BufferList.prototype.shift = function shift() { + if (this.length === 0) return; + var ret = this.head.data; + if (this.length === 1) this.head = this.tail = null;else this.head = this.head.next; + --this.length; + return ret; + }; -BufferList.prototype.join = function (s) { - if (this.length === 0) return ''; - var p = this.head; - var ret = '' + p.data; - while (p = p.next) { - ret += s + p.data; - }return ret; -}; + BufferList.prototype.clear = function clear() { + this.head = this.tail = null; + this.length = 0; + }; -BufferList.prototype.concat = function (n) { - if (this.length === 0) return bufferShim.alloc(0); - if (this.length === 1) return this.head.data; - var ret = bufferShim.allocUnsafe(n >>> 0); - var p = this.head; - var i = 0; - while (p) { - p.data.copy(ret, i); - i += p.data.length; - p = p.next; - } - return ret; -};
\ No newline at end of file + BufferList.prototype.join = function join(s) { + if (this.length === 0) return ''; + var p = this.head; + var ret = '' + p.data; + while (p = p.next) { + ret += s + p.data; + }return ret; + }; + + BufferList.prototype.concat = function concat(n) { + if (this.length === 0) return Buffer.alloc(0); + if (this.length === 1) return this.head.data; + var ret = Buffer.allocUnsafe(n >>> 0); + var p = this.head; + var i = 0; + while (p) { + copyBuffer(p.data, ret, i); + i += p.data.length; + p = p.next; + } + return ret; + }; + + return BufferList; +}();
\ No newline at end of file diff --git a/node_modules/readable-stream/lib/internal/streams/destroy.js b/node_modules/readable-stream/lib/internal/streams/destroy.js new file mode 100644 index 000000000..b3e58c33b --- /dev/null +++ b/node_modules/readable-stream/lib/internal/streams/destroy.js @@ -0,0 +1,72 @@ +'use strict'; + +/*<replacement>*/ + +var processNextTick = require('process-nextick-args'); +/*</replacement>*/ + +// undocumented cb() API, needed for core, not for public API +function destroy(err, cb) { + var _this = this; + + var readableDestroyed = this._readableState && this._readableState.destroyed; + var writableDestroyed = this._writableState && this._writableState.destroyed; + + if (readableDestroyed || writableDestroyed) { + if (cb) { + cb(err); + } else if (err && (!this._writableState || !this._writableState.errorEmitted)) { + processNextTick(emitErrorNT, this, err); + } + return; + } + + // we set destroyed to true before firing error callbacks in order + // to make it re-entrance safe in case destroy() is called within callbacks + + if (this._readableState) { + this._readableState.destroyed = true; + } + + // if this is a duplex stream mark the writable part as destroyed as well + if (this._writableState) { + this._writableState.destroyed = true; + } + + this._destroy(err || null, function (err) { + if (!cb && err) { + processNextTick(emitErrorNT, _this, err); + if (_this._writableState) { + _this._writableState.errorEmitted = true; + } + } else if (cb) { + cb(err); + } + }); +} + +function undestroy() { + if (this._readableState) { + this._readableState.destroyed = false; + this._readableState.reading = false; + this._readableState.ended = false; + this._readableState.endEmitted = false; + } + + if (this._writableState) { + this._writableState.destroyed = false; + this._writableState.ended = false; + this._writableState.ending = false; + this._writableState.finished = false; + this._writableState.errorEmitted = false; + } +} + +function emitErrorNT(self, err) { + self.emit('error', err); +} + +module.exports = { + destroy: destroy, + undestroy: undestroy +};
\ No newline at end of file |