implement stream backpressure

Author: vdiez

lib/api/createReadStream.js

@@ -45,39 +45,53 @@ module.exports = function createReadStream(path, options, cb) {
         });
       };
     }
+
     var running = false;
+
+    function read(size) {
+      request(
+          'read',
+          {
+            FileId: file.FileId,
+            Length: Math.min(MAX_READ_LENGTH, size, end - offset),
+            Offset: new BigInt(8, offset).toBuffer(),
+          },
+          connection,
+          function(err, content) {
+            if (err != null) {
+              return process.nextTick(stream.emit.bind(stream, 'error', err));
+            }
+
+            offset += content.length;
+            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() {
+            ? close(function() {
               stream.push(null);
             })
-          : stream.push(null);
+            : stream.push(null);
+      }
+      else {
+        running = true;
+        read(size)
       }
-
-      running = true;
-      request(
-        'read',
-        {
-          FileId: file.FileId,
-          Length: Math.min(MAX_READ_LENGTH, size, end - offset),
-          Offset: new BigInt(8, offset).toBuffer(),
-        },
-        connection,
-        function(err, content) {
-          running = false;
-          if (err != null) {
-            return process.nextTick(stream.emit.bind(stream, 'error', err));
-          }
-
-          offset += content.length;
-          stream.push(content);
-        }
-      );
     };
     cb(null, stream);
   }