Ensure that readable stream gets correctly paused. Fixes #8

lib/webhdfs.js

@@ -521,6 +521,7 @@ WebHDFS.prototype.createWriteStream = function createWriteStream (path, append,
 
  var self = this;
  var stream = null;
+ var canResume = true;
  var params = extend({
  method: append ? 'POST' : 'PUT',
  url: endpoint,
@@ -544,6 +545,8 @@ WebHDFS.prototype.createWriteStream = function createWriteStream (path, append,
  }
  });
 
+ canResume = true; // Enable resume
+
  stream.pipe(upload);
  stream.resume();
  }
@@ -563,13 +566,23 @@ WebHDFS.prototype.createWriteStream = function createWriteStream (path, append,
  });
 
  req.on('pipe', function onPipe (src) {
- // Unpipe initial request
- src.unpipe(req);
- req.end();
-
  // Pause read stream
  stream = src;
  stream.pause();
+
+ // This is not an elegant solution but here we go
+ // Basically we don't allow pipe() method to resume reading input
+ // and set internal _readableState.flowing to false
+ canResume = false;
+ stream.on('resume', function () {
+ if (!canResume) {
+ stream._readableState.flowing = false;
+ }
+ });
+
+ // Unpipe initial request
+ src.unpipe(req);
+ req.end();
  });
 
  return req;

package.json

@@ -1,6 +1,6 @@
 {
  "name": "webhdfs",
- "version": "0.2.0",
+ "version": "0.2.1",
  "description": "Node.js WebHDFS REST API client",
  "main": "lib/webhdfs.js",
  "scripts": {
@@ -27,8 +27,9 @@
  "webhdfs-proxy-memory": "^0.1.2"
  },
  "dependencies": {
- "request": "~2.40.0",
- "extend": "~1.3.0",
- "buffer-stream-reader": "~0.1.1"
+ "buffer-stream-reader": "~0.1.1",
+ "extend": "^3.0.0",
+ "mocha": "^2.3.0",
+ "request": "^2.61.0"
  }
 }