Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

not working wip

  • Loading branch information...
commit d0c363c899c721825bebbbb31b79401e0565af5c 1 parent 24469cc
@isaacs authored
View
7 lib/_stream_readable.js
@@ -37,6 +37,7 @@ function ReadableState(options, stream) {
options.bufferSize : 16 * 1024;
// number of bytes that we ought to have before emitting 'readable'
+ // default to emitting for every chunk we get.
this.lowWaterMark = options.hasOwnProperty('lowWaterMark') ?
options.lowWaterMark : 0;
@@ -187,6 +188,7 @@ Readable.prototype.read = function(n) {
ret = null;
if (ret === null || ret.length === 0) {
+ // wanted something, but didn't get it.
state.needReadable = true;
n = 0;
}
@@ -219,9 +221,10 @@ function onread(er, chunk) {
// if we've ended and we have some data left, then emit
// 'readable' now to make sure it gets picked up.
if (!sync) {
- if (state.length > 0)
+ if (state.length > 0) {
+ state.needReadable = false;
this.emit('readable');
- else
+ } else
endReadable(this);
}
return;
View
24 lib/_stream_transform.js
@@ -62,12 +62,13 @@
// would be consumed, and then the rest would wait (un-transformed) until
// the results of the previous transformed chunk were consumed. Because
// the transform happens on-demand, it will only transform as much as is
-// necessary to fill the readable buffer to the specified lowWaterMark.
+// necessary to fill the readable buffer to the specified highWaterMark.
module.exports = Transform;
var Duplex = require('_stream_duplex');
var util = require('util');
+var assert = require('assert');
util.inherits(Transform, Duplex);
function TransformState() {
@@ -85,10 +86,10 @@ function Transform(options) {
// duplex defaults to making the lowWaterMark 0,
// but we should use the standard defaults for transform
// streams, since it's not a true duplex conversation.
- if (!options || !options.hasOwnProperty('lowWaterMark')) {
- this._readableState.lowWaterMark = 1024;
- this._writableState.lowWaterMark = 1024;
- }
+ // if (!options || !options.hasOwnProperty('lowWaterMark')) {
+ // this._readableState.lowWaterMark = 1024;
+ // this._writableState.lowWaterMark = 1024;
+ // }
// bind output so that it can be passed around as a regular function.
this._output = this._output.bind(this);
@@ -133,13 +134,14 @@ Transform.prototype._write = function(chunk, cb) {
if (ts.pendingReadCb) {
var readcb = ts.pendingReadCb;
ts.pendingReadCb = null;
- this._read(0, readcb);
+ this._read(rs.bufferSize, readcb);
}
// if we weren't waiting for it, but nothing is queued up, then
// still kick off a transform, just so it's there when the user asks.
- var doRead = rs.needReadable || rs.length <= rs.lowWaterMark;
+ var doRead = rs.needReadable || rs.length <= rs.highWaterMark;
if (doRead && !rs.reading) {
+ console.error('TS: call read(0)');
var ret = this.read(0);
if (ret !== null)
return cb(new Error('invalid stream transform state'));
@@ -154,6 +156,8 @@ Transform.prototype._read = function(n, readcb) {
if (ts.pendingReadCb)
throw new Error('_read while _read already in progress');
+ console.error('TS: _read(%j, readcb)', n);
+ assert(typeof n === 'number');
ts.pendingReadCb = readcb;
// if there's nothing pending, then we just wait.
@@ -188,6 +192,7 @@ Transform.prototype._output = function(chunk) {
var readcb = ts.pendingReadCb;
if (readcb) {
ts.pendingReadCb = null;
+ console.error('call readcb', this);
readcb(null, chunk);
return;
}
@@ -228,8 +233,9 @@ function done(er) {
// we may have gotten a 'null' read before, and since there is
// no more data coming from the writable side, we need to emit
// now so that the consumer knows to pick up the tail bits.
- if (rs.length && rs.needReadable)
+ if (rs.length && rs.needReadable) {
+ rs.needReadable = false;
this.emit('readable');
- else if (rs.length === 0)
+ } else if (rs.length === 0)
this.emit('end');
}
View
165 lib/_stream_writable.js
@@ -27,17 +27,31 @@ module.exports = Writable;
Writable.WritableState = WritableState;
var util = require('util');
+var assert = require('assert');
var Stream = require('stream');
util.inherits(Writable, Stream);
-function WritableState(options) {
+function WritableState(options, stream) {
options = options || {};
- this.highWaterMark = options.highWaterMark || 16 * 1024;
+
+ // the point at which write() starts returning false
this.highWaterMark = options.hasOwnProperty('highWaterMark') ?
- options.highWaterMark : 16 * 1024;
+ options.highWaterMark : 1024;
+
+ // the point that it has to get to before we call _write(chunk,cb)
+ // default to pushing everything out as fast as possible.
this.lowWaterMark = options.hasOwnProperty('lowWaterMark') ?
- options.lowWaterMark : 1024;
+ options.lowWaterMark : 0;
+
+ // cast to ints.
+ assert(typeof this.lowWaterMark === 'number');
+ assert(typeof this.highWaterMark === 'number');
+ this.lowWaterMark = ~~this.lowWaterMark;
+ this.highWaterMark = ~~this.highWaterMark;
+ assert(this.lowWaterMark >= 0);
+ assert(this.highWaterMark >= this.lowWaterMark);
+
this.needDrain = false;
// at the start of calling end()
this.ending = false;
@@ -59,7 +73,22 @@ function WritableState(options) {
// socket or file.
this.length = 0;
+ // a flag to see when we're in the middle of a write.
this.writing = false;
+
+ // a flag to be able to tell if the onwrite cb is called immediately,
+ // or on a later tick.
+ this.sync = false;
+
+ // the callback that's passed to _write(chunk,cb)
+ this.onwrite = onwrite.bind(stream);
+
+ // the callback that the user supplies to write(chunk,encoding,cb)
+ this.writecb = null;
+
+ // the amount that is being written when _write is called.
+ this.writelen = 0;
+
this.buffer = [];
}
@@ -69,7 +98,7 @@ function Writable(options) {
if (!(this instanceof Writable) && !(this instanceof Stream.Duplex))
return new Writable(options);
- this._writableState = new WritableState(options);
+ this._writableState = new WritableState(options, this);
// legacy.
this.writable = true;
@@ -77,8 +106,7 @@ function Writable(options) {
Stream.call(this);
}
-// Override this method for sync streams
-// override the _write(chunk, cb) method for async streams
+// Override this method or _write(chunk, cb)
Writable.prototype.write = function(chunk, encoding, cb) {
var state = this._writableState;
@@ -95,6 +123,8 @@ Writable.prototype.write = function(chunk, encoding, cb) {
return;
}
+ state.writecb = cb;
+
var l = chunk.length;
if (false === state.decodeStrings)
chunk = [chunk, encoding || 'utf8'];
@@ -117,72 +147,77 @@ Writable.prototype.write = function(chunk, encoding, cb) {
}
state.writing = true;
- var sync = true;
- this._write(chunk, writecb.bind(this));
- sync = false;
+ state.sync = true;
+ state.writelen = l;
+ this._write(chunk, state.onwrite);
+ state.sync = false;
return ret;
+};
- function writecb(er) {
- state.writing = false;
- if (er) {
- if (cb) {
- if (sync)
- process.nextTick(cb.bind(null, er));
- else
- cb(er);
- } else
- this.emit('error', er);
- return;
- }
- state.length -= l;
+function onwrite(er) {
+ var state = this._writableState;
+ var sync = state.sync;
+ var cb = state.writecb;
+ var l = state.writelen;
+ state.writing = false;
+ if (er) {
if (cb) {
- // don't call the cb until the next tick if we're in sync mode.
- // also, defer if we're about to write some more right now.
- if (sync || state.buffer.length)
- process.nextTick(cb);
- else
- cb();
- }
-
- if (state.length === 0 && (state.ended || state.ending)) {
- // emit 'finish' at the very end.
- state.finishing = true;
- this.emit('finish');
- state.finished = true;
- return;
- }
-
- // if there's something in the buffer waiting, then do that, too.
- if (state.buffer.length) {
- var chunkCb = state.buffer.shift();
- chunk = chunkCb[0];
- cb = chunkCb[1];
-
- if (false === state.decodeStrings)
- l = chunk[0].length;
+ if (sync)
+ process.nextTick(cb.bind(null, er));
else
- l = chunk.length;
-
- state.writing = true;
- this._write(chunk, writecb.bind(this));
- }
-
- if (state.length <= state.lowWaterMark && state.needDrain) {
- // Must force callback to be called on nextTick, so that we don't
- // emit 'drain' before the write() consumer gets the 'false' return
- // value, and has a chance to attach a 'drain' listener.
- process.nextTick(function() {
- if (!state.needDrain)
- return;
- state.needDrain = false;
- this.emit('drain');
- }.bind(this));
- }
+ cb(er);
+ } else
+ this.emit('error', er);
+ return;
+ }
+ state.length -= l;
+
+ if (cb) {
+ // don't call the cb until the next tick if we're in sync mode.
+ // also, defer if we're about to write some more right now.
+ if (sync || state.buffer.length)
+ process.nextTick(cb);
+ else
+ cb();
}
-};
+ if (state.length === 0 && (state.ended || state.ending)) {
+ // emit 'finish' at the very end.
+ state.finishing = true;
+ this.emit('finish');
+ state.finished = true;
+ return;
+ }
+
+ // if there's something in the buffer waiting, then do that, too.
+ if (state.buffer.length) {
+ var chunkCb = state.buffer.shift();
+ var chunk = chunkCb[0];
+ cb = chunkCb[1];
+
+ if (false === state.decodeStrings)
+ l = chunk[0].length;
+ else
+ l = chunk.length;
+
+ state.writing = true;
+ this._write(chunk, state.onwrite);
+ }
+
+ if (state.length <= state.lowWaterMark && state.needDrain) {
+ // Must force callback to be called on nextTick, so that we don't
+ // emit 'drain' before the write() consumer gets the 'false' return
+ // value, and has a chance to attach a 'drain' listener.
+ process.nextTick(function() {
+ if (!state.needDrain)
+ return;
+ state.needDrain = false;
+ this.emit('drain');
+ }.bind(this));
+ }
+}
Writable.prototype._write = function(chunk, cb) {
process.nextTick(cb.bind(this, new Error('not implemented')));
View
37 test/simple/test-stream2-transform.js
@@ -37,7 +37,7 @@ function run() {
var name = next[0];
var fn = next[1];
- console.log('# %s', name);
+ console.error('# %s', name);
fn({
same: assert.deepEqual,
equal: assert.equal,
@@ -203,12 +203,14 @@ test('assymetric transform (compress)', function(t) {
test('passthrough event emission', function(t) {
var pt = new PassThrough({
- lowWaterMark: 0
+ lowWaterMark: 0,
+ highWaterMark: 0
});
var emits = 0;
pt.on('readable', function() {
var state = pt._readableState;
- console.error('>>> emit readable %d', emits);
+ console.error(state);
+ console.error('emit readable ' + emits)
emits++;
});
@@ -250,7 +252,6 @@ test('passthrough event emission reordered', function(t) {
var pt = new PassThrough;
var emits = 0;
pt.on('readable', function() {
- console.error('emit readable', emits)
emits++;
});
@@ -259,24 +260,36 @@ test('passthrough event emission reordered', function(t) {
t.equal(pt.read(5).toString(), 'foogb');
t.equal(pt.read(5), null);
+ console.error('got null', pt);
console.error('need emit 0');
pt.once('readable', function() {
- t.equal(pt.read(5).toString(), 'arkba');
- t.equal(pt.read(5).toString(), 'zykue');
+ console.error('>>> readable', pt);
+ console.trace('>>> readable');
+ process.nextTick(function() {
+ console.error('>>> readable nt', pt);
+ });
+ return;
+
+ t.equal(pt.read(5) + '', 'arkba');
t.equal(pt.read(5), null);
console.error('need emit 1');
pt.once('readable', function() {
- t.equal(pt.read(5).toString(), 'l');
- t.equal(pt.read(5), null);
-
- t.equal(emits, 2);
- t.end();
+ t.equal(pt.read(5) + '', 'zykue');
+ pt.once('readable', function() {
+ t.equal(pt.read(5).toString(), 'l');
+ t.equal(pt.read(5), null);
+
+ t.equal(emits, 2);
+ t.end();
+ });
+ pt.end();
});
- pt.end();
});
+ console.error('\n\n\n\nabout to trigger readable...');
pt.write(new Buffer('bazy'));
+ console.error('^^^^ theres the problem\n\n\n\n\n\n\n');
pt.write(new Buffer('kuel'));
});

0 comments on commit d0c363c

Please sign in to comment.
Something went wrong with that request. Please try again.