streams2: Set flowing=true when flowing

lib/_stream_readable.js

@@ -159,8 +159,10 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   dest.emit('pipe', src);
 
   // start the flow.
-  if (!state.flowing)
+  if (!state.flowing) {
+    state.flowing = true;
     process.nextTick(flow.bind(null, src, pipeOpts));
+  }
 
   return dest;
 };
@@ -168,7 +170,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
 function flow(src, pipeOpts) {
   var state = src._readableState;
   var chunk;
-  var dest;
   var needDrain = 0;
 
   function ondrain() {