stream: don't create unnecessary buffers in Readable

If there is an encoding, and we do 'stream.push(chunk, enc)', and the
encoding argument matches the stated encoding, then we're converting from
a string, to a buffer, and then back to a string.  Of course, this is a
completely pointless bit of work, so it's best to avoid it when we know
that we can do so safely.

doc/api/stream.markdown

@@ -131,13 +131,15 @@ TLS, may ignore this argument, and simply provide data whenever it
 becomes available.  There is no need, for example to "wait" until
 `size` bytes are available before calling `stream.push(chunk)`.
 
-### readable.push(chunk)
+### readable.push(chunk, [encoding])
 
 * `chunk` {Buffer | null | String} Chunk of data to push into the read queue
+* `encoding` {String} Encoding of String chunks.  Must be a valid
+  Buffer encoding, such as `'utf8'` or `'ascii'`
 * return {Boolean} Whether or not more pushes should be performed
 
 Note: **This function should be called by Readable implementors, NOT
-by consumers of Readable subclasses.**  The `_read()` function will not
+by consumers of Readable streams.**  The `_read()` function will not
 be called again until at least one `push(chunk)` call is made.  If no
 data is available, then you MAY call `push('')` (an empty string) to
 allow a future `_read` call, without adding any data to the queue.

lib/_stream_readable.js

@@ -83,10 +83,12 @@ function ReadableState(options, stream) {
   this.readingMore = false;
 
   this.decoder = null;
+  this.encoding = null;
   if (options.encoding) {
     if (!StringDecoder)
       StringDecoder = require('string_decoder').StringDecoder;
     this.decoder = new StringDecoder(options.encoding);
+    this.encoding = options.encoding;
   }
 }
 
@@ -106,19 +108,27 @@ function Readable(options) {
 // This returns true if the highWaterMark has not been hit yet,
 // similar to how Writable.write() returns true if you should
 // write() some more.
-Readable.prototype.push = function(chunk) {
+Readable.prototype.push = function(chunk, encoding) {
   var state = this._readableState;
-  if (typeof chunk === 'string' && !state.objectMode)
-    chunk = new Buffer(chunk, arguments[1]);
-  return readableAddChunk(this, state, chunk, false);
+
+  if (typeof chunk === 'string' && !state.objectMode) {
+    encoding = encoding || 'utf8';
+    if (encoding !== state.encoding) {
+      chunk = new Buffer(chunk, encoding);
+      encoding = '';
+    }
+  }
+
+  return readableAddChunk(this, state, chunk, encoding, false);
 };
 
+// Unshift should *always* be something directly out of read()
 Readable.prototype.unshift = function(chunk) {
   var state = this._readableState;
-  return readableAddChunk(this, state, chunk, true);
+  return readableAddChunk(this, state, chunk, '', true);
 };
 
-function readableAddChunk(stream, state, chunk, addToFront) {
+function readableAddChunk(stream, state, chunk, encoding, addToFront) {
   var er = chunkInvalid(state, chunk);
   if (er) {
     stream.emit('error', er);
@@ -134,7 +144,7 @@ function readableAddChunk(stream, state, chunk, addToFront) {
       var e = new Error('stream.unshift() after end event');
       stream.emit('error', e);
     } else {
-      if (state.decoder && !addToFront)
+      if (state.decoder && !addToFront && !encoding)
         chunk = state.decoder.write(chunk);
 
       // update the buffer info.
@@ -179,6 +189,7 @@ Readable.prototype.setEncoding = function(enc) {
   if (!StringDecoder)
     StringDecoder = require('string_decoder').StringDecoder;
   this._readableState.decoder = new StringDecoder(enc);
+  this._readableState.encoding = enc;
 };
 
 // Don't raise the hwm > 128MB

lib/_stream_transform.js

@@ -135,9 +135,9 @@ function Transform(options) {
   });
 }
 
-Transform.prototype.push = function(chunk) {
+Transform.prototype.push = function(chunk, encoding) {
   this._transformState.needTransform = false;
-  return Duplex.prototype.push.call(this, chunk);
+  return Duplex.prototype.push.call(this, chunk, encoding);
 };
 
 // This is the part where you do stuff!