stream: add isFlowing method

doc/api/stream.markdown

@@ -328,6 +328,24 @@ readable.resume()
 readable.isPaused() // === false
 ```
 
+#### readable.isFlowing()
+
+* Return: `Boolean`
+
+This method returns whether or not the `readable` is in flowing mode.
+
+```javascript
+var readable = new stream.Readable
+
+readable.isFlowing() // === false
+readable.resume()
+readable.isFlowing() // === true
+readable.pause()
+readable.isFlowing() // === false
+```
+
+
+
 #### readable.pipe(destination[, options])
 
 * `destination` {[Writable][] Stream} The destination for writing data

lib/_stream_readable.js

@@ -116,6 +116,10 @@ Readable.prototype.isPaused = function() {
   return this._readableState.flowing === false;
 };
 
+Readable.prototype.isFlowing = function() {
+  return this._readableState.flowing === true;
+};
+
 function readableAddChunk(stream, state, chunk, encoding, addToFront) {
   var er = chunkInvalid(state, chunk);
   if (er) {

lib/child_process.js

@@ -1058,7 +1058,7 @@ function flushStdio(subprocess) {
   if (subprocess.stdio == null) return;
   subprocess.stdio.forEach(function(stream, fd, stdio) {
     if (!stream || !stream.readable || stream._consuming ||
-        stream._readableState.flowing)
+        stream.isFlowing())
       return;
     stream.resume();
   });

test/parallel/test-stream-isflowing.js

@@ -0,0 +1,20 @@
+var assert = require('assert');
+var common = require('../common');
+
+var stream = require('stream');
+
+var readable = new stream.Readable;
+
+readable._read = function(){};
+
+assert.ok(!readable.isFlowing());
+
+// make the stream start flowing...
+readable.on('data', function(){});
+
+assert.ok(readable.isFlowing());
+
+readable.pause();
+assert.ok(!readable.isFlowing());
+readable.resume();
+assert.ok(readable.isFlowing());