aboutsummaryrefslogtreecommitdiff
path: root/node_modules/readable-stream/lib/_stream_writable.js
diff options
context:
space:
mode:
Diffstat (limited to 'node_modules/readable-stream/lib/_stream_writable.js')
-rw-r--r--node_modules/readable-stream/lib/_stream_writable.js367
1 files changed, 208 insertions, 159 deletions
diff --git a/node_modules/readable-stream/lib/_stream_writable.js b/node_modules/readable-stream/lib/_stream_writable.js
index db8539cd5..ed5efcbd2 100644
--- a/node_modules/readable-stream/lib/_stream_writable.js
+++ b/node_modules/readable-stream/lib/_stream_writable.js
@@ -1,73 +1,80 @@
-// 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, cb), and it'll handle all
+// 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 Buffer = require('buffer').Buffer;
+var processNextTick = require('process-nextick-args');
/*</replacement>*/
-Writable.WritableState = WritableState;
+/*<replacement>*/
+var asyncWrite = !process.browser && ['v0.10', 'v0.9.'].indexOf(process.version.slice(0, 5)) > -1 ? setImmediate : processNextTick;
+/*</replacement>*/
+Writable.WritableState = WritableState;
/*<replacement>*/
var util = require('core-util-is');
util.inherits = require('inherits');
/*</replacement>*/
-var Stream = require('stream');
+/*<replacement>*/
+var internalUtil = {
+ deprecate: require('util-deprecate')
+};
+/*</replacement>*/
+
+/*<replacement>*/
+var Stream;
+(function () {
+ try {
+ Stream = require('st' + 'ream');
+ } catch (_) {} finally {
+ if (!Stream) Stream = require('events').EventEmitter;
+ }
+})();
+/*</replacement>*/
+
+var Buffer = require('buffer').Buffer;
+/*<replacement>*/
+var bufferShim = require('buffer-shims');
+/*</replacement>*/
util.inherits(Writable, Stream);
+function nop() {}
+
function WriteReq(chunk, encoding, cb) {
this.chunk = chunk;
this.encoding = encoding;
this.callback = cb;
+ this.next = null;
}
+var Duplex;
function WritableState(options, stream) {
- var Duplex = require('./_stream_duplex');
+ Duplex = Duplex || require('./_stream_duplex');
options = options || {};
- // the point at which write() starts returning false
- // Note: 0 is a valid value, means that we always return false if
- // the entire buffer is not flushed immediately on write()
- var hwm = options.highWaterMark;
- var defaultHwm = options.objectMode ? 16 : 16 * 1024;
- this.highWaterMark = (hwm || hwm === 0) ? hwm : defaultHwm;
-
// object stream flag to indicate whether or not this stream
// contains buffers or objects.
this.objectMode = !!options.objectMode;
- if (stream instanceof Duplex)
- this.objectMode = this.objectMode || !!options.writableObjectMode;
+ if (stream instanceof Duplex) this.objectMode = this.objectMode || !!options.writableObjectMode;
+
+ // the point at which write() starts returning false
+ // Note: 0 is a valid value, means that we always return false if
+ // the entire buffer is not flushed immediately on write()
+ var hwm = options.highWaterMark;
+ var defaultHwm = this.objectMode ? 16 : 16 * 1024;
+ this.highWaterMark = hwm || hwm === 0 ? hwm : defaultHwm;
// cast to ints.
- this.highWaterMark = ~~this.highWaterMark;
+ this.highWaterMark = ~ ~this.highWaterMark;
this.needDrain = false;
// at the start of calling end()
@@ -111,7 +118,7 @@ function WritableState(options, stream) {
this.bufferProcessing = false;
// the callback that's passed to _write(chunk,cb)
- this.onwrite = function(er) {
+ this.onwrite = function (er) {
onwrite(stream, er);
};
@@ -121,7 +128,8 @@ function WritableState(options, stream) {
// the amount that is being written when _write is called.
this.writelen = 0;
- this.buffer = [];
+ this.bufferedRequest = null;
+ this.lastBufferedRequest = null;
// number of pending user-supplied write callbacks
// this must be 0 before 'finish' can be emitted
@@ -133,37 +141,67 @@ function WritableState(options, stream) {
// True if the error was already emitted and should not be thrown again
this.errorEmitted = false;
+
+ // count buffered requests
+ this.bufferedRequestCount = 0;
+
+ // allocate the first CorkedRequest, there is always
+ // one allocated and free to use, and we maintain at most two
+ this.corkedRequestsFree = new CorkedRequest(this);
}
+WritableState.prototype.getBuffer = function writableStateGetBuffer() {
+ var current = this.bufferedRequest;
+ var out = [];
+ while (current) {
+ out.push(current);
+ current = current.next;
+ }
+ return out;
+};
+
+(function () {
+ try {
+ Object.defineProperty(WritableState.prototype, 'buffer', {
+ get: internalUtil.deprecate(function () {
+ return this.getBuffer();
+ }, '_writableState.buffer is deprecated. Use _writableState.getBuffer ' + 'instead.')
+ });
+ } catch (_) {}
+})();
+
+var Duplex;
function Writable(options) {
- var Duplex = require('./_stream_duplex');
+ Duplex = Duplex || require('./_stream_duplex');
// Writable ctor is applied to Duplexes, though they're not
// instanceof Writable, they're instanceof Readable.
- if (!(this instanceof Writable) && !(this instanceof Duplex))
- return new Writable(options);
+ if (!(this instanceof Writable) && !(this instanceof Duplex)) return new Writable(options);
this._writableState = new WritableState(options, this);
// legacy.
this.writable = true;
+ if (options) {
+ if (typeof options.write === 'function') this._write = options.write;
+
+ if (typeof options.writev === 'function') this._writev = options.writev;
+ }
+
Stream.call(this);
}
// Otherwise people can pipe Writable streams, which is just wrong.
-Writable.prototype.pipe = function() {
- this.emit('error', new Error('Cannot pipe. Not readable.'));
+Writable.prototype.pipe = function () {
+ this.emit('error', new Error('Cannot pipe, not readable'));
};
-
-function writeAfterEnd(stream, state, cb) {
+function writeAfterEnd(stream, cb) {
var er = new Error('write after end');
// TODO: defer error events consistently everywhere, not just the cb
stream.emit('error', er);
- process.nextTick(function() {
- cb(er);
- });
+ processNextTick(cb, er);
}
// If we get something that is not a buffer, string, null, or undefined,
@@ -173,40 +211,37 @@ function writeAfterEnd(stream, state, cb) {
// how many bytes or characters.
function validChunk(stream, state, chunk, cb) {
var valid = true;
- if (!util.isBuffer(chunk) &&
- !util.isString(chunk) &&
- !util.isNullOrUndefined(chunk) &&
- !state.objectMode) {
- var er = new TypeError('Invalid non-string/buffer chunk');
+ var er = false;
+ // Always throw error if a null is written
+ // if we are not in object mode then throw
+ // if it is not a buffer, string, or undefined.
+ if (chunk === null) {
+ er = new TypeError('May not write null values to stream');
+ } else if (!Buffer.isBuffer(chunk) && typeof chunk !== 'string' && chunk !== undefined && !state.objectMode) {
+ er = new TypeError('Invalid non-string/buffer chunk');
+ }
+ if (er) {
stream.emit('error', er);
- process.nextTick(function() {
- cb(er);
- });
+ processNextTick(cb, er);
valid = false;
}
return valid;
}
-Writable.prototype.write = function(chunk, encoding, cb) {
+Writable.prototype.write = function (chunk, encoding, cb) {
var state = this._writableState;
var ret = false;
- if (util.isFunction(encoding)) {
+ if (typeof encoding === 'function') {
cb = encoding;
encoding = null;
}
- if (util.isBuffer(chunk))
- encoding = 'buffer';
- else if (!encoding)
- encoding = state.defaultEncoding;
+ if (Buffer.isBuffer(chunk)) encoding = 'buffer';else if (!encoding) encoding = state.defaultEncoding;
- if (!util.isFunction(cb))
- cb = function() {};
+ if (typeof cb !== 'function') cb = nop;
- if (state.ended)
- writeAfterEnd(this, state, cb);
- else if (validChunk(this, state, chunk, cb)) {
+ if (state.ended) writeAfterEnd(this, cb);else if (validChunk(this, state, chunk, cb)) {
state.pendingcb++;
ret = writeOrBuffer(this, state, chunk, encoding, cb);
}
@@ -214,32 +249,33 @@ Writable.prototype.write = function(chunk, encoding, cb) {
return ret;
};
-Writable.prototype.cork = function() {
+Writable.prototype.cork = function () {
var state = this._writableState;
state.corked++;
};
-Writable.prototype.uncork = function() {
+Writable.prototype.uncork = function () {
var state = this._writableState;
if (state.corked) {
state.corked--;
- if (!state.writing &&
- !state.corked &&
- !state.finished &&
- !state.bufferProcessing &&
- state.buffer.length)
- clearBuffer(this, state);
+ if (!state.writing && !state.corked && !state.finished && !state.bufferProcessing && state.bufferedRequest) clearBuffer(this, state);
}
};
+Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) {
+ // node::ParseEncoding() requires lower case.
+ if (typeof encoding === 'string') encoding = encoding.toLowerCase();
+ if (!(['hex', 'utf8', 'utf-8', 'ascii', 'binary', 'base64', 'ucs2', 'ucs-2', 'utf16le', 'utf-16le', 'raw'].indexOf((encoding + '').toLowerCase()) > -1)) throw new TypeError('Unknown encoding: ' + encoding);
+ this._writableState.defaultEncoding = encoding;
+ return this;
+};
+
function decodeChunk(state, chunk, encoding) {
- if (!state.objectMode &&
- state.decodeStrings !== false &&
- util.isString(chunk)) {
- chunk = new Buffer(chunk, encoding);
+ if (!state.objectMode && state.decodeStrings !== false && typeof chunk === 'string') {
+ chunk = bufferShim.from(chunk, encoding);
}
return chunk;
}
@@ -249,21 +285,28 @@ function decodeChunk(state, chunk, encoding) {
// If we return false, then we need a drain event, so set that flag.
function writeOrBuffer(stream, state, chunk, encoding, cb) {
chunk = decodeChunk(state, chunk, encoding);
- if (util.isBuffer(chunk))
- encoding = 'buffer';
+
+ if (Buffer.isBuffer(chunk)) encoding = 'buffer';
var len = state.objectMode ? 1 : chunk.length;
state.length += len;
var ret = state.length < state.highWaterMark;
// we must ensure that previous needDrain will not be reset to false.
- if (!ret)
- state.needDrain = true;
+ if (!ret) state.needDrain = true;
- if (state.writing || state.corked)
- state.buffer.push(new WriteReq(chunk, encoding, cb));
- else
+ if (state.writing || state.corked) {
+ var last = state.lastBufferedRequest;
+ state.lastBufferedRequest = new WriteReq(chunk, encoding, cb);
+ if (last) {
+ last.next = state.lastBufferedRequest;
+ } else {
+ state.bufferedRequest = state.lastBufferedRequest;
+ }
+ state.bufferedRequestCount += 1;
+ } else {
doWrite(stream, state, false, len, chunk, encoding, cb);
+ }
return ret;
}
@@ -273,23 +316,13 @@ function doWrite(stream, state, writev, len, chunk, encoding, cb) {
state.writecb = cb;
state.writing = true;
state.sync = true;
- if (writev)
- stream._writev(chunk, state.onwrite);
- else
- stream._write(chunk, encoding, state.onwrite);
+ if (writev) stream._writev(chunk, state.onwrite);else stream._write(chunk, encoding, state.onwrite);
state.sync = false;
}
function onwriteError(stream, state, sync, er, cb) {
- if (sync)
- process.nextTick(function() {
- state.pendingcb--;
- cb(er);
- });
- else {
- state.pendingcb--;
- cb(er);
- }
+ --state.pendingcb;
+ if (sync) processNextTick(cb, er);else cb(er);
stream._writableState.errorEmitted = true;
stream.emit('error', er);
@@ -309,32 +342,26 @@ function onwrite(stream, er) {
onwriteStateUpdate(state);
- if (er)
- onwriteError(stream, state, sync, er, cb);
- else {
+ if (er) onwriteError(stream, state, sync, er, cb);else {
// Check if we're actually ready to finish, but don't emit yet
- var finished = needFinish(stream, state);
+ var finished = needFinish(state);
- if (!finished &&
- !state.corked &&
- !state.bufferProcessing &&
- state.buffer.length) {
+ if (!finished && !state.corked && !state.bufferProcessing && state.bufferedRequest) {
clearBuffer(stream, state);
}
if (sync) {
- process.nextTick(function() {
- afterWrite(stream, state, finished, cb);
- });
+ /*<replacement>*/
+ asyncWrite(afterWrite, stream, state, finished, cb);
+ /*</replacement>*/
} else {
- afterWrite(stream, state, finished, cb);
- }
+ afterWrite(stream, state, finished, cb);
+ }
}
}
function afterWrite(stream, state, finished, cb) {
- if (!finished)
- onwriteDrain(stream, state);
+ if (!finished) onwriteDrain(stream, state);
state.pendingcb--;
cb();
finishMaybe(stream, state);
@@ -350,80 +377,83 @@ function onwriteDrain(stream, state) {
}
}
-
// if there's something in the buffer waiting, then process it
function clearBuffer(stream, state) {
state.bufferProcessing = true;
+ var entry = state.bufferedRequest;
- if (stream._writev && state.buffer.length > 1) {
+ if (stream._writev && entry && entry.next) {
// Fast case, write everything using _writev()
- var cbs = [];
- for (var c = 0; c < state.buffer.length; c++)
- cbs.push(state.buffer[c].callback);
+ var l = state.bufferedRequestCount;
+ var buffer = new Array(l);
+ var holder = state.corkedRequestsFree;
+ holder.entry = entry;
+
+ var count = 0;
+ while (entry) {
+ buffer[count] = entry;
+ entry = entry.next;
+ count += 1;
+ }
- // count the one we are adding, as well.
- // TODO(isaacs) clean this up
- state.pendingcb++;
- doWrite(stream, state, true, state.length, state.buffer, '', function(err) {
- for (var i = 0; i < cbs.length; i++) {
- state.pendingcb--;
- cbs[i](err);
- }
- });
+ doWrite(stream, state, true, state.length, buffer, '', holder.finish);
- // Clear buffer
- state.buffer = [];
+ // doWrite is almost always async, defer these to save a bit of time
+ // as the hot path ends with doWrite
+ state.pendingcb++;
+ state.lastBufferedRequest = null;
+ if (holder.next) {
+ state.corkedRequestsFree = holder.next;
+ holder.next = null;
+ } else {
+ state.corkedRequestsFree = new CorkedRequest(state);
+ }
} else {
// Slow case, write chunks one-by-one
- for (var c = 0; c < state.buffer.length; c++) {
- var entry = state.buffer[c];
+ while (entry) {
var chunk = entry.chunk;
var encoding = entry.encoding;
var cb = entry.callback;
var len = state.objectMode ? 1 : chunk.length;
doWrite(stream, state, false, len, chunk, encoding, cb);
-
+ entry = entry.next;
// if we didn't call the onwrite immediately, then
// it means that we need to wait until it does.
// also, that means that the chunk and cb are currently
// being processed, so move the buffer counter past them.
if (state.writing) {
- c++;
break;
}
}
- if (c < state.buffer.length)
- state.buffer = state.buffer.slice(c);
- else
- state.buffer.length = 0;
+ if (entry === null) state.lastBufferedRequest = null;
}
+ state.bufferedRequestCount = 0;
+ state.bufferedRequest = entry;
state.bufferProcessing = false;
}
-Writable.prototype._write = function(chunk, encoding, cb) {
+Writable.prototype._write = function (chunk, encoding, cb) {
cb(new Error('not implemented'));
-
};
Writable.prototype._writev = null;
-Writable.prototype.end = function(chunk, encoding, cb) {
+Writable.prototype.end = function (chunk, encoding, cb) {
var state = this._writableState;
- if (util.isFunction(chunk)) {
+ if (typeof chunk === 'function') {
cb = chunk;
chunk = null;
encoding = null;
- } else if (util.isFunction(encoding)) {
+ } else if (typeof encoding === 'function') {
cb = encoding;
encoding = null;
}
- if (!util.isNullOrUndefined(chunk))
- this.write(chunk, encoding);
+ if (chunk !== null && chunk !== undefined) this.write(chunk, encoding);
// .end() fully uncorks
if (state.corked) {
@@ -432,16 +462,11 @@ Writable.prototype.end = function(chunk, encoding, cb) {
}
// ignore unnecessary end() calls.
- if (!state.ending && !state.finished)
- endWritable(this, state, cb);
+ if (!state.ending && !state.finished) endWritable(this, state, cb);
};
-
-function needFinish(stream, state) {
- return (state.ending &&
- state.length === 0 &&
- !state.finished &&
- !state.writing);
+function needFinish(state) {
+ return state.ending && state.length === 0 && state.bufferedRequest === null && !state.finished && !state.writing;
}
function prefinish(stream, state) {
@@ -452,14 +477,15 @@ function prefinish(stream, state) {
}
function finishMaybe(stream, state) {
- var need = needFinish(stream, state);
+ var need = needFinish(state);
if (need) {
if (state.pendingcb === 0) {
prefinish(stream, state);
state.finished = true;
stream.emit('finish');
- } else
+ } else {
prefinish(stream, state);
+ }
}
return need;
}
@@ -468,10 +494,33 @@ function endWritable(stream, state, cb) {
state.ending = true;
finishMaybe(stream, state);
if (cb) {
- if (state.finished)
- process.nextTick(cb);
- else
- stream.once('finish', cb);
+ if (state.finished) processNextTick(cb);else stream.once('finish', cb);
}
state.ended = true;
+ 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;
+
+ 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;
+ }
+ if (state.corkedRequestsFree) {
+ state.corkedRequestsFree.next = _this;
+ } else {
+ state.corkedRequestsFree = _this;
+ }
+ };
+} \ No newline at end of file