implement stream backpressure

Author: vdiez

lib/api/createReadStream.js

@@ -45,21 +45,10 @@ module.exports = function createReadStream(path, options, cb) {
         });
       };
     }
-    var running = false;
-    stream._read = function(size) {
-      if (running) {
-        return;
-      }
 
-      if (offset >= end) {
-        return shouldClose
-          ? close(function() {
-              stream.push(null);
-            })
-          : stream.push(null);
-      }
+    var running = false;
 
-      running = true;
+    function read(size) {
       request(
         'read',
         {
@@ -69,15 +58,37 @@ module.exports = function createReadStream(path, options, cb) {
         },
         connection,
         function(err, content) {
-          running = false;
           if (err != null) {
             return process.nextTick(stream.emit.bind(stream, 'error', err));
           }
 
           offset += content.length;
-          stream.push(content);
+          if (stream.push(content)) {
+            if (end - offset === 0) {
+              running = false;
+              stream.push(null);
+            } else read(size);
+          } else {
+            running = false;
+          }
         }
       );
+    }
+    stream._read = function(size) {
+      if (running) {
+        return;
+      }
+
+      if (offset >= end) {
+        return shouldClose
+          ? close(function() {
+              stream.push(null);
+            })
+          : stream.push(null);
+      } else {
+        running = true;
+        read(size);
+      }
     };
     cb(null, stream);
   }