Just use stream.push(outputChunk) instead.
constructor so that the buffering settings can be properly
initialized.
-### transform.\_transform(chunk, outputFn, callback)
+### transform.\_transform(chunk, callback)
* `chunk` {Buffer} The chunk to be transformed.
-* `outputFn` {Function} Call this function with any output data to be
- passed to the readable interface.
* `callback` {Function} Call this function (optionally with an error
argument) when you are done processing the supplied chunk.
to the readable portion of the interface. Do asynchronous I/O,
process things, and so on.
+Call `transform.push(outputChunk)` 0 or more times to generate output
+from this input chunk, depending on how much data you want to output
+as a result of this chunk.
+
Call the callback function only when the current chunk is completely
-consumed. Note that this may mean that you call the `outputFn` zero
-or more times, depending on how much data you want to output as a
-result of this chunk.
+consumed. Note that there may or may not be output as a result of any
+particular input chunk.
This method is prefixed with an underscore because it is internal to
the class that defines it, and should not be called directly by user
programs. However, you **are** expected to override this method in
your own extension classes.
-### transform.\_flush(outputFn, callback)
+### transform.\_flush(callback)
-* `outputFn` {Function} Call this function with any output data to be
- passed to the readable interface.
* `callback` {Function} Call this function (optionally with an error
argument) when you are done flushing any remaining data.
In those cases, you can implement a `_flush` method, which will be
called at the very end, after all the written data is consumed, but
before emitting `end` to signal the end of the readable side. Just
-like with `_transform`, call `outputFn` zero or more times, as
-appropriate, and call `callback` when the flush operation is complete.
+like with `_transform`, call `transform.push(chunk)` zero or more
+times, as appropriate, and call `callback` when the flush operation is
+complete.
This method is prefixed with an underscore because it is internal to
the class that defines it, and should not be called directly by user
SimpleProtocol.prototype = Object.create(
Transform.prototype, { constructor: { value: SimpleProtocol }});
-SimpleProtocol.prototype._transform = function(chunk, output, done) {
+SimpleProtocol.prototype._transform = function(chunk, done) {
if (!this._inBody) {
// check if the chunk has a \n\n
var split = -1;
this.emit('header', this.header);
// now, because we got some extra data, emit this first.
- output(b);
+ this.push(b);
}
} else {
// from there on, just provide the data to our consumer as-is.
- output(b);
+ this.push(b);
}
done();
};
Transform.call(this, options);
}
-PassThrough.prototype._transform = function(chunk, output, cb) {
+PassThrough.prototype._transform = function(chunk, cb) {
cb(null, chunk);
};
function TransformState(options, stream) {
var ts = this;
- this.output = function(chunk) {
- ts.needTransform = false;
- stream.push(chunk);
- };
this.afterTransform = function(er, data) {
return afterTransform(stream, er, data);
ts.writecb = null;
if (data !== null && data !== undefined)
- ts.output(data);
+ stream.push(data);
if (cb)
cb(er);
this.once('finish', function() {
if ('function' === typeof this._flush)
- this._flush(ts.output, function(er) {
+ this._flush(function(er) {
done(stream, er);
});
else
});
}
+Transform.prototype.push = function(chunk) {
+ this._transformState.needTransform = false;
+ return Duplex.prototype.push.call(this, chunk);
+};
+
// This is the part where you do stuff!
// override this function in implementation classes.
// 'chunk' is an input chunk.
//
-// Call `output(newChunk)` to pass along transformed output
-// to the readable side. You may call 'output' zero or more times.
+// Call `push(newChunk)` to pass along transformed output
+// to the readable side. You may call 'push' zero or more times.
//
// Call `cb(err)` when you are done with this chunk. If you pass
// an error, then that'll put the hurt on the whole operation. If you
var ts = this._transformState;
ts.writecb = cb;
ts.writechunk = chunk;
- if (ts.transforming)
- return;
- var rs = this._readableState;
- if (ts.needTransform || rs.needReadable || rs.length < rs.highWaterMark)
- this._read(rs.bufferSize);
+ if (!ts.transforming) {
+ var rs = this._readableState;
+ if (ts.needTransform ||
+ rs.needReadable ||
+ rs.length < rs.highWaterMark)
+ this._read(rs.bufferSize);
+ }
};
// Doesn't matter what the args are here.
if (ts.writechunk && ts.writecb && !ts.transforming) {
ts.transforming = true;
- this._transform(ts.writechunk, ts.output, ts.afterTransform);
- return;
+ this._transform(ts.writechunk, ts.afterTransform);
+ } else {
+ // mark that we need a transform, so that any data that comes in
+ // will get processed, now that we've asked for it.
+ ts.needTransform = true;
}
-
- // mark that we need a transform, so that any data that comes in
- // will get processed, now that we've asked for it.
- ts.needTransform = true;
};
util.inherits(Hash, stream.Transform);
-Hash.prototype._transform = function(chunk, output, callback) {
+Hash.prototype._transform = function(chunk, callback) {
this._binding.update(chunk);
callback();
};
-Hash.prototype._flush = function(output, callback) {
- output(this._binding.digest());
+Hash.prototype._flush = function(callback) {
+ this.push(this._binding.digest());
callback();
};
util.inherits(Cipher, stream.Transform);
-Cipher.prototype._transform = function(chunk, output, callback) {
- output(this._binding.update(chunk));
+Cipher.prototype._transform = function(chunk, callback) {
+ this.push(this._binding.update(chunk));
callback();
};
-Cipher.prototype._flush = function(output, callback) {
- output(this._binding.final());
+Cipher.prototype._flush = function(callback) {
+ this.push(this._binding.final());
callback();
};
return this._binding.reset();
};
-Zlib.prototype._flush = function(output, callback) {
- this._transform(null, output, callback);
+Zlib.prototype._flush = function(callback) {
+ this._transform(null, callback);
};
Zlib.prototype.flush = function(callback) {
ws.needDrain = true;
var self = this;
this.once('drain', function() {
- self._flush(ts.output, callback);
+ self._flush(callback);
});
- return;
- }
-
- this._flush(ts.output, callback || function() {});
+ } else
+ this._flush(callback || function() {});
};
Zlib.prototype.close = function(callback) {
});
};
-Zlib.prototype._transform = function(chunk, output, cb) {
+Zlib.prototype._transform = function(chunk, cb) {
var flushFlag;
var ws = this._writableState;
var ending = ws.ending || ws.ended;
var out = self._buffer.slice(self._offset, self._offset + have);
self._offset += have;
// serve some output to the consumer.
- output(out);
+ self.push(out);
}
// exhausted the output buffer, or used all the input create a new one.
});
var transformed = 0;
- tx._transform = function(chunk, output, cb) {
+ tx._transform = function(chunk, cb) {
transformed += chunk.length;
- output(chunk);
+ tx.push(chunk);
cb();
};
test('simple transform', function(t) {
var pt = new Transform;
- pt._transform = function(c, output, cb) {
+ pt._transform = function(c, cb) {
var ret = new Buffer(c.length);
ret.fill('x');
- output(ret);
+ pt.push(ret);
cb();
};
test('async passthrough', function(t) {
var pt = new Transform;
- pt._transform = function(chunk, output, cb) {
+ pt._transform = function(chunk, cb) {
setTimeout(function() {
- output(chunk);
+ pt.push(chunk);
cb();
}, 10);
};
var pt = new Transform;
// emit each chunk 2 times.
- pt._transform = function(chunk, output, cb) {
+ pt._transform = function(chunk, cb) {
setTimeout(function() {
- output(chunk);
+ pt.push(chunk);
setTimeout(function() {
- output(chunk);
+ pt.push(chunk);
cb();
}, 10)
}, 10);
// or whatever's left.
pt.state = '';
- pt._transform = function(chunk, output, cb) {
+ pt._transform = function(chunk, cb) {
if (!chunk)
chunk = '';
var s = chunk.toString();
setTimeout(function() {
this.state += s.charAt(0);
if (this.state.length === 3) {
- output(new Buffer(this.state));
+ pt.push(new Buffer(this.state));
this.state = '';
}
cb();
}.bind(this), 10);
};
- pt._flush = function(output, cb) {
+ pt._flush = function(cb) {
// just output whatever we have.
setTimeout(function() {
- output(new Buffer(this.state));
+ pt.push(new Buffer(this.state));
this.state = '';
cb();
}.bind(this), 10);
test('object transform (json parse)', function(t) {
console.error('json parse stream');
var jp = new Transform({ objectMode: true });
- jp._transform = function(data, output, cb) {
+ jp._transform = function(data, cb) {
try {
- output(JSON.parse(data));
+ jp.push(JSON.parse(data));
cb();
} catch (er) {
cb(er);
test('object transform (json stringify)', function(t) {
console.error('json parse stream');
var js = new Transform({ objectMode: true });
- js._transform = function(data, output, cb) {
+ js._transform = function(data, cb) {
try {
- output(JSON.stringify(data));
+ js.push(JSON.stringify(data));
cb();
} catch (er) {
cb(er);
}
util.inherits(TestWriter, stream.Writable);
-TestWriter.prototype._write = function (buffer, callback) {
+TestWriter.prototype._write = function (buffer, encoding, callback) {
console.log('write called');
// super slow write stream (callback never called)
};
}
util.inherits(TestWriter, stream.Writable);
-TestWriter.prototype._write = function(buffer, callback) {
+TestWriter.prototype._write = function(buffer, encoding, callback) {
callback(null);
};