support for RX timeout

Don't hang indefinitely when the client stops sending data to us. Close the
connection when timeout is reached.

Author: dankeder

lib/incoming_form.js

@@ -41,11 +41,30 @@ function IncomingForm(opts) {
   this._fieldsSize = 0;
   this.openedFiles = [];
 
+  this._rxTimeoutId = null;
+  this.rxTimeout = 10 * 60 * 1000;  // 10 minutes (in ms)
+
   return this;
 }
 util.inherits(IncomingForm, EventEmitter);
 exports.IncomingForm = IncomingForm;
 
+IncomingForm.prototype.resetRxTimeout = function() {
+  if (this._rxTimeoutId !== null) {
+    clearTimeout(this._rxTimeoutId);
+  }
+  this._rxTimeoutId = setTimeout(function() {
+    this.emit('timeout');
+    req.connection.destroy();
+  }.bind(this), this.rxTimeout);
+};
+
+IncomingForm.prototype.clearRxTimeout = function() {
+  if (this._rxTimeoutId !== null) {
+    clearTimeout(this._rxTimeoutId);
+  }
+};
+
 IncomingForm.prototype.parse = function(req, cb) {
   this.pause = function() {
     try {
@@ -208,6 +227,8 @@ IncomingForm.prototype.handlePart = function(part) {
     hash: self.hash
   });
 
+  this.resetRxTimeout();
+
   this.emit('fileBegin', part.name, file);
 
   file.open();
@@ -217,6 +238,7 @@ IncomingForm.prototype.handlePart = function(part) {
     if (buffer.length == 0) {
       return;
     }
+    self.resetRxTimeout();
     self.pause();
     file.write(buffer, function() {
       self.resume();
@@ -226,6 +248,7 @@ IncomingForm.prototype.handlePart = function(part) {
   part.on('end', function() {
     file.end(function() {
       self._flushing--;
+      self.clearRxTimeout();
       self.emit('file', part.name, file);
       self._maybeEnd();
     });
@@ -464,6 +487,8 @@ IncomingForm.prototype._initOctetStream = function() {
     type: mime
   });
 
+  this.resetRxTimeout();
+
   this.emit('fileBegin', filename, file);
   file.open();
 
@@ -477,6 +502,7 @@ IncomingForm.prototype._initOctetStream = function() {
   var outstandingWrites = 0;
 
   self._parser.on('data', function(buffer){
+    self.resetRxTimeout();
     self.pause();
     outstandingWrites++;
 
@@ -496,6 +522,7 @@ IncomingForm.prototype._initOctetStream = function() {
 
     var done = function(){
       file.end(function() {
+        self.clearRxTimeout();
         self.emit('file', 'file', file);
         self._maybeEnd();
       });