<commit_before>"use strict";

var inherits = require('util').inherits,
  EventEmitter = require('events').EventEmitter,
  Connection = require('./connection'),
  MongoError = require('../error'),
  Logger = require('./logger'),
  f = require('util').format,
  CommandResult = require('./command_result');

var MongoCR = require('../auth/mongocr')
  , X509 = require('../auth/x509')
  , Plain = require('../auth/plain')
  , GSSAPI = require('../auth/gssapi')
  , SSPI = require('../auth/sspi')
  , ScramSHA1 = require('../auth/scram');

var DISCONNECTED = 'disconnected';
var CONNECTING = 'connecting';
var CONNECTED = 'connected';
var DESTROYED = 'destroyed';

var Pool = function(options) {
  var self = this;
  // Add event listener
  EventEmitter.call(this);
  // Add the options
  this.options = Object.assign({
    // Host and port settings
    host: 'localhost',
    port: 27017,
    // Pool default max size
    size: 5,
    // socket settings
    connectionTimeout: 30000,
    socketTimeout: 30000,
    keepAlive: true,
    keepAliveInitialDelay: 0,
    noDelay: true,
    // SSL Settings
    ssl: false, checkServerIdentity: false,
    ca: null, cert: null, key: null, passPhrase: null,
    rejectUnauthorized: false,
    promoteLongs: true,
    // Reconnection options
    reconnect: true,
    reconnectInterval: 1000,
    reconnectTries: 30
  }, options);

  // Current reconnect retries
  this.retriesLeft = this.options.reconnectTries;
  this.reconnectId = null;
  // No bson parser passed in
  if(!options.bson || (options.bson
    && (typeof options.bson.serialize != 'function' || typeof options.bson.deserialize != 'function'))) throw new Error("must pass in valid bson parser");
  // Logger instance
  this.logger = Logger('Pool', options);
  // Pool state
  this.state = DISCONNECTED;
  // Connections
  this.availableConnections = [];
  this.inUseConnections = [];
  this.connectingConnections = [];
  // Currently executing
  this.executing = false;
  // Operation work queue
  this.queue = [];
  // All the authProviders
  this.authProviders = {
      'mongocr': new MongoCR(options.bson), 'x509': new X509(options.bson)
    , 'plain': new Plain(options.bson), 'gssapi': new GSSAPI(options.bson)
    , 'sspi': new SSPI(options.bson), 'scram-sha-1': new ScramSHA1(options.bson)
  }
  // Are we currently authenticating
  this.authenticating = false;
  this.nonAuthenticatedConnections = [];
  this.authenticatingTimestamp = null;
}

inherits(Pool, EventEmitter);

function authenticate(pool, auth, connection, cb) {
  if(auth[0] === undefined) return cb(null);
  // We need to authenticate the server
  var mechanism = auth[0];
  var db = auth[1];
  // Validate if the mechanism exists
  if(!pool.authProviders[mechanism]) {
    throw new MongoError(f('authMechanism %s not supported', mechanism));
  }

  // Get the provider
  var provider = pool.authProviders[mechanism];

  // The write function used by the authentication mechanism (bypasses external)
  function write(connection, buffer, callback) {
    // Set the connection workItem callback
    connection.workItem = {cb: callback};
    // Write the buffer out to the connection
    connection.write(buffer);
  };

  // Authenticate using the provided mechanism
  provider.auth.apply(provider, [write, [connection], db].concat(auth.slice(2)).concat([cb]));
}

function reauthenticate(pool, connection, cb) {
  // Authenticate
  function authenticateAgainstProvider(pool, connection, providers, cb) {
    // Finished re-authenticating against providers
    if(providers.length == 0) return cb();
    // Get the provider name
    var provider = pool.authProviders[providers.pop()];

    // The write function used by the authentication mechanism (bypasses external)
    function write(connection, buffer, callback) {
      // Set the connection workItem callback
      connection.workItem = {cb: callback};
      // Write the buffer out to the connection
      connection.write(buffer);
    };

    // Auth provider
    provider.reauthenticate(write, [connection], function(err, r) {
      // We got an error return immediately
      if(err) return cb(err);
      // Continue authenticating the connection
      authenticateAgainstProvider(pool, connection, providers, cb);
    });
  }

  // Start re-authenticating process
  authenticateAgainstProvider(pool, connection, Object.keys(pool.authProviders), cb);
}

function connectionFailureHandler(self, event) {
  return function(err) {
    // console.log("============== this")
    // console.dir(this.workItem)
    removeConnection(self, this);

    // Flush out the callback if there is one
    if(this.workItem && this.workItem.cb) {
      this.workItem.cb(err);
    }

    // No more socket available propegate the event
    if(self.socketCount() == 0 && !self.options.reconnect) {
      self.state = DISCONNECTED;
      // Do not emit error events, they are always close events
      // do not trigger the low level error handler in node
      event = event == 'error' ? 'close' : event;
      self.emit(event, err);
    }

    // Start reconnection attempts
    if(self.socketCount() == 0 && self.options.reconnect && !self.reconnectId) {
      self.state = DISCONNECTED;
      self.reconnectId = setTimeout(attempReconnect(self), self.options.reconnectInterval);
    }
  };
}

function attempReconnect(self) {
  return function() {
    if(self.state == DESTROYED) return;
    // If we have failure schedule a retry
    function _connectionFailureHandler(self, event) {
      return function() {
        self.retriesLeft = self.retriesLeft - 1;
        // How many retries are left
        if(self.retriesLeft == 0) {
          // Destroy the instance
          self.destroy();
          // Emit close event
          self.emit('close', self);
        } else {
          self.reconnectId = setTimeout(attempReconnect(self), self.options.reconnectInterval);
        }
      }
    }

    // Got a connect handler
    function _connectHandler(self) {
      return function() {
        // Assign
        var connection = this;

        // Pool destroyed stop the connection
        if(self.state == DESTROYED) {
          return connection.destroy();
        }

        // Clear out all handlers
        handlers.forEach(function(event) {
          connection.removeAllListeners(event);
        });

        // Reset reconnect id
        self.reconnectId = null;

        // Apply pool connection handlers
        connection.once('error', connectionFailureHandler(self, 'error'));
        connection.once('close', connectionFailureHandler(self, 'close'));
        connection.once('timeout', connectionFailureHandler(self, 'timeout'));
        connection.once('parseError', connectionFailureHandler(self, 'parseError'));

        // Apply any auth to the connection
        reauthenticate(self, this, function(err) {
          // Reset retries
          self.retriesLeft = self.options.reconnectTries;
          // Push to available connections
          self.availableConnections.push(connection);
          // Emit reconnect event<commit_msg>trace issues on travis<commit_after>"use strict";

var inherits = require('util').inherits,
  EventEmitter = require('events').EventEmitter,
  Connection = require('./connection'),
  MongoError = require('../error'),
  Logger = require('./logger'),
  f = require('util').format,
  CommandResult = require('./command_result');

var MongoCR = require('../auth/mongocr')
  , X509 = require('../auth/x509')
  , Plain = require('../auth/plain')
  , GSSAPI = require('../auth/gssapi')
  , SSPI = require('../auth/sspi')
  , ScramSHA1 = require('../auth/scram');

var DISCONNECTED = 'disconnected';
var CONNECTING = 'connecting';
var CONNECTED = 'connected';
var DESTROYED = 'destroyed';

var Pool = function(options) {
  var self = this;
  // Add event listener
  EventEmitter.call(this);
  // Add the options
  this.options = Object.assign({
    // Host and port settings
    host: 'localhost',
    port: 27017,
    // Pool default max size
    size: 5,
    // socket settings
    connectionTimeout: 30000,
    socketTimeout: 30000,
    keepAlive: true,
    keepAliveInitialDelay: 0,
    noDelay: true,
    // SSL Settings
    ssl: false, checkServerIdentity: false,
    ca: null, cert: null, key: null, passPhrase: null,
    rejectUnauthorized: false,
    promoteLongs: true,
    // Reconnection options
    reconnect: true,
    reconnectInterval: 1000,
    reconnectTries: 30
  }, options);

  // Current reconnect retries
  this.retriesLeft = this.options.reconnectTries;
  this.reconnectId = null;
  // No bson parser passed in
  if(!options.bson || (options.bson
    && (typeof options.bson.serialize != 'function' || typeof options.bson.deserialize != 'function'))) throw new Error("must pass in valid bson parser");
  // Logger instance
  this.logger = Logger('Pool', options);
  // Pool state
  this.state = DISCONNECTED;
  // Connections
  this.availableConnections = [];
  this.inUseConnections = [];
  this.connectingConnections = [];
  // Currently executing
  this.executing = false;
  // Operation work queue
  this.queue = [];
  // All the authProviders
  this.authProviders = {
      'mongocr': new MongoCR(options.bson), 'x509': new X509(options.bson)
    , 'plain': new Plain(options.bson), 'gssapi': new GSSAPI(options.bson)
    , 'sspi': new SSPI(options.bson), 'scram-sha-1': new ScramSHA1(options.bson)
  }
  // Are we currently authenticating
  this.authenticating = false;
  this.nonAuthenticatedConnections = [];
  this.authenticatingTimestamp = null;
}

inherits(Pool, EventEmitter);

function authenticate(pool, auth, connection, cb) {
  if(auth[0] === undefined) return cb(null);
  // We need to authenticate the server
  var mechanism = auth[0];
  var db = auth[1];
  // Validate if the mechanism exists
  if(!pool.authProviders[mechanism]) {
    throw new MongoError(f('authMechanism %s not supported', mechanism));
  }

  // Get the provider
  var provider = pool.authProviders[mechanism];

  // The write function used by the authentication mechanism (bypasses external)
  function write(connection, buffer, callback) {
    // Set the connection workItem callback
    connection.workItem = {cb: callback};
    // Write the buffer out to the connection
    connection.write(buffer);
  };

  // Authenticate using the provided mechanism
  provider.auth.apply(provider, [write, [connection], db].concat(auth.slice(2)).concat([cb]));
}

function reauthenticate(pool, connection, cb) {
  // Authenticate
  function authenticateAgainstProvider(pool, connection, providers, cb) {
    // Finished re-authenticating against providers
    if(providers.length == 0) return cb();
    // Get the provider name
    var provider = pool.authProviders[providers.pop()];

    // The write function used by the authentication mechanism (bypasses external)
    function write(connection, buffer, callback) {
      // Set the connection workItem callback
      connection.workItem = {cb: callback};
      // Write the buffer out to the connection
      connection.write(buffer);
    };

    // Auth provider
    provider.reauthenticate(write, [connection], function(err, r) {
      // We got an error return immediately
      if(err) return cb(err);
      // Continue authenticating the connection
      authenticateAgainstProvider(pool, connection, providers, cb);
    });
  }

  // Start re-authenticating process
  authenticateAgainstProvider(pool, connection, Object.keys(pool.authProviders), cb);
}

function connectionFailureHandler(self, event) {
  return function(err) {
    // console.log("============== this")
    // console.dir(this.workItem)
    removeConnection(self, this);

    // Flush out the callback if there is one
    if(this.workItem && this.workItem.cb) {
      this.workItem.cb(err);
    }

    // No more socket available propegate the event
    if(self.socketCount() == 0 && !self.options.reconnect) {
      self.state = DISCONNECTED;
      // Do not emit error events, they are always close events
      // do not trigger the low level error handler in node
      event = event == 'error' ? 'close' : event;
      self.emit(event, err);
    }

    // Start reconnection attempts
    if(self.socketCount() == 0 && self.options.reconnect && !self.reconnectId) {
      self.state = DISCONNECTED;
      self.reconnectId = setTimeout(attemptReconnect(self), self.options.reconnectInterval);
    }
  };
}

function attemptReconnect(self) {
  return function() {
    console.log("==== attemptReconnect :: " + self.state)
    if(self.state == DESTROYED) return;
    // If we have failure schedule a retry
    function _connectionFailureHandler(self, event) {
      return function() {
        self.retriesLeft = self.retriesLeft - 1;
        // How many retries are left
        if(self.retriesLeft == 0) {
          console.log("==== attemptReconnect :: destroy")
          // Destroy the instance
          self.destroy();
          // Emit close event
          self.emit('close', self);
        } else {
          self.reconnectId = setTimeout(attemptReconnect(self), self.options.reconnectInterval);
        }
      }
    }

    // Got a connect handler
    function _connectHandler(self) {
      return function() {
        console.log("==== attemptReconnect :: connect")
        // Assign
        var connection = this;

        // Pool destroyed stop the connection
        if(self.state == DESTROYED) {
          return connection.destroy();
        }

        // Clear out all handlers
        handlers.forEach(function(event) {
          connection.removeAllListeners(event);
        });

        // Reset reconnect id
        self.reconnectId = null;

        // Apply pool connection handlers
        connection.once('error', connectionFailureHandler(self, 'error'));
        connection.once('close', connectionFailureHandler(self, 'close'));
        connection.once('timeout', connectionFailureHandler(self, 'timeout'));
        connection.once('parseError', connectionFailureHandler(self, 'parseError'));

        // Apply any auth to the connection
        reauthenticate(self, this, function(err) {
          // Reset retries
          self.retriesLeft = self.options.reconnectTries;
          // Push to available connections
          self.availableConnections.push(connection);
          // Emit reconnect event
{'fraction_non_alphanumeric': 0.12526168876482904, 'fraction_numerical': 0.006559665038381019, 'mean_word_length': 3.046019198193111, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5250982', 'n_tokens_mistral': 4195, 'n_tokens_neox': 4056, 'n_words': 1089}