stream: fix no data on partial decode

Before this commit, it was possible to push a partial character
to a readable stream where it was decoded as an empty string and
then added to the internal buffer. This caused the stream to not
emit any data, even when the rest of the character bytes were pushed
separately, because of a non-zero length check of the first chunk in
the internal buffer.

Fixes: #5223
PR-URL: #5226
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: James M Snell <[email protected]>

lib/_stream_readable.js

@@ -136,26 +136,33 @@ function readableAddChunk(stream, state, chunk, encoding, addToFront) {
       const e = new Error('stream.unshift() after end event');
       stream.emit('error', e);
     } else {
-      if (state.decoder && !addToFront && !encoding)
+      var skipAdd;
+      if (state.decoder && !addToFront && !encoding) {
         chunk = state.decoder.write(chunk);
+        skipAdd = (!state.objectMode && chunk.length === 0);
+      }
 
       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);
-      } 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);
+      // Don't add to the buffer if we've decoded to an empty string chunk and
+      // we're not in object mode
+      if (!skipAdd) {
+        // if we want the data now, just emit it.
+        if (state.flowing && state.length === 0 && !state.sync) {
+          stream.emit('data', chunk);
+          stream.read(0);
+        } 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);
+        }
       }
 
       maybeReadMore(stream, state);

test/parallel/test-stream2-decode-partial.js

@@ -0,0 +1,23 @@
+'use strict';
+require('../common');
+const Readable = require('_stream_readable');
+const assert = require('assert');
+
+var buf = '';
+const euro = new Buffer([0xE2, 0x82, 0xAC]);
+const cent = new Buffer([0xC2, 0xA2]);
+const source = Buffer.concat([euro, cent]);
+
+const readable = Readable({ encoding: 'utf8' });
+readable.push(source.slice(0, 2));
+readable.push(source.slice(2, 4));
+readable.push(source.slice(4, 6));
+readable.push(null);
+
+readable.on('data', function(data) {
+  buf += data;
+});
+
+process.on('exit', function() {
+  assert.strictEqual(buf, '€¢');
+});