Retry Agent

forever.js

@@ -0,0 +1,84 @@
+module.exports = ForeverAgent
+
+var util = require('util')
+  , Agent = require('http').Agent
+  , net = require('net')
+
+function ForeverAgent(options) {
+  var self = this
+  self.options = options || {}
+  self.requests = {}
+  self.sockets = {}
+  self.freeSockets = {}
+  self.maxSockets = self.options.maxSockets || Agent.defaultMaxSockets
+  self.minSockets = self.options.minSockets || ForeverAgent.defaultMinSockets
+  self.on('free', function(socket, host, port) {
+    var name = host + ':' + port
+    if (self.requests[name] && self.requests[name].length) {
+      self.requests[name].shift().onSocket(socket)
+    } else if (self.sockets[name].length < self.minSockets) {
+      if (!self.freeSockets[name]) self.freeSockets[name] = []
+      self.freeSockets[name].push(socket)
+
+      // if an error happens while we don't use the socket anyway, meh, throw the socket away
+      function onIdleError() {
+        socket.destroy()
+      }
+      socket._onIdleError = onIdleError
+      socket.on('error', onIdleError)
+    } else {
+      // If there are no pending requests just destroy the
+      // socket and it will get removed from the pool. This
+      // gets us out of timeout issues and allows us to
+      // default to Connection:keep-alive.
+      socket.destroy();
+    }
+  })
+  self.createConnection = net.createConnection
+}
+util.inherits(ForeverAgent, Agent)
+
+ForeverAgent.defaultMinSockets = 5
+
+ForeverAgent.prototype.addRequestNoreuse = Agent.prototype.addRequest
+ForeverAgent.prototype.addRequest = function(req, host, port) {
+  var name = host + ':' + port
+  if (this.freeSockets[name] && this.freeSockets[name].length > 0 && !req.useChunkedEncodingByDefault) {
+    var idleSocket = this.freeSockets[name].pop()
+    idleSocket.removeListener('error', idleSocket._onIdleError)
+    delete idleSocket._onIdleError
+    req._reusedSocket = true
+    req.onSocket(idleSocket)
+  } else {
+    this.addRequestNoreuse(req, host, port)
+  }
+}
+
+ForeverAgent.prototype.removeSocket = function(s, name, host, port) {
+  if (this.sockets[name]) {
+    var index = this.sockets[name].indexOf(s);
+    if (index !== -1) {
+      this.sockets[name].splice(index, 1);
+    }
+  } else if (this.sockets[name] && this.sockets[name].length === 0) {
+    // don't leak
+    delete this.sockets[name];
+    delete this.requests[name];
+  }
+
+  if (this.freeSockets[name]) {
+    var index = this.freeSockets[name].indexOf(s)
+    if (index !== -1) {
+      this.freeSockets[name].splice(index, 1)
+      if (this.freeSockets[name].length === 0) {
+        delete this.freeSockets[name]
+      }
+    }
+  }
+
+  if (this.requests[name] && this.requests[name].length) {
+    // If we have pending requests and a socket gets closed a new one
+    // needs to be created to take over in the pool for the one that closed.
+    this.createSocket(name, host, port).emit('free');
+  }
+}

main.js

@@ -22,6 +22,7 @@ var http = require('http')
   , mimetypes = require('./mimetypes')
   , oauth = require('./oauth')
   , uuid = require('./uuid')
+  , ForeverAgent = require('./forever')
   , Cookie = require('./vendor/cookie')
   , CookieJar = require('./vendor/cookie/jar')
   , cookieJar = new CookieJar
@@ -184,6 +185,12 @@ Request.prototype.request = function () {
 
   var clientErrorHandler = function (error) {
     if (setHost) delete self.headers.host
+    if (self.req._reusedSocket && error.code === 'ECONNRESET') {
+      self.agent = {addRequest: ForeverAgent.prototype.addRequestNoreuse.bind(self.agent)}
+      self.start()
+      self.req.end()
+      return
+    }
     if (self.timeout && self.timeoutTimer) clearTimeout(self.timeoutTimer)
     self.emit('error', error)
   }
@@ -317,12 +324,12 @@ Request.prototype.request = function () {
   } else {
     if (self.maxSockets) {
       // Don't use our pooling if node has the refactored client
-      self.agent = self.httpModule.globalAgent || self.getAgent(self.host, self.port)
+      self.agent = self.agent || self.httpModule.globalAgent || self.getAgent(self.host, self.port)
       self.agent.maxSockets = self.maxSockets
     }
     if (self.pool.maxSockets) {
       // Don't use our pooling if node has the refactored client
-      self.agent = self.httpModule.globalAgent || self.getAgent(self.host, self.port)
+      self.agent = self.agent || self.httpModule.globalAgent || self.getAgent(self.host, self.port)
       self.agent.maxSockets = self.pool.maxSockets
     }
   }
@@ -471,7 +478,7 @@ Request.prototype.request = function () {
       }
     })
 
-    if (self.timeout) {
+    if (self.timeout && !self.timeoutTimer) {
       self.timeoutTimer = setTimeout(function() {
         self.req.abort()
         var e = new Error("ETIMEDOUT")
@@ -595,6 +602,17 @@ request.defaults = function (options) {
   return de
 }
 
+request.forever = function (agentOptions, optionsArg) {
+  var options = {}
+  if (agentOptions) {
+    for (option in optionsArg) {
+      options[option] = optionsArg[option]
+    }
+  }
+  options.agent = new ForeverAgent(agentOptions)
+  return request.defaults(options)
+}
+
 request.get = request
 request.post = function (options, callback) {
   if (typeof options === 'string') options = {uri:options}