AK-21/Graphite-Industrial-Intelligence

1'use strict'2 3var net = require('net')4  , tls = require('tls')5  , http = require('http')6  , https = require('https')7  , events = require('events')8  , assert = require('assert')9  , util = require('util')10  , Buffer = require('safe-buffer').Buffer11  ;12 13exports.httpOverHttp = httpOverHttp14exports.httpsOverHttp = httpsOverHttp15exports.httpOverHttps = httpOverHttps16exports.httpsOverHttps = httpsOverHttps17 18 19function httpOverHttp(options) {20  var agent = new TunnelingAgent(options)21  agent.request = http.request22  return agent23}24 25function httpsOverHttp(options) {26  var agent = new TunnelingAgent(options)27  agent.request = http.request28  agent.createSocket = createSecureSocket29  agent.defaultPort = 44330  return agent31}32 33function httpOverHttps(options) {34  var agent = new TunnelingAgent(options)35  agent.request = https.request36  return agent37}38 39function httpsOverHttps(options) {40  var agent = new TunnelingAgent(options)41  agent.request = https.request42  agent.createSocket = createSecureSocket43  agent.defaultPort = 44344  return agent45}46 47 48function TunnelingAgent(options) {49  var self = this50  self.options = options || {}51  self.proxyOptions = self.options.proxy || {}52  self.maxSockets = self.options.maxSockets || http.Agent.defaultMaxSockets53  self.requests = []54  self.sockets = []55 56  self.on('free', function onFree(socket, host, port) {57    for (var i = 0, len = self.requests.length; i < len; ++i) {58      var pending = self.requests[i]59      if (pending.host === host && pending.port === port) {60        // Detect the request to connect same origin server,61        // reuse the connection.62        self.requests.splice(i, 1)63        pending.request.onSocket(socket)64        return65      }66    }67    socket.destroy()68    self.removeSocket(socket)69  })70}71util.inherits(TunnelingAgent, events.EventEmitter)72 73TunnelingAgent.prototype.addRequest = function addRequest(req, options) {74  var self = this75 76   // Legacy API: addRequest(req, host, port, path)77  if (typeof options === 'string') {78    options = {79      host: options,80      port: arguments[2],81      path: arguments[3]82    };83  }84 85  if (self.sockets.length >= this.maxSockets) {86    // We are over limit so we'll add it to the queue.87    self.requests.push({host: options.host, port: options.port, request: req})88    return89  }90 91  // If we are under maxSockets create a new one.92  self.createConnection({host: options.host, port: options.port, request: req})93}94 95TunnelingAgent.prototype.createConnection = function createConnection(pending) {96  var self = this97 98  self.createSocket(pending, function(socket) {99    socket.on('free', onFree)100    socket.on('close', onCloseOrRemove)101    socket.on('agentRemove', onCloseOrRemove)102    pending.request.onSocket(socket)103 104    function onFree() {105      self.emit('free', socket, pending.host, pending.port)106    }107 108    function onCloseOrRemove(err) {109      self.removeSocket(socket)110      socket.removeListener('free', onFree)111      socket.removeListener('close', onCloseOrRemove)112      socket.removeListener('agentRemove', onCloseOrRemove)113    }114  })115}116 117TunnelingAgent.prototype.createSocket = function createSocket(options, cb) {118  var self = this119  var placeholder = {}120  self.sockets.push(placeholder)121 122  var connectOptions = mergeOptions({}, self.proxyOptions,123    { method: 'CONNECT'124    , path: options.host + ':' + options.port125    , agent: false126    }127  )128  if (connectOptions.proxyAuth) {129    connectOptions.headers = connectOptions.headers || {}130    connectOptions.headers['Proxy-Authorization'] = 'Basic ' +131        Buffer.from(connectOptions.proxyAuth).toString('base64')132  }133 134  debug('making CONNECT request')135  var connectReq = self.request(connectOptions)136  connectReq.useChunkedEncodingByDefault = false // for v0.6137  connectReq.once('response', onResponse) // for v0.6138  connectReq.once('upgrade', onUpgrade)   // for v0.6139  connectReq.once('connect', onConnect)   // for v0.7 or later140  connectReq.once('error', onError)141  connectReq.end()142 143  function onResponse(res) {144    // Very hacky. This is necessary to avoid http-parser leaks.145    res.upgrade = true146  }147 148  function onUpgrade(res, socket, head) {149    // Hacky.150    process.nextTick(function() {151      onConnect(res, socket, head)152    })153  }154 155  function onConnect(res, socket, head) {156    connectReq.removeAllListeners()157    socket.removeAllListeners()158 159    if (res.statusCode === 200) {160      assert.equal(head.length, 0)161      debug('tunneling connection has established')162      self.sockets[self.sockets.indexOf(placeholder)] = socket163      cb(socket)164    } else {165      debug('tunneling socket could not be established, statusCode=%d', res.statusCode)166      var error = new Error('tunneling socket could not be established, ' + 'statusCode=' + res.statusCode)167      error.code = 'ECONNRESET'168      options.request.emit('error', error)169      self.removeSocket(placeholder)170    }171  }172 173  function onError(cause) {174    connectReq.removeAllListeners()175 176    debug('tunneling socket could not be established, cause=%s\n', cause.message, cause.stack)177    var error = new Error('tunneling socket could not be established, ' + 'cause=' + cause.message)178    error.code = 'ECONNRESET'179    options.request.emit('error', error)180    self.removeSocket(placeholder)181  }182}183 184TunnelingAgent.prototype.removeSocket = function removeSocket(socket) {185  var pos = this.sockets.indexOf(socket)186  if (pos === -1) return187 188  this.sockets.splice(pos, 1)189 190  var pending = this.requests.shift()191  if (pending) {192    // If we have pending requests and a socket gets closed a new one193    // needs to be created to take over in the pool for the one that closed.194    this.createConnection(pending)195  }196}197 198function createSecureSocket(options, cb) {199  var self = this200  TunnelingAgent.prototype.createSocket.call(self, options, function(socket) {201    // 0 is dummy port for v0.6202    var secureSocket = tls.connect(0, mergeOptions({}, self.options,203      { servername: options.host204      , socket: socket205      }206    ))207    self.sockets[self.sockets.indexOf(socket)] = secureSocket208    cb(secureSocket)209  })210}211 212 213function mergeOptions(target) {214  for (var i = 1, len = arguments.length; i < len; ++i) {215    var overrides = arguments[i]216    if (typeof overrides === 'object') {217      var keys = Object.keys(overrides)218      for (var j = 0, keyLen = keys.length; j < keyLen; ++j) {219        var k = keys[j]220        if (overrides[k] !== undefined) {221          target[k] = overrides[k]222        }223      }224    }225  }226  return target227}228 229 230var debug231if (process.env.NODE_DEBUG && /\btunnel\b/.test(process.env.NODE_DEBUG)) {232  debug = function() {233    var args = Array.prototype.slice.call(arguments)234    if (typeof args[0] === 'string') {235      args[0] = 'TUNNEL: ' + args[0]236    } else {237      args.unshift('TUNNEL:')238    }239    console.error.apply(console, args)240  }241} else {242  debug = function() {}243}244exports.debug = debug // for test245