agent.js

'use strict';

const OriginalAgent = require('http').Agent;
const ms = require('humanize-ms');
const debug = require('debug')('agentkeepalive');
const deprecate = require('depd')('agentkeepalive');
const {
  INIT_SOCKET,
  CURRENT_ID,
  CREATE_ID,
  SOCKET_CREATED_TIME,
  SOCKET_NAME,
  SOCKET_REQUEST_COUNT,
  SOCKET_REQUEST_FINISHED_COUNT,
} = require('./constants');

// OriginalAgent come from
// - https://github.com/nodejs/node/blob/v8.12.0/lib/_http_agent.js
// - https://github.com/nodejs/node/blob/v10.12.0/lib/_http_agent.js

class Agent extends OriginalAgent {
  constructor(options) {
    options = options || {};
    options.keepAlive = options.keepAlive !== false;
    // default is keep-alive and 15s free socket timeout
    if (options.freeSocketTimeout === undefined) {
      options.freeSocketTimeout = 15000;
    }
    // Legacy API: keepAliveTimeout should be rename to `freeSocketTimeout`
    if (options.keepAliveTimeout) {
      deprecate('options.keepAliveTimeout is deprecated, please use options.freeSocketTimeout instead');
      options.freeSocketTimeout = options.keepAliveTimeout;
      delete options.keepAliveTimeout;
    }
    // Legacy API: freeSocketKeepAliveTimeout should be rename to `freeSocketTimeout`
    if (options.freeSocketKeepAliveTimeout) {
      deprecate('options.freeSocketKeepAliveTimeout is deprecated, please use options.freeSocketTimeout instead');
      options.freeSocketTimeout = options.freeSocketKeepAliveTimeout;
      delete options.freeSocketKeepAliveTimeout;
    }

    // Sets the socket to timeout after timeout milliseconds of inactivity on the socket.
    // By default is double free socket timeout.
    if (options.timeout === undefined) {
      // make sure socket default inactivity timeout >= 30s
      options.timeout = Math.max(options.freeSocketTimeout * 2, 30000);
    }

    // support humanize format
    options.timeout = ms(options.timeout);
    options.freeSocketTimeout = ms(options.freeSocketTimeout);
    options.socketActiveTTL = options.socketActiveTTL ? ms(options.socketActiveTTL) : 0;

    super(options);

    this[CURRENT_ID] = 0;

    // create socket success counter
    this.createSocketCount = 0;
    this.createSocketCountLastCheck = 0;

    this.createSocketErrorCount = 0;
    this.createSocketErrorCountLastCheck = 0;

    this.closeSocketCount = 0;
    this.closeSocketCountLastCheck = 0;

    // socket error event count
    this.errorSocketCount = 0;
    this.errorSocketCountLastCheck = 0;

    // request finished counter
    this.requestCount = 0;
    this.requestCountLastCheck = 0;

    // including free socket timeout counter
    this.timeoutSocketCount = 0;
    this.timeoutSocketCountLastCheck = 0;
  }

  get freeSocketKeepAliveTimeout() {
    deprecate('agent.freeSocketKeepAliveTimeout is deprecated, please use agent.options.freeSocketTimeout instead');
    return this.options.freeSocketTimeout;
  }

  get timeout() {
    deprecate('agent.timeout is deprecated, please use agent.options.timeout instead');
    return this.options.timeout;
  }

  get socketActiveTTL() {
    deprecate('agent.socketActiveTTL is deprecated, please use agent.options.socketActiveTTL instead');
    return this.options.socketActiveTTL;
  }

  keepSocketAlive(socket) {
    const result = super.keepSocketAlive(socket);
    // should not keepAlive, do nothing
    if (!result) return result;

    let freeSocketTimeout = this.options.freeSocketTimeout;
    const socketActiveTTL = this.options.socketActiveTTL;
    if (socketActiveTTL) {
      // check socketActiveTTL
      const aliveTime = Date.now() - socket[SOCKET_CREATED_TIME];
      const diff = socketActiveTTL - aliveTime;
      // destroy it
      if (diff <= 0) {
        debug('%s(requests: %s, finished: %s) free but need to destroy by TTL, alive %sms(max %sms)',
          socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT],
          aliveTime, socketActiveTTL);
        return false;
      }

      if (freeSocketTimeout && diff < freeSocketTimeout) {
        debug('%s(requests: %s, finished: %s) free and wait for %sms TTL timeout',
          socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT],
          diff);
        freeSocketTimeout = diff;
      }
    }
    // set freeSocketTimeout
    if (freeSocketTimeout) {
      // set free keepalive timer
      // try to use socket custom freeSocketTimeout first, support headers['keep-alive']
      // https://github.com/node-modules/urllib/blob/b76053020923f4d99a1c93cf2e16e0c5ba10bacf/lib/urllib.js#L498
      const customFreeSocketTimeout = socket.freeSocketTimeout || socket.freeSocketKeepAliveTimeout;
      if (customFreeSocketTimeout && customFreeSocketTimeout < freeSocketTimeout) {
        freeSocketTimeout = customFreeSocketTimeout;
      }
      // FIXME: need to make setRequestSocket as a method on Agent class
      // then we can reset the agent.options.timeout when free socket is reused.
      socket.setTimeout(freeSocketTimeout);
    }
    return true;
  }

  // only call on addRequest
  reuseSocket(...args) {
    // reuseSocket(socket, req)
    super.reuseSocket(...args);
    const socket = args[0];
    const req = args[1];
    req.reusedSocket = true;
    const agentTimeout = this.options.timeout;
    if (getSocketTimeout(socket) !== agentTimeout) {
      // reset timeout before use
      socket.setTimeout(agentTimeout);
      debug('%s reset timeout to %sms', socket[SOCKET_NAME], agentTimeout);
    }
    socket[SOCKET_REQUEST_COUNT]++;
    debug('%s(requests: %s, finished: %s) reuse on addRequest, timeout %sms',
      socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT],
      getSocketTimeout(socket));
  }

  [CREATE_ID]() {
    const id = this[CURRENT_ID]++;
    if (this[CURRENT_ID] === Number.MAX_SAFE_INTEGER) this[CURRENT_ID] = 0;
    return id;
  }

  [INIT_SOCKET](socket, options) {
    // bugfix here.
    // https on node 8, 10 won't set agent.options.timeout by default
    // TODO: need to fix on node itself
    if (options.timeout) {
      const timeout = getSocketTimeout(socket);
      if (!timeout) {
        socket.setTimeout(options.timeout);
      }
    }

    if (this.options.keepAlive) {
      // Disable Nagle's algorithm: http://blog.caustik.com/2012/04/08/scaling-node-js-to-100k-concurrent-connections/
      // https://fengmk2.com/benchmark/nagle-algorithm-delayed-ack-mock.html
      socket.setNoDelay(true);
    }
    this.createSocketCount++;
    if (this.options.socketActiveTTL) {
      socket[SOCKET_CREATED_TIME] = Date.now();
    }
    // don't show the hole '-----BEGIN CERTIFICATE----' key string
    socket[SOCKET_NAME] = `sock[${this[CREATE_ID]()}#${options._agentKey}]`.split('-----BEGIN', 1)[0];
    socket[SOCKET_REQUEST_COUNT] = 1;
    socket[SOCKET_REQUEST_FINISHED_COUNT] = 0;
    installListeners(this, socket, options);
  }

  createConnection(options, oncreate) {
    let called = false;
    const onNewCreate = (err, socket) => {
      if (called) return;
      called = true;

      if (err) {
        this.createSocketErrorCount++;
        return oncreate(err);
      }
      this[INIT_SOCKET](socket, options);
      oncreate(err, socket);
    };

    const newSocket = super.createConnection(options, onNewCreate);
    if (newSocket) onNewCreate(null, newSocket);
  }

  get statusChanged() {
    const changed = this.createSocketCount !== this.createSocketCountLastCheck ||
      this.createSocketErrorCount !== this.createSocketErrorCountLastCheck ||
      this.closeSocketCount !== this.closeSocketCountLastCheck ||
      this.errorSocketCount !== this.errorSocketCountLastCheck ||
      this.timeoutSocketCount !== this.timeoutSocketCountLastCheck ||
      this.requestCount !== this.requestCountLastCheck;
    if (changed) {
      this.createSocketCountLastCheck = this.createSocketCount;
      this.createSocketErrorCountLastCheck = this.createSocketErrorCount;
      this.closeSocketCountLastCheck = this.closeSocketCount;
      this.errorSocketCountLastCheck = this.errorSocketCount;
      this.timeoutSocketCountLastCheck = this.timeoutSocketCount;
      this.requestCountLastCheck = this.requestCount;
    }
    return changed;
  }

  getCurrentStatus() {
    return {
      createSocketCount: this.createSocketCount,
      createSocketErrorCount: this.createSocketErrorCount,
      closeSocketCount: this.closeSocketCount,
      errorSocketCount: this.errorSocketCount,
      timeoutSocketCount: this.timeoutSocketCount,
      requestCount: this.requestCount,
      freeSockets: inspect(this.freeSockets),
      sockets: inspect(this.sockets),
      requests: inspect(this.requests),
    };
  }
}

// node 8 don't has timeout attribute on socket
// https://github.com/nodejs/node/pull/21204/files#diff-e6ef024c3775d787c38487a6309e491dR408
function getSocketTimeout(socket) {
  return socket.timeout || socket._idleTimeout;
}

function installListeners(agent, socket, options) {
  debug('%s create, timeout %sms', socket[SOCKET_NAME], getSocketTimeout(socket));

  // listener socket events: close, timeout, error, free
  function onFree() {
    // create and socket.emit('free') logic
    // https://github.com/nodejs/node/blob/master/lib/_http_agent.js#L311
    // no req on the socket, it should be the new socket
    if (!socket._httpMessage && socket[SOCKET_REQUEST_COUNT] === 1) return;

    socket[SOCKET_REQUEST_FINISHED_COUNT]++;
    agent.requestCount++;
    debug('%s(requests: %s, finished: %s) free',
      socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT]);

    // should reuse on pedding requests?
    const name = agent.getName(options);
    if (socket.writable && agent.requests[name] && agent.requests[name].length) {
      // will be reuse on agent free listener
      socket[SOCKET_REQUEST_COUNT]++;
      debug('%s(requests: %s, finished: %s) will be reuse on agent free event',
        socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT]);
    }
  }
  socket.on('free', onFree);

  function onClose(isError) {
    debug('%s(requests: %s, finished: %s) close, isError: %s',
      socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT], isError);
    agent.closeSocketCount++;
  }
  socket.on('close', onClose);

  // start socket timeout handler
  function onTimeout() {
    const listenerCount = socket.listeners('timeout').length;
    debug('%s(requests: %s, finished: %s) timeout after %sms, listeners %s',
      socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT],
      getSocketTimeout(socket), listenerCount);
    agent.timeoutSocketCount++;
    const name = agent.getName(options);
    if (agent.freeSockets[name] && agent.freeSockets[name].indexOf(socket) !== -1) {
      // free socket timeout, destroy quietly
      socket.destroy();
      // Remove it from freeSockets list immediately to prevent new requests
      // from being sent through this socket.
      agent.removeSocket(socket, options);
      debug('%s is free, destroy quietly', socket[SOCKET_NAME]);
    } else {
      // if there is no any request socket timeout handler,
      // agent need to handle socket timeout itself.
      //
      // custom request socket timeout handle logic must follow these rules:
      //  1. Destroy socket first
      //  2. Must emit socket 'agentRemove' event tell agent remove socket
      //     from freeSockets list immediately.
      //     Otherise you may be get 'socket hang up' error when reuse
      //     free socket and timeout happen in the same time.
      if (listenerCount === 1) {
        const error = new Error('Socket timeout');
        error.code = 'ERR_SOCKET_TIMEOUT';
        error.timeout = getSocketTimeout(socket);
        // must manually call socket.end() or socket.destroy() to end the connection.
        // https://nodejs.org/dist/latest-v10.x/docs/api/net.html#net_socket_settimeout_timeout_callback
        socket.destroy(error);
        agent.removeSocket(socket, options);
        debug('%s destroy with timeout error', socket[SOCKET_NAME]);
      }
    }
  }
  socket.on('timeout', onTimeout);

  function onError(err) {
    const listenerCount = socket.listeners('error').length;
    debug('%s(requests: %s, finished: %s) error: %s, listenerCount: %s',
      socket[SOCKET_NAME], socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT],
      err, listenerCount);
    agent.errorSocketCount++;
    if (listenerCount === 1) {
      // if socket don't contain error event handler, don't catch it, emit it again
      debug('%s emit uncaught error event', socket[SOCKET_NAME]);
      socket.removeListener('error', onError);
      socket.emit('error', err);
    }
  }
  socket.on('error', onError);

  function onRemove() {
    debug('%s(requests: %s, finished: %s) agentRemove',
      socket[SOCKET_NAME],
      socket[SOCKET_REQUEST_COUNT], socket[SOCKET_REQUEST_FINISHED_COUNT]);
    // We need this function for cases like HTTP 'upgrade'
    // (defined by WebSockets) where we need to remove a socket from the
    // pool because it'll be locked up indefinitely
    socket.removeListener('close', onClose);
    socket.removeListener('error', onError);
    socket.removeListener('free', onFree);
    socket.removeListener('timeout', onTimeout);
    socket.removeListener('agentRemove', onRemove);
  }
  socket.on('agentRemove', onRemove);
}

module.exports = Agent;

function inspect(obj) {
  const res = {};
  for (const key in obj) {
    res[key] = obj[key].length;
  }
  return res;
}