|
@@ -49,29 +49,15 @@ util.inherits = require('inherits');
|
|
|
|
|
|
|
|
var StringDecoder;
|
|
var StringDecoder;
|
|
|
|
|
|
|
|
-
|
|
|
|
|
-/*<replacement>*/
|
|
|
|
|
-var debug = require('util');
|
|
|
|
|
-if (debug && debug.debuglog) {
|
|
|
|
|
- debug = debug.debuglog('stream');
|
|
|
|
|
-} else {
|
|
|
|
|
- debug = function () {};
|
|
|
|
|
-}
|
|
|
|
|
-/*</replacement>*/
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
util.inherits(Readable, Stream);
|
|
util.inherits(Readable, Stream);
|
|
|
|
|
|
|
|
function ReadableState(options, stream) {
|
|
function ReadableState(options, stream) {
|
|
|
- var Duplex = require('./_stream_duplex');
|
|
|
|
|
-
|
|
|
|
|
options = options || {};
|
|
options = options || {};
|
|
|
|
|
|
|
|
// the point at which it stops calling _read() to fill the buffer
|
|
// the point at which it stops calling _read() to fill the buffer
|
|
|
// Note: 0 is a valid value, means "don't call _read preemptively ever"
|
|
// Note: 0 is a valid value, means "don't call _read preemptively ever"
|
|
|
var hwm = options.highWaterMark;
|
|
var hwm = options.highWaterMark;
|
|
|
- var defaultHwm = options.objectMode ? 16 : 16 * 1024;
|
|
|
|
|
- this.highWaterMark = (hwm || hwm === 0) ? hwm : defaultHwm;
|
|
|
|
|
|
|
+ this.highWaterMark = (hwm || hwm === 0) ? hwm : 16 * 1024;
|
|
|
|
|
|
|
|
// cast to ints.
|
|
// cast to ints.
|
|
|
this.highWaterMark = ~~this.highWaterMark;
|
|
this.highWaterMark = ~~this.highWaterMark;
|
|
@@ -80,13 +66,19 @@ function ReadableState(options, stream) {
|
|
|
this.length = 0;
|
|
this.length = 0;
|
|
|
this.pipes = null;
|
|
this.pipes = null;
|
|
|
this.pipesCount = 0;
|
|
this.pipesCount = 0;
|
|
|
- this.flowing = null;
|
|
|
|
|
|
|
+ this.flowing = false;
|
|
|
this.ended = false;
|
|
this.ended = false;
|
|
|
this.endEmitted = false;
|
|
this.endEmitted = false;
|
|
|
this.reading = false;
|
|
this.reading = false;
|
|
|
|
|
|
|
|
|
|
+ // In streams that never have any data, and do push(null) right away,
|
|
|
|
|
+ // the consumer can miss the 'end' event if they do some I/O before
|
|
|
|
|
+ // consuming the stream. So, we don't emit('end') until some reading
|
|
|
|
|
+ // happens.
|
|
|
|
|
+ this.calledRead = false;
|
|
|
|
|
+
|
|
|
// a flag to be able to tell if the onwrite cb is called immediately,
|
|
// 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
|
|
|
|
|
|
|
+ // or on a later tick. We set this to true at first, becuase any
|
|
|
// actions that shouldn't happen until "later" should generally also
|
|
// actions that shouldn't happen until "later" should generally also
|
|
|
// not happen before the first write call.
|
|
// not happen before the first write call.
|
|
|
this.sync = true;
|
|
this.sync = true;
|
|
@@ -102,9 +94,6 @@ function ReadableState(options, stream) {
|
|
|
// make all the buffer merging and length checks go away
|
|
// make all the buffer merging and length checks go away
|
|
|
this.objectMode = !!options.objectMode;
|
|
this.objectMode = !!options.objectMode;
|
|
|
|
|
|
|
|
- if (stream instanceof Duplex)
|
|
|
|
|
- this.objectMode = this.objectMode || !!options.readableObjectMode;
|
|
|
|
|
-
|
|
|
|
|
// Crypto is kind of old and crusty. Historically, its default string
|
|
// Crypto is kind of old and crusty. Historically, its default string
|
|
|
// encoding is 'binary' so we have to make this configurable.
|
|
// encoding is 'binary' so we have to make this configurable.
|
|
|
// Everything else in the universe uses 'utf8', though.
|
|
// Everything else in the universe uses 'utf8', though.
|
|
@@ -131,8 +120,6 @@ function ReadableState(options, stream) {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
function Readable(options) {
|
|
function Readable(options) {
|
|
|
- var Duplex = require('./_stream_duplex');
|
|
|
|
|
-
|
|
|
|
|
if (!(this instanceof Readable))
|
|
if (!(this instanceof Readable))
|
|
|
return new Readable(options);
|
|
return new Readable(options);
|
|
|
|
|
|
|
@@ -151,7 +138,7 @@ function Readable(options) {
|
|
|
Readable.prototype.push = function(chunk, encoding) {
|
|
Readable.prototype.push = function(chunk, encoding) {
|
|
|
var state = this._readableState;
|
|
var state = this._readableState;
|
|
|
|
|
|
|
|
- if (util.isString(chunk) && !state.objectMode) {
|
|
|
|
|
|
|
+ if (typeof chunk === 'string' && !state.objectMode) {
|
|
|
encoding = encoding || state.defaultEncoding;
|
|
encoding = encoding || state.defaultEncoding;
|
|
|
if (encoding !== state.encoding) {
|
|
if (encoding !== state.encoding) {
|
|
|
chunk = new Buffer(chunk, encoding);
|
|
chunk = new Buffer(chunk, encoding);
|
|
@@ -172,7 +159,7 @@ function readableAddChunk(stream, state, chunk, encoding, addToFront) {
|
|
|
var er = chunkInvalid(state, chunk);
|
|
var er = chunkInvalid(state, chunk);
|
|
|
if (er) {
|
|
if (er) {
|
|
|
stream.emit('error', er);
|
|
stream.emit('error', er);
|
|
|
- } else if (util.isNullOrUndefined(chunk)) {
|
|
|
|
|
|
|
+ } else if (chunk === null || chunk === undefined) {
|
|
|
state.reading = false;
|
|
state.reading = false;
|
|
|
if (!state.ended)
|
|
if (!state.ended)
|
|
|
onEofChunk(stream, state);
|
|
onEofChunk(stream, state);
|
|
@@ -187,25 +174,18 @@ function readableAddChunk(stream, state, chunk, encoding, addToFront) {
|
|
|
if (state.decoder && !addToFront && !encoding)
|
|
if (state.decoder && !addToFront && !encoding)
|
|
|
chunk = state.decoder.write(chunk);
|
|
chunk = state.decoder.write(chunk);
|
|
|
|
|
|
|
|
- if (!addToFront)
|
|
|
|
|
- state.reading = false;
|
|
|
|
|
-
|
|
|
|
|
- // if we want the data now, just emit it.
|
|
|
|
|
- if (state.flowing && state.length === 0 && !state.sync) {
|
|
|
|
|
- stream.emit('data', chunk);
|
|
|
|
|
- stream.read(0);
|
|
|
|
|
|
|
+ // update the buffer info.
|
|
|
|
|
+ state.length += state.objectMode ? 1 : chunk.length;
|
|
|
|
|
+ if (addToFront) {
|
|
|
|
|
+ state.buffer.unshift(chunk);
|
|
|
} else {
|
|
} 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);
|
|
|
|
|
|
|
+ state.reading = false;
|
|
|
|
|
+ state.buffer.push(chunk);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ if (state.needReadable)
|
|
|
|
|
+ emitReadable(stream);
|
|
|
|
|
+
|
|
|
maybeReadMore(stream, state);
|
|
maybeReadMore(stream, state);
|
|
|
}
|
|
}
|
|
|
} else if (!addToFront) {
|
|
} else if (!addToFront) {
|
|
@@ -237,7 +217,6 @@ Readable.prototype.setEncoding = function(enc) {
|
|
|
StringDecoder = require('string_decoder/').StringDecoder;
|
|
StringDecoder = require('string_decoder/').StringDecoder;
|
|
|
this._readableState.decoder = new StringDecoder(enc);
|
|
this._readableState.decoder = new StringDecoder(enc);
|
|
|
this._readableState.encoding = enc;
|
|
this._readableState.encoding = enc;
|
|
|
- return this;
|
|
|
|
|
};
|
|
};
|
|
|
|
|
|
|
|
// Don't raise the hwm > 128MB
|
|
// Don't raise the hwm > 128MB
|
|
@@ -261,7 +240,7 @@ function howMuchToRead(n, state) {
|
|
|
if (state.objectMode)
|
|
if (state.objectMode)
|
|
|
return n === 0 ? 0 : 1;
|
|
return n === 0 ? 0 : 1;
|
|
|
|
|
|
|
|
- if (isNaN(n) || util.isNull(n)) {
|
|
|
|
|
|
|
+ if (n === null || isNaN(n)) {
|
|
|
// only flow one buffer at a time
|
|
// only flow one buffer at a time
|
|
|
if (state.flowing && state.buffer.length)
|
|
if (state.flowing && state.buffer.length)
|
|
|
return state.buffer[0].length;
|
|
return state.buffer[0].length;
|
|
@@ -293,11 +272,12 @@ function howMuchToRead(n, state) {
|
|
|
|
|
|
|
|
// you can override either this method, or the async _read(n) below.
|
|
// you can override either this method, or the async _read(n) below.
|
|
|
Readable.prototype.read = function(n) {
|
|
Readable.prototype.read = function(n) {
|
|
|
- debug('read', n);
|
|
|
|
|
var state = this._readableState;
|
|
var state = this._readableState;
|
|
|
|
|
+ state.calledRead = true;
|
|
|
var nOrig = n;
|
|
var nOrig = n;
|
|
|
|
|
+ var ret;
|
|
|
|
|
|
|
|
- if (!util.isNumber(n) || n > 0)
|
|
|
|
|
|
|
+ if (typeof n !== 'number' || n > 0)
|
|
|
state.emittedReadable = false;
|
|
state.emittedReadable = false;
|
|
|
|
|
|
|
|
// if we're doing read(0) to trigger a readable event, but we
|
|
// if we're doing read(0) to trigger a readable event, but we
|
|
@@ -306,11 +286,7 @@ Readable.prototype.read = function(n) {
|
|
|
if (n === 0 &&
|
|
if (n === 0 &&
|
|
|
state.needReadable &&
|
|
state.needReadable &&
|
|
|
(state.length >= state.highWaterMark || state.ended)) {
|
|
(state.length >= state.highWaterMark || state.ended)) {
|
|
|
- debug('read: emitReadable', state.length, state.ended);
|
|
|
|
|
- if (state.length === 0 && state.ended)
|
|
|
|
|
- endReadable(this);
|
|
|
|
|
- else
|
|
|
|
|
- emitReadable(this);
|
|
|
|
|
|
|
+ emitReadable(this);
|
|
|
return null;
|
|
return null;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -318,9 +294,28 @@ Readable.prototype.read = function(n) {
|
|
|
|
|
|
|
|
// if we've ended, and we're now clear, then finish it up.
|
|
// if we've ended, and we're now clear, then finish it up.
|
|
|
if (n === 0 && state.ended) {
|
|
if (n === 0 && state.ended) {
|
|
|
|
|
+ ret = null;
|
|
|
|
|
+
|
|
|
|
|
+ // In cases where the decoder did not receive enough data
|
|
|
|
|
+ // to produce a full chunk, then immediately received an
|
|
|
|
|
+ // EOF, state.buffer will contain [<Buffer >, <Buffer 00 ...>].
|
|
|
|
|
+ // howMuchToRead will see this and coerce the amount to
|
|
|
|
|
+ // read to zero (because it's looking at the length of the
|
|
|
|
|
+ // first <Buffer > in state.buffer), and we'll end up here.
|
|
|
|
|
+ //
|
|
|
|
|
+ // This can only happen via state.decoder -- no other venue
|
|
|
|
|
+ // exists for pushing a zero-length chunk into state.buffer
|
|
|
|
|
+ // and triggering this behavior. In this case, we return our
|
|
|
|
|
+ // remaining data and end the stream, if appropriate.
|
|
|
|
|
+ if (state.length > 0 && state.decoder) {
|
|
|
|
|
+ ret = fromList(n, state);
|
|
|
|
|
+ state.length -= ret.length;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
if (state.length === 0)
|
|
if (state.length === 0)
|
|
|
endReadable(this);
|
|
endReadable(this);
|
|
|
- return null;
|
|
|
|
|
|
|
+
|
|
|
|
|
+ return ret;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// All the actual chunk generation logic needs to be
|
|
// All the actual chunk generation logic needs to be
|
|
@@ -347,23 +342,17 @@ Readable.prototype.read = function(n) {
|
|
|
|
|
|
|
|
// if we need a readable event, then we need to do some reading.
|
|
// if we need a readable event, then we need to do some reading.
|
|
|
var doRead = state.needReadable;
|
|
var doRead = state.needReadable;
|
|
|
- debug('need readable', doRead);
|
|
|
|
|
|
|
|
|
|
// if we currently have less than the highWaterMark, then also read some
|
|
// if we currently have less than the highWaterMark, then also read some
|
|
|
- if (state.length === 0 || state.length - n < state.highWaterMark) {
|
|
|
|
|
|
|
+ if (state.length - n <= state.highWaterMark)
|
|
|
doRead = true;
|
|
doRead = true;
|
|
|
- debug('length less than watermark', doRead);
|
|
|
|
|
- }
|
|
|
|
|
|
|
|
|
|
// however, if we've ended, then there's no point, and if we're already
|
|
// however, if we've ended, then there's no point, and if we're already
|
|
|
// reading, then it's unnecessary.
|
|
// reading, then it's unnecessary.
|
|
|
- if (state.ended || state.reading) {
|
|
|
|
|
|
|
+ if (state.ended || state.reading)
|
|
|
doRead = false;
|
|
doRead = false;
|
|
|
- debug('reading or ended', doRead);
|
|
|
|
|
- }
|
|
|
|
|
|
|
|
|
|
if (doRead) {
|
|
if (doRead) {
|
|
|
- debug('do read');
|
|
|
|
|
state.reading = true;
|
|
state.reading = true;
|
|
|
state.sync = true;
|
|
state.sync = true;
|
|
|
// if the length is currently zero, then we *need* a readable event.
|
|
// if the length is currently zero, then we *need* a readable event.
|
|
@@ -374,18 +363,18 @@ Readable.prototype.read = function(n) {
|
|
|
state.sync = false;
|
|
state.sync = false;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- // If _read pushed data synchronously, then `reading` will be false,
|
|
|
|
|
- // and we need to re-evaluate how much data we can return to the user.
|
|
|
|
|
|
|
+ // If _read called its callback synchronously, then `reading`
|
|
|
|
|
+ // will be false, and we need to re-evaluate how much data we
|
|
|
|
|
+ // can return to the user.
|
|
|
if (doRead && !state.reading)
|
|
if (doRead && !state.reading)
|
|
|
n = howMuchToRead(nOrig, state);
|
|
n = howMuchToRead(nOrig, state);
|
|
|
|
|
|
|
|
- var ret;
|
|
|
|
|
if (n > 0)
|
|
if (n > 0)
|
|
|
ret = fromList(n, state);
|
|
ret = fromList(n, state);
|
|
|
else
|
|
else
|
|
|
ret = null;
|
|
ret = null;
|
|
|
|
|
|
|
|
- if (util.isNull(ret)) {
|
|
|
|
|
|
|
+ if (ret === null) {
|
|
|
state.needReadable = true;
|
|
state.needReadable = true;
|
|
|
n = 0;
|
|
n = 0;
|
|
|
}
|
|
}
|
|
@@ -397,21 +386,21 @@ Readable.prototype.read = function(n) {
|
|
|
if (state.length === 0 && !state.ended)
|
|
if (state.length === 0 && !state.ended)
|
|
|
state.needReadable = true;
|
|
state.needReadable = true;
|
|
|
|
|
|
|
|
- // If we tried to read() past the EOF, then emit end on the next tick.
|
|
|
|
|
- if (nOrig !== n && state.ended && state.length === 0)
|
|
|
|
|
|
|
+ // If we happened to read() exactly the remaining amount in the
|
|
|
|
|
+ // buffer, and the EOF has been seen at this point, then make sure
|
|
|
|
|
+ // that we emit 'end' on the very next tick.
|
|
|
|
|
+ if (state.ended && !state.endEmitted && state.length === 0)
|
|
|
endReadable(this);
|
|
endReadable(this);
|
|
|
|
|
|
|
|
- if (!util.isNull(ret))
|
|
|
|
|
- this.emit('data', ret);
|
|
|
|
|
-
|
|
|
|
|
return ret;
|
|
return ret;
|
|
|
};
|
|
};
|
|
|
|
|
|
|
|
function chunkInvalid(state, chunk) {
|
|
function chunkInvalid(state, chunk) {
|
|
|
var er = null;
|
|
var er = null;
|
|
|
- if (!util.isBuffer(chunk) &&
|
|
|
|
|
- !util.isString(chunk) &&
|
|
|
|
|
- !util.isNullOrUndefined(chunk) &&
|
|
|
|
|
|
|
+ if (!Buffer.isBuffer(chunk) &&
|
|
|
|
|
+ 'string' !== typeof chunk &&
|
|
|
|
|
+ chunk !== null &&
|
|
|
|
|
+ chunk !== undefined &&
|
|
|
!state.objectMode) {
|
|
!state.objectMode) {
|
|
|
er = new TypeError('Invalid non-string/buffer chunk');
|
|
er = new TypeError('Invalid non-string/buffer chunk');
|
|
|
}
|
|
}
|
|
@@ -429,8 +418,12 @@ function onEofChunk(stream, state) {
|
|
|
}
|
|
}
|
|
|
state.ended = true;
|
|
state.ended = true;
|
|
|
|
|
|
|
|
- // emit 'readable' now to make sure it gets picked up.
|
|
|
|
|
- emitReadable(stream);
|
|
|
|
|
|
|
+ // if we've ended and we have some data left, then emit
|
|
|
|
|
+ // 'readable' now to make sure it gets picked up.
|
|
|
|
|
+ if (state.length > 0)
|
|
|
|
|
+ emitReadable(stream);
|
|
|
|
|
+ else
|
|
|
|
|
+ endReadable(stream);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// Don't emit readable right away in sync mode, because this can trigger
|
|
// Don't emit readable right away in sync mode, because this can trigger
|
|
@@ -439,22 +432,20 @@ function onEofChunk(stream, state) {
|
|
|
function emitReadable(stream) {
|
|
function emitReadable(stream) {
|
|
|
var state = stream._readableState;
|
|
var state = stream._readableState;
|
|
|
state.needReadable = false;
|
|
state.needReadable = false;
|
|
|
- if (!state.emittedReadable) {
|
|
|
|
|
- debug('emitReadable', state.flowing);
|
|
|
|
|
- state.emittedReadable = true;
|
|
|
|
|
- if (state.sync)
|
|
|
|
|
- process.nextTick(function() {
|
|
|
|
|
- emitReadable_(stream);
|
|
|
|
|
- });
|
|
|
|
|
- else
|
|
|
|
|
|
|
+ if (state.emittedReadable)
|
|
|
|
|
+ return;
|
|
|
|
|
+
|
|
|
|
|
+ state.emittedReadable = true;
|
|
|
|
|
+ if (state.sync)
|
|
|
|
|
+ process.nextTick(function() {
|
|
|
emitReadable_(stream);
|
|
emitReadable_(stream);
|
|
|
- }
|
|
|
|
|
|
|
+ });
|
|
|
|
|
+ else
|
|
|
|
|
+ emitReadable_(stream);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
function emitReadable_(stream) {
|
|
function emitReadable_(stream) {
|
|
|
- debug('emit readable');
|
|
|
|
|
stream.emit('readable');
|
|
stream.emit('readable');
|
|
|
- flow(stream);
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@@ -477,7 +468,6 @@ function maybeReadMore_(stream, state) {
|
|
|
var len = state.length;
|
|
var len = state.length;
|
|
|
while (!state.reading && !state.flowing && !state.ended &&
|
|
while (!state.reading && !state.flowing && !state.ended &&
|
|
|
state.length < state.highWaterMark) {
|
|
state.length < state.highWaterMark) {
|
|
|
- debug('maybeReadMore read 0');
|
|
|
|
|
stream.read(0);
|
|
stream.read(0);
|
|
|
if (len === state.length)
|
|
if (len === state.length)
|
|
|
// didn't get any data, stop spinning.
|
|
// didn't get any data, stop spinning.
|
|
@@ -512,7 +502,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
state.pipesCount += 1;
|
|
state.pipesCount += 1;
|
|
|
- debug('pipe count=%d opts=%j', state.pipesCount, pipeOpts);
|
|
|
|
|
|
|
|
|
|
var doEnd = (!pipeOpts || pipeOpts.end !== false) &&
|
|
var doEnd = (!pipeOpts || pipeOpts.end !== false) &&
|
|
|
dest !== process.stdout &&
|
|
dest !== process.stdout &&
|
|
@@ -526,14 +515,11 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
|
|
|
|
|
dest.on('unpipe', onunpipe);
|
|
dest.on('unpipe', onunpipe);
|
|
|
function onunpipe(readable) {
|
|
function onunpipe(readable) {
|
|
|
- debug('onunpipe');
|
|
|
|
|
- if (readable === src) {
|
|
|
|
|
- cleanup();
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ if (readable !== src) return;
|
|
|
|
|
+ cleanup();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
function onend() {
|
|
function onend() {
|
|
|
- debug('onend');
|
|
|
|
|
dest.end();
|
|
dest.end();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -545,7 +531,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
dest.on('drain', ondrain);
|
|
dest.on('drain', ondrain);
|
|
|
|
|
|
|
|
function cleanup() {
|
|
function cleanup() {
|
|
|
- debug('cleanup');
|
|
|
|
|
// cleanup event handlers once the pipe is broken
|
|
// cleanup event handlers once the pipe is broken
|
|
|
dest.removeListener('close', onclose);
|
|
dest.removeListener('close', onclose);
|
|
|
dest.removeListener('finish', onfinish);
|
|
dest.removeListener('finish', onfinish);
|
|
@@ -554,34 +539,19 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
dest.removeListener('unpipe', onunpipe);
|
|
dest.removeListener('unpipe', onunpipe);
|
|
|
src.removeListener('end', onend);
|
|
src.removeListener('end', onend);
|
|
|
src.removeListener('end', cleanup);
|
|
src.removeListener('end', cleanup);
|
|
|
- src.removeListener('data', ondata);
|
|
|
|
|
|
|
|
|
|
// if the reader is waiting for a drain event from this
|
|
// if the reader is waiting for a drain event from this
|
|
|
// specific writer, then it would cause it to never start
|
|
// specific writer, then it would cause it to never start
|
|
|
// flowing again.
|
|
// flowing again.
|
|
|
// So, if this is awaiting a drain, then we just call it now.
|
|
// So, if this is awaiting a drain, then we just call it now.
|
|
|
// If we don't know, then assume that we are waiting for one.
|
|
// If we don't know, then assume that we are waiting for one.
|
|
|
- if (state.awaitDrain &&
|
|
|
|
|
- (!dest._writableState || dest._writableState.needDrain))
|
|
|
|
|
|
|
+ if (!dest._writableState || dest._writableState.needDrain)
|
|
|
ondrain();
|
|
ondrain();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- src.on('data', ondata);
|
|
|
|
|
- function ondata(chunk) {
|
|
|
|
|
- debug('ondata');
|
|
|
|
|
- var ret = dest.write(chunk);
|
|
|
|
|
- if (false === ret) {
|
|
|
|
|
- debug('false write response, pause',
|
|
|
|
|
- src._readableState.awaitDrain);
|
|
|
|
|
- src._readableState.awaitDrain++;
|
|
|
|
|
- src.pause();
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
// if the dest has an error, then stop piping into it.
|
|
// if the dest has an error, then stop piping into it.
|
|
|
// however, don't suppress the throwing behavior for this.
|
|
// however, don't suppress the throwing behavior for this.
|
|
|
function onerror(er) {
|
|
function onerror(er) {
|
|
|
- debug('onerror', er);
|
|
|
|
|
unpipe();
|
|
unpipe();
|
|
|
dest.removeListener('error', onerror);
|
|
dest.removeListener('error', onerror);
|
|
|
if (EE.listenerCount(dest, 'error') === 0)
|
|
if (EE.listenerCount(dest, 'error') === 0)
|
|
@@ -605,14 +575,12 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
}
|
|
}
|
|
|
dest.once('close', onclose);
|
|
dest.once('close', onclose);
|
|
|
function onfinish() {
|
|
function onfinish() {
|
|
|
- debug('onfinish');
|
|
|
|
|
dest.removeListener('close', onclose);
|
|
dest.removeListener('close', onclose);
|
|
|
unpipe();
|
|
unpipe();
|
|
|
}
|
|
}
|
|
|
dest.once('finish', onfinish);
|
|
dest.once('finish', onfinish);
|
|
|
|
|
|
|
|
function unpipe() {
|
|
function unpipe() {
|
|
|
- debug('unpipe');
|
|
|
|
|
src.unpipe(dest);
|
|
src.unpipe(dest);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -621,8 +589,16 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
|
|
|
|
|
// start the flow if it hasn't been started already.
|
|
// start the flow if it hasn't been started already.
|
|
|
if (!state.flowing) {
|
|
if (!state.flowing) {
|
|
|
- debug('pipe resume');
|
|
|
|
|
- src.resume();
|
|
|
|
|
|
|
+ // the handler that waits for readable events after all
|
|
|
|
|
+ // the data gets sucked out in flow.
|
|
|
|
|
+ // This would be easier to follow with a .once() handler
|
|
|
|
|
+ // in flow(), but that is too slow.
|
|
|
|
|
+ this.on('readable', pipeOnReadable);
|
|
|
|
|
+
|
|
|
|
|
+ state.flowing = true;
|
|
|
|
|
+ process.nextTick(function() {
|
|
|
|
|
+ flow(src);
|
|
|
|
|
+ });
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
return dest;
|
|
return dest;
|
|
@@ -630,17 +606,65 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|
|
|
|
|
|
|
function pipeOnDrain(src) {
|
|
function pipeOnDrain(src) {
|
|
|
return function() {
|
|
return function() {
|
|
|
|
|
+ var dest = this;
|
|
|
var state = src._readableState;
|
|
var state = src._readableState;
|
|
|
- debug('pipeOnDrain', state.awaitDrain);
|
|
|
|
|
- if (state.awaitDrain)
|
|
|
|
|
- state.awaitDrain--;
|
|
|
|
|
- if (state.awaitDrain === 0 && EE.listenerCount(src, 'data')) {
|
|
|
|
|
- state.flowing = true;
|
|
|
|
|
|
|
+ state.awaitDrain--;
|
|
|
|
|
+ if (state.awaitDrain === 0)
|
|
|
flow(src);
|
|
flow(src);
|
|
|
- }
|
|
|
|
|
};
|
|
};
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+function flow(src) {
|
|
|
|
|
+ var state = src._readableState;
|
|
|
|
|
+ var chunk;
|
|
|
|
|
+ state.awaitDrain = 0;
|
|
|
|
|
+
|
|
|
|
|
+ function write(dest, i, list) {
|
|
|
|
|
+ var written = dest.write(chunk);
|
|
|
|
|
+ if (false === written) {
|
|
|
|
|
+ state.awaitDrain++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ while (state.pipesCount && null !== (chunk = src.read())) {
|
|
|
|
|
+
|
|
|
|
|
+ if (state.pipesCount === 1)
|
|
|
|
|
+ write(state.pipes, 0, null);
|
|
|
|
|
+ else
|
|
|
|
|
+ forEach(state.pipes, write);
|
|
|
|
|
+
|
|
|
|
|
+ src.emit('data', chunk);
|
|
|
|
|
+
|
|
|
|
|
+ // if anyone needs a drain, then we have to wait for that.
|
|
|
|
|
+ if (state.awaitDrain > 0)
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // if every destination was unpiped, either before entering this
|
|
|
|
|
+ // function, or in the while loop, then stop flowing.
|
|
|
|
|
+ //
|
|
|
|
|
+ // NB: This is a pretty rare edge case.
|
|
|
|
|
+ if (state.pipesCount === 0) {
|
|
|
|
|
+ state.flowing = false;
|
|
|
|
|
+
|
|
|
|
|
+ // if there were data event listeners added, then switch to old mode.
|
|
|
|
|
+ if (EE.listenerCount(src, 'data') > 0)
|
|
|
|
|
+ emitDataEvents(src);
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // at this point, no one needed a drain, so we just ran out of data
|
|
|
|
|
+ // on the next readable event, start it over again.
|
|
|
|
|
+ state.ranOut = true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+function pipeOnReadable() {
|
|
|
|
|
+ if (this._readableState.ranOut) {
|
|
|
|
|
+ this._readableState.ranOut = false;
|
|
|
|
|
+ flow(this);
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
|
|
|
Readable.prototype.unpipe = function(dest) {
|
|
Readable.prototype.unpipe = function(dest) {
|
|
|
var state = this._readableState;
|
|
var state = this._readableState;
|
|
@@ -661,6 +685,7 @@ Readable.prototype.unpipe = function(dest) {
|
|
|
// got a match.
|
|
// got a match.
|
|
|
state.pipes = null;
|
|
state.pipes = null;
|
|
|
state.pipesCount = 0;
|
|
state.pipesCount = 0;
|
|
|
|
|
+ this.removeListener('readable', pipeOnReadable);
|
|
|
state.flowing = false;
|
|
state.flowing = false;
|
|
|
if (dest)
|
|
if (dest)
|
|
|
dest.emit('unpipe', this);
|
|
dest.emit('unpipe', this);
|
|
@@ -675,6 +700,7 @@ Readable.prototype.unpipe = function(dest) {
|
|
|
var len = state.pipesCount;
|
|
var len = state.pipesCount;
|
|
|
state.pipes = null;
|
|
state.pipes = null;
|
|
|
state.pipesCount = 0;
|
|
state.pipesCount = 0;
|
|
|
|
|
+ this.removeListener('readable', pipeOnReadable);
|
|
|
state.flowing = false;
|
|
state.flowing = false;
|
|
|
|
|
|
|
|
for (var i = 0; i < len; i++)
|
|
for (var i = 0; i < len; i++)
|
|
@@ -702,11 +728,8 @@ Readable.prototype.unpipe = function(dest) {
|
|
|
Readable.prototype.on = function(ev, fn) {
|
|
Readable.prototype.on = function(ev, fn) {
|
|
|
var res = Stream.prototype.on.call(this, ev, fn);
|
|
var res = Stream.prototype.on.call(this, ev, fn);
|
|
|
|
|
|
|
|
- // If listening to data, and it has not explicitly been paused,
|
|
|
|
|
- // then call resume to start the flow of data on the next tick.
|
|
|
|
|
- if (ev === 'data' && false !== this._readableState.flowing) {
|
|
|
|
|
- this.resume();
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ if (ev === 'data' && !this._readableState.flowing)
|
|
|
|
|
+ emitDataEvents(this);
|
|
|
|
|
|
|
|
if (ev === 'readable' && this.readable) {
|
|
if (ev === 'readable' && this.readable) {
|
|
|
var state = this._readableState;
|
|
var state = this._readableState;
|
|
@@ -715,11 +738,7 @@ Readable.prototype.on = function(ev, fn) {
|
|
|
state.emittedReadable = false;
|
|
state.emittedReadable = false;
|
|
|
state.needReadable = true;
|
|
state.needReadable = true;
|
|
|
if (!state.reading) {
|
|
if (!state.reading) {
|
|
|
- var self = this;
|
|
|
|
|
- process.nextTick(function() {
|
|
|
|
|
- debug('readable nexttick read 0');
|
|
|
|
|
- self.read(0);
|
|
|
|
|
- });
|
|
|
|
|
|
|
+ this.read(0);
|
|
|
} else if (state.length) {
|
|
} else if (state.length) {
|
|
|
emitReadable(this, state);
|
|
emitReadable(this, state);
|
|
|
}
|
|
}
|
|
@@ -733,54 +752,63 @@ Readable.prototype.addListener = Readable.prototype.on;
|
|
|
// pause() and resume() are remnants of the legacy readable stream API
|
|
// pause() and resume() are remnants of the legacy readable stream API
|
|
|
// If the user uses them, then switch into old mode.
|
|
// If the user uses them, then switch into old mode.
|
|
|
Readable.prototype.resume = function() {
|
|
Readable.prototype.resume = function() {
|
|
|
- var state = this._readableState;
|
|
|
|
|
- if (!state.flowing) {
|
|
|
|
|
- debug('resume');
|
|
|
|
|
- state.flowing = true;
|
|
|
|
|
- if (!state.reading) {
|
|
|
|
|
- debug('resume read 0');
|
|
|
|
|
- this.read(0);
|
|
|
|
|
- }
|
|
|
|
|
- resume(this, state);
|
|
|
|
|
- }
|
|
|
|
|
- return this;
|
|
|
|
|
|
|
+ emitDataEvents(this);
|
|
|
|
|
+ this.read(0);
|
|
|
|
|
+ this.emit('resume');
|
|
|
};
|
|
};
|
|
|
|
|
|
|
|
-function resume(stream, state) {
|
|
|
|
|
- if (!state.resumeScheduled) {
|
|
|
|
|
- state.resumeScheduled = true;
|
|
|
|
|
- process.nextTick(function() {
|
|
|
|
|
- resume_(stream, state);
|
|
|
|
|
- });
|
|
|
|
|
- }
|
|
|
|
|
-}
|
|
|
|
|
-
|
|
|
|
|
-function resume_(stream, state) {
|
|
|
|
|
- state.resumeScheduled = false;
|
|
|
|
|
- stream.emit('resume');
|
|
|
|
|
- flow(stream);
|
|
|
|
|
- if (state.flowing && !state.reading)
|
|
|
|
|
- stream.read(0);
|
|
|
|
|
-}
|
|
|
|
|
-
|
|
|
|
|
Readable.prototype.pause = function() {
|
|
Readable.prototype.pause = function() {
|
|
|
- debug('call pause flowing=%j', this._readableState.flowing);
|
|
|
|
|
- if (false !== this._readableState.flowing) {
|
|
|
|
|
- debug('pause');
|
|
|
|
|
- this._readableState.flowing = false;
|
|
|
|
|
- this.emit('pause');
|
|
|
|
|
- }
|
|
|
|
|
- return this;
|
|
|
|
|
|
|
+ emitDataEvents(this, true);
|
|
|
|
|
+ this.emit('pause');
|
|
|
};
|
|
};
|
|
|
|
|
|
|
|
-function flow(stream) {
|
|
|
|
|
|
|
+function emitDataEvents(stream, startPaused) {
|
|
|
var state = stream._readableState;
|
|
var state = stream._readableState;
|
|
|
- debug('flow', state.flowing);
|
|
|
|
|
|
|
+
|
|
|
if (state.flowing) {
|
|
if (state.flowing) {
|
|
|
- do {
|
|
|
|
|
- var chunk = stream.read();
|
|
|
|
|
- } while (null !== chunk && state.flowing);
|
|
|
|
|
|
|
+ // https://github.com/isaacs/readable-stream/issues/16
|
|
|
|
|
+ throw new Error('Cannot switch to old mode now.');
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ var paused = startPaused || false;
|
|
|
|
|
+ var readable = false;
|
|
|
|
|
+
|
|
|
|
|
+ // convert to an old-style stream.
|
|
|
|
|
+ stream.readable = true;
|
|
|
|
|
+ stream.pipe = Stream.prototype.pipe;
|
|
|
|
|
+ stream.on = stream.addListener = Stream.prototype.on;
|
|
|
|
|
+
|
|
|
|
|
+ stream.on('readable', function() {
|
|
|
|
|
+ readable = true;
|
|
|
|
|
+
|
|
|
|
|
+ var c;
|
|
|
|
|
+ while (!paused && (null !== (c = stream.read())))
|
|
|
|
|
+ stream.emit('data', c);
|
|
|
|
|
+
|
|
|
|
|
+ if (c === null) {
|
|
|
|
|
+ readable = false;
|
|
|
|
|
+ stream._readableState.needReadable = true;
|
|
|
|
|
+ }
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ stream.pause = function() {
|
|
|
|
|
+ paused = true;
|
|
|
|
|
+ this.emit('pause');
|
|
|
|
|
+ };
|
|
|
|
|
+
|
|
|
|
|
+ stream.resume = function() {
|
|
|
|
|
+ paused = false;
|
|
|
|
|
+ if (readable)
|
|
|
|
|
+ process.nextTick(function() {
|
|
|
|
|
+ stream.emit('readable');
|
|
|
|
|
+ });
|
|
|
|
|
+ else
|
|
|
|
|
+ this.read(0);
|
|
|
|
|
+ this.emit('resume');
|
|
|
|
|
+ };
|
|
|
|
|
+
|
|
|
|
|
+ // now make it start, just in case it hadn't already.
|
|
|
|
|
+ stream.emit('readable');
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// wrap an old-style stream as the async data source.
|
|
// wrap an old-style stream as the async data source.
|
|
@@ -792,7 +820,6 @@ Readable.prototype.wrap = function(stream) {
|
|
|
|
|
|
|
|
var self = this;
|
|
var self = this;
|
|
|
stream.on('end', function() {
|
|
stream.on('end', function() {
|
|
|
- debug('wrapped end');
|
|
|
|
|
if (state.decoder && !state.ended) {
|
|
if (state.decoder && !state.ended) {
|
|
|
var chunk = state.decoder.end();
|
|
var chunk = state.decoder.end();
|
|
|
if (chunk && chunk.length)
|
|
if (chunk && chunk.length)
|
|
@@ -803,10 +830,14 @@ Readable.prototype.wrap = function(stream) {
|
|
|
});
|
|
});
|
|
|
|
|
|
|
|
stream.on('data', function(chunk) {
|
|
stream.on('data', function(chunk) {
|
|
|
- debug('wrapped data');
|
|
|
|
|
if (state.decoder)
|
|
if (state.decoder)
|
|
|
chunk = state.decoder.write(chunk);
|
|
chunk = state.decoder.write(chunk);
|
|
|
- if (!chunk || !state.objectMode && !chunk.length)
|
|
|
|
|
|
|
+
|
|
|
|
|
+ // don't skip over falsy values in objectMode
|
|
|
|
|
+ //if (state.objectMode && util.isNullOrUndefined(chunk))
|
|
|
|
|
+ if (state.objectMode && (chunk === null || chunk === undefined))
|
|
|
|
|
+ return;
|
|
|
|
|
+ else if (!state.objectMode && (!chunk || !chunk.length))
|
|
|
return;
|
|
return;
|
|
|
|
|
|
|
|
var ret = self.push(chunk);
|
|
var ret = self.push(chunk);
|
|
@@ -819,7 +850,8 @@ Readable.prototype.wrap = function(stream) {
|
|
|
// proxy all the other methods.
|
|
// proxy all the other methods.
|
|
|
// important when wrapping filters and duplexes.
|
|
// important when wrapping filters and duplexes.
|
|
|
for (var i in stream) {
|
|
for (var i in stream) {
|
|
|
- if (util.isFunction(stream[i]) && util.isUndefined(this[i])) {
|
|
|
|
|
|
|
+ if (typeof stream[i] === 'function' &&
|
|
|
|
|
+ typeof this[i] === 'undefined') {
|
|
|
this[i] = function(method) { return function() {
|
|
this[i] = function(method) { return function() {
|
|
|
return stream[method].apply(stream, arguments);
|
|
return stream[method].apply(stream, arguments);
|
|
|
}}(i);
|
|
}}(i);
|
|
@@ -835,7 +867,6 @@ Readable.prototype.wrap = function(stream) {
|
|
|
// when we try to consume some more bytes, simply unpause the
|
|
// when we try to consume some more bytes, simply unpause the
|
|
|
// underlying stream.
|
|
// underlying stream.
|
|
|
self._read = function(n) {
|
|
self._read = function(n) {
|
|
|
- debug('wrapped _read', n);
|
|
|
|
|
if (paused) {
|
|
if (paused) {
|
|
|
paused = false;
|
|
paused = false;
|
|
|
stream.resume();
|
|
stream.resume();
|
|
@@ -924,7 +955,7 @@ function endReadable(stream) {
|
|
|
if (state.length > 0)
|
|
if (state.length > 0)
|
|
|
throw new Error('endReadable called on non-empty stream');
|
|
throw new Error('endReadable called on non-empty stream');
|
|
|
|
|
|
|
|
- if (!state.endEmitted) {
|
|
|
|
|
|
|
+ if (!state.endEmitted && state.calledRead) {
|
|
|
state.ended = true;
|
|
state.ended = true;
|
|
|
process.nextTick(function() {
|
|
process.nextTick(function() {
|
|
|
// Check that we didn't get one last unshift.
|
|
// Check that we didn't get one last unshift.
|