1berry/nodejs
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

stream: Use push() for readable.wrap()

Browse Source
This commit is contained in:
isaacs 2013-01-07 19:40:08 -08:00
parent a993f740f0
commit 530585b2d1
1 changed files with 9 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
lib/_stream_readable.js
View File

@ -611,16 +611,11 @@ Readable.prototype.wrap = function(stream) {
state.ended = true; state.ended = true;
if (state.decoder) { if (state.decoder) {
var chunk = state.decoder.end(); var chunk = state.decoder.end();
if (chunk && chunk.length) { if (chunk && chunk.length)
state.buffer.push(chunk); self.push(chunk);
state.length += chunk.length;
}
} }
if (state.length > 0) self.push(null);
self.emit('readable');
else
endReadable(self);
}); });
stream.on('data', function(chunk) { stream.on('data', function(chunk) {
@ -629,12 +624,8 @@ Readable.prototype.wrap = function(stream) {
if (!chunk || !chunk.length) if (!chunk || !chunk.length)
return; return;
state.buffer.push(chunk); var ret = self.push(chunk);
state.length += chunk.length; if (!ret) {
self.emit('readable');
// if not consumed, then pause the stream.
if (state.length > state.lowWaterMark && !paused) {
paused = true; paused = true;
stream.pause(); stream.pause();
} }
@ -657,40 +648,13 @@ Readable.prototype.wrap = function(stream) {
stream.on(ev, self.emit.bind(self, ev)); stream.on(ev, self.emit.bind(self, ev));
}); });
// consume some bytes. if not all is consumed, then // when we try to consume some more bytes, simply unpause the
// pause the underlying stream. // underlying stream.
this.read = function(n) { self._read = function(n, cb) {
if (state.length === 0) { if (paused) {
state.needReadable = true;
return null;
}
if (isNaN(n) || n <= 0)
n = state.length;
if (n > state.length) {
if (!state.ended) {
state.needReadable = true;
return null;
} else
n = state.length;
}
var ret = fromList(n, state.buffer, state.length, !!state.decoder);
state.length -= n;
if (state.length === 0 && !state.ended)
state.needReadable = true;
if (state.length <= state.lowWaterMark && paused) {
stream.resume(); stream.resume();
paused = false; paused = false;
} }
if (state.length === 0 && state.ended)
endReadable(this);
return ret;
}; };
}; };