FIX MEMORY LEAK.

Log all of the internals of the connections to track them. Then nulled out the two exit points for a connection.

Author: trestletech

lib/node-http-proxy.js

@@ -212,7 +212,7 @@ exports.buffer = function (obj) {
   return {
     end: function () {
       obj.removeListener('data', onData);
-      obj.removeListener('end', onEnd);
+      obj.removeListener('end', onEnd);      
     },
     destroy: function () {
       this.end();
@@ -281,6 +281,8 @@ exports._getAgent = function _getAgent (options) {
   // require('http-proxy').setMaxSockets() should override http's default
   // configuration value (which is pretty low).
   options.maxSockets = options.maxSockets || maxSockets;
+  options.maxSockets = 1;
+  http.globalAgent.maxSockets = 1;
   agent = new Agent(options);
 
   return agent;

lib/node-http-proxy/http-proxy.js

@@ -103,13 +103,43 @@ var HttpProxy = exports.HttpProxy = function (options) {
 // Inherit from events.EventEmitter
 util.inherits(HttpProxy, events.EventEmitter);
 
+function ReqCanary(){
+  this.___events = [];
+}
+function ResCanary(){
+  this.___events = [];
+}
+
+var cc = {};
+
+var logEvt = function(canary, event){
+  if (canary && canary.___events){
+    canary.___events.push(event);
+  } else{
+    console.log("No canary for event : " + event);
+  }
+  
+  if (!cc[event]){
+    cc[event] = 1;
+  } else{
+    cc[event]+=1;
+  }
+
+  console.dir(cc);
+}
+
 //
 // ### function proxyRequest (req, res, buffer)
 // #### @req {ServerRequest} Incoming HTTP Request to proxy.
 // #### @res {ServerResponse} Outgoing HTTP Request to write proxied data to.
 // #### @buffer {Object} Result from `httpProxy.buffer(req)`
 //
 HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
+  req.canary = new ReqCanary();
+  res.canary = new ResCanary();
+
+  logEvt(res.canary, "Init");
+
   var self = this,
       errState = false,
       outgoing = new(this.target.base),
@@ -166,6 +196,8 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
   //
   this.emit('start', req, res, this.target);
 
+  logEvt(res.canary, "Start");
+
   //
   // #### function proxyError (err)
   // #### @err {Error} Error contacting the proxy target
@@ -175,6 +207,8 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
   function proxyError(err) {
     errState = true;
 
+    logEvt(res.canary, "ProxyError");
+
     //
     // Emit an `error` event, allowing the application to use custom
     // error handling. The error handler should end the response.
@@ -183,6 +217,8 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
       return;
     }
 
+    logEvt(res.canary, "Writing 500");
+
     res.writeHead(500, { 'Content-Type': 'text/plain' });
 
     if (req.method !== 'HEAD') {
@@ -198,7 +234,23 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
       }
     }
 
-    try { res.end() }
+    try { 
+      res.end(); 
+      logEvt(res.canary, "Ended properly"); 
+      //CLEANUP
+      if (res){
+        res.removeAllListeners();
+        res = null;
+      }
+      if (req){
+        req.removeAllListeners();
+        req = null;
+      }
+      if (response){
+        response.removeAllListeners();
+        response = null;
+      }
+    }
     catch (ex) { console.error("res.end error: %s", ex.message) }
   }
 
@@ -213,12 +265,16 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
   outgoing.method     = req.method;
   outgoing.path       = req.url;
   outgoing.headers    = req.headers;
+  outgoing.agent      = false;
 
   //
   // Open new HTTP request to internal resource with will act
   // as a reverse proxy pass
   //
   reverseProxy = this.target.protocol.request(outgoing, function (response) {
+    
+    logEvt(res.canary, "Reverse proxying");
+
     //
     // Process the `reverseProxy` `response` when it's received.
     //
@@ -244,6 +300,7 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
     //
     var ended = false;
     response.on('close', function () {
+      logEvt(res.canary, "Close, emiting 'end'");
       if (!ended) { response.emit('end') }
     });
 
@@ -255,15 +312,48 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
     // This code makes sure that we flush pending data.
     //
     response.connection.on('end', function () {
-      if (response.readable && response.resume) {
+      if (res)
+        logEvt(res.canary, 'response connection ended');
+      
+      if (response && response.readable && response.resume) {
+        if (res)
+          logEvt(res.canary, 'response resuming');
+
         response.resume();
       }
+      // CLEANUP?
+      if (res){
+        res.removeAllListeners();
+        res = null;
+      }
+      if (req){
+        req.removeAllListeners();
+        req = null;
+      }
+      if (response){
+        response.removeAllListeners();
+        response = null;  
+      }
     });
 
     response.on('end', function () {
+      logEvt(res.canary, 'response ended');
       ended = true;
       if (!errState) {
-        try { res.end() }
+        try { 
+          res.end(); 
+          logEvt(res.canary, 'end res'); 
+          // CLEANUP?
+          res.removeAllListeners();
+          // Note that nulling this will prevent the response resuming' event from
+          // coming in, since it won't fire on a null event.
+          res = null;
+          req.removeAllListeners();
+          req = null;
+          response.removeAllListeners();
+          response = null;
+          
+        }
         catch (ex) { console.error("res.end error: %s", ex.message) }
 
         // Emit the `end` event now that we have completed proxying
@@ -285,13 +375,15 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
     res.writeHead(response.statusCode);
 
     function ondata(chunk) {
+      logEvt(res.canary, "Data");
       if (res.writable) {
         // Only pause if the underlying buffers are full,
         // *and* the connection is not in 'closing' state.
         // Otherwise, the pause will cause pending data to
         // be discarded and silently lost.
         if (false === res.write(chunk) && response.pause
             && response.connection.readable) {
+          logEvt(res.canary, "Pausing due to full buffer");
           response.pause();
         }
       }
@@ -300,7 +392,9 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
     response.on('data', ondata);
 
     function ondrain() {
+      logEvt(res.canary, "On drain");
       if (response.readable && response.resume) {
+        logEvt(res.canary, "On drain resuming.");
         response.resume();
       }
     }
@@ -318,6 +412,7 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
 
   // Set a timeout on the socket if `this.timeout` is specified.
   reverseProxy.once('socket', function (socket) {
+    logEvt(res.canary, "Socket");
     if (self.timeout) {
       socket.setTimeout(self.timeout);
     }
@@ -332,6 +427,7 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
   // If `req` is aborted, we abort our `reverseProxy` request as well.
   //
   req.on('aborted', function () {
+    logEvt(res.canary, "Aborted");
     reverseProxy.abort();
   });
 
@@ -340,11 +436,13 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
   // `req` write it to the `reverseProxy` request.
   //
   req.on('data', function (chunk) {
+    logEvt(res.canary, 'req data');
     if (!errState) {
       var flushed = reverseProxy.write(chunk);
       if (!flushed) {
         req.pause();
         reverseProxy.once('drain', function () {
+          logEvt(res.canary, "Drain");
           try { req.resume() }
           catch (er) { console.error("req.resume error: %s", er.message) }
         });
@@ -354,6 +452,7 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
         // happened on its own.
         //
         setTimeout(function () {
+          logEvt(res.canary, "Force drain");
           reverseProxy.emit('drain');
         }, 100);
       }
@@ -365,22 +464,29 @@ HttpProxy.prototype.proxyRequest = function (req, res, buffer) {
   // request unless we have entered an error state.
   //
   req.on('end', function () {
+    logEvt(res.canary, "Req ended");
     if (!errState) {
       reverseProxy.end();
+    } else{
+      logEvt(res.canary, "Req errored end");
     }
   });
 
   //Aborts reverseProxy if client aborts the connection.
   req.on('close', function () {
+    logEvt(res.canary, "Req closed");
     if (!errState) {
       reverseProxy.abort();
+    } else {
+      logEvt(res.canary, "Req errored close");
     }
   });
 
   //
   // If we have been passed buffered data, resume it.
   //
   if (buffer) {
+    logEvt(res.canary, "Resuming buffer " + errState ? "1" : "0");    
     return !errState
       ? buffer.resume()
       : buffer.destroy();