support for RX timeout

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

lib/incoming_form.js

@@ -41,6 +41,9 @@ 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);
@@ -76,6 +79,22 @@ IncomingForm.prototype.parse = function(req, cb) {
     return true;
   };
 
+  this.resetRxTimeout = function() {
+    if (this._rxTimeoutId !== null) {
+      clearTimeout(this._rxTimeoutId);
+    }
+    this._rxTimeoutId = setTimeout(function() {
+      this.emit('timeout');
+      req.connection.destroy();
+    }.bind(this), this.rxTimeout);
+  };
+
+  this.clearRxTimeout = function() {
+    if (this._rxTimeoutId !== null) {
+      clearTimeout(this._rxTimeoutId);
+    }
+  };
+
   // Setup callback first, so we don't miss anything from data events emitted
   // immediately.
   if (cb) {
@@ -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();
       });