aboutsummaryrefslogtreecommitdiff
path: root/node_modules/mongodb-core/lib/connection
diff options
context:
space:
mode:
authornanalelfe <nargiza.nosirova@mail.utoronto.ca>2016-07-27 12:46:04 +0000
committernanalelfe <nargiza.nosirova@mail.utoronto.ca>2016-07-27 12:46:04 +0000
commit7bee9b7e0968b650b3dbe2e35bf3a0d7d6e5e00e (patch)
tree4e63af46a6e306bfeac1ef4b8bd83e1b45a37257 /node_modules/mongodb-core/lib/connection
parent6520aa92a114d65b17b178a952c8985e84afd231 (diff)
Removed some unused modules with prune
Diffstat (limited to 'node_modules/mongodb-core/lib/connection')
-rw-r--r--node_modules/mongodb-core/lib/connection/command_result.js38
-rw-r--r--node_modules/mongodb-core/lib/connection/commands.js539
-rw-r--r--node_modules/mongodb-core/lib/connection/connection.js531
-rw-r--r--node_modules/mongodb-core/lib/connection/logger.js229
-rw-r--r--node_modules/mongodb-core/lib/connection/pool.js1122
-rw-r--r--node_modules/mongodb-core/lib/connection/utils.js67
6 files changed, 0 insertions, 2526 deletions
diff --git a/node_modules/mongodb-core/lib/connection/command_result.js b/node_modules/mongodb-core/lib/connection/command_result.js
deleted file mode 100644
index eb7b27a..0000000
--- a/node_modules/mongodb-core/lib/connection/command_result.js
+++ /dev/null
@@ -1,38 +0,0 @@
-"use strict";
-
-var setProperty = require('../connection/utils').setProperty
- , getProperty = require('../connection/utils').getProperty
- , getSingleProperty = require('../connection/utils').getSingleProperty;
-
-/**
- * Creates a new CommandResult instance
- * @class
- * @param {object} result CommandResult object
- * @param {Connection} connection A connection instance associated with this result
- * @return {CommandResult} A cursor instance
- */
-var CommandResult = function(result, connection, message) {
- this.result = result;
- this.connection = connection;
- this.message = message;
-}
-
-/**
- * Convert CommandResult to JSON
- * @method
- * @return {object}
- */
-CommandResult.prototype.toJSON = function() {
- return this.result;
-}
-
-/**
- * Convert CommandResult to String representation
- * @method
- * @return {string}
- */
-CommandResult.prototype.toString = function() {
- return JSON.stringify(this.toJSON());
-}
-
-module.exports = CommandResult;
diff --git a/node_modules/mongodb-core/lib/connection/commands.js b/node_modules/mongodb-core/lib/connection/commands.js
deleted file mode 100644
index dba0d1e..0000000
--- a/node_modules/mongodb-core/lib/connection/commands.js
+++ /dev/null
@@ -1,539 +0,0 @@
-"use strict";
-
-var f = require('util').format
- , Long = require('bson').Long
- , setProperty = require('./utils').setProperty
- , getProperty = require('./utils').getProperty
- , getSingleProperty = require('./utils').getSingleProperty;
-
-// Incrementing request id
-var _requestId = 0;
-
-// Wire command operation ids
-var OP_QUERY = 2004;
-var OP_GETMORE = 2005;
-var OP_KILL_CURSORS = 2007;
-
-// Query flags
-var OPTS_NONE = 0;
-var OPTS_TAILABLE_CURSOR = 2;
-var OPTS_SLAVE = 4;
-var OPTS_OPLOG_REPLAY = 8;
-var OPTS_NO_CURSOR_TIMEOUT = 16;
-var OPTS_AWAIT_DATA = 32;
-var OPTS_EXHAUST = 64;
-var OPTS_PARTIAL = 128;
-
-// Response flags
-var CURSOR_NOT_FOUND = 0;
-var QUERY_FAILURE = 2;
-var SHARD_CONFIG_STALE = 4;
-var AWAIT_CAPABLE = 8;
-
-/**************************************************************
- * QUERY
- **************************************************************/
-var Query = function(bson, ns, query, options) {
- var self = this;
- // Basic options needed to be passed in
- if(ns == null) throw new Error("ns must be specified for query");
- if(query == null) throw new Error("query must be specified for query");
-
- // Validate that we are not passing 0x00 in the colletion name
- if(!!~ns.indexOf("\x00")) {
- throw new Error("namespace cannot contain a null character");
- }
-
- // Basic options
- this.bson = bson;
- this.ns = ns;
- this.query = query;
-
- // Ensure empty options
- this.options = options || {};
-
- // Additional options
- this.numberToSkip = options.numberToSkip || 0;
- this.numberToReturn = options.numberToReturn || 0;
- this.returnFieldSelector = options.returnFieldSelector || null;
- this.requestId = _requestId++;
-
- // Serialization option
- this.serializeFunctions = typeof options.serializeFunctions == 'boolean' ? options.serializeFunctions : false;
- this.ignoreUndefined = typeof options.ignoreUndefined == 'boolean' ? options.ignoreUndefined : false;
- this.maxBsonSize = options.maxBsonSize || 1024 * 1024 * 16;
- this.checkKeys = typeof options.checkKeys == 'boolean' ? options.checkKeys : true;
- this.batchSize = self.numberToReturn;
-
- // Flags
- this.tailable = false;
- this.slaveOk = typeof options.slaveOk == 'boolean'? options.slaveOk : false;
- this.oplogReplay = false;
- this.noCursorTimeout = false;
- this.awaitData = false;
- this.exhaust = false;
- this.partial = false;
-}
-
-//
-// Assign a new request Id
-Query.prototype.incRequestId = function() {
- this.requestId = _requestId++;
-}
-
-//
-// Assign a new request Id
-Query.nextRequestId = function() {
- return _requestId + 1;
-}
-
-//
-// Uses a single allocated buffer for the process, avoiding multiple memory allocations
-Query.prototype.toBin = function() {
- var self = this;
- var buffers = [];
- var projection = null;
-
- // Set up the flags
- var flags = 0;
- if(this.tailable) {
- flags |= OPTS_TAILABLE_CURSOR;
- }
-
- if(this.slaveOk) {
- flags |= OPTS_SLAVE;
- }
-
- if(this.oplogReplay) {
- flags |= OPTS_OPLOG_REPLAY;
- }
-
- if(this.noCursorTimeout) {
- flags |= OPTS_NO_CURSOR_TIMEOUT;
- }
-
- if(this.awaitData) {
- flags |= OPTS_AWAIT_DATA;
- }
-
- if(this.exhaust) {
- flags |= OPTS_EXHAUST;
- }
-
- if(this.partial) {
- flags |= OPTS_PARTIAL;
- }
-
- // If batchSize is different to self.numberToReturn
- if(self.batchSize != self.numberToReturn) self.numberToReturn = self.batchSize;
-
- // Allocate write protocol header buffer
- var header = new Buffer(
- 4 * 4 // Header
- + 4 // Flags
- + Buffer.byteLength(self.ns) + 1 // namespace
- + 4 // numberToSkip
- + 4 // numberToReturn
- );
-
- // Add header to buffers
- buffers.push(header);
-
- // Serialize the query
- var query = self.bson.serialize(this.query
- , this.checkKeys
- , true
- , this.serializeFunctions
- , 0, this.ignoreUndefined);
-
- // Add query document
- buffers.push(query);
-
- if(self.returnFieldSelector && Object.keys(self.returnFieldSelector).length > 0) {
- // Serialize the projection document
- projection = self.bson.serialize(this.returnFieldSelector, this.checkKeys, true, this.serializeFunctions, this.ignoreUndefined);
- // Add projection document
- buffers.push(projection);
- }
-
- // Total message size
- var totalLength = header.length + query.length + (projection ? projection.length : 0);
-
- // Set up the index
- var index = 4;
-
- // Write total document length
- header[3] = (totalLength >> 24) & 0xff;
- header[2] = (totalLength >> 16) & 0xff;
- header[1] = (totalLength >> 8) & 0xff;
- header[0] = (totalLength) & 0xff;
-
- // Write header information requestId
- header[index + 3] = (this.requestId >> 24) & 0xff;
- header[index + 2] = (this.requestId >> 16) & 0xff;
- header[index + 1] = (this.requestId >> 8) & 0xff;
- header[index] = (this.requestId) & 0xff;
- index = index + 4;
-
- // Write header information responseTo
- header[index + 3] = (0 >> 24) & 0xff;
- header[index + 2] = (0 >> 16) & 0xff;
- header[index + 1] = (0 >> 8) & 0xff;
- header[index] = (0) & 0xff;
- index = index + 4;
-
- // Write header information OP_QUERY
- header[index + 3] = (OP_QUERY >> 24) & 0xff;
- header[index + 2] = (OP_QUERY >> 16) & 0xff;
- header[index + 1] = (OP_QUERY >> 8) & 0xff;
- header[index] = (OP_QUERY) & 0xff;
- index = index + 4;
-
- // Write header information flags
- header[index + 3] = (flags >> 24) & 0xff;
- header[index + 2] = (flags >> 16) & 0xff;
- header[index + 1] = (flags >> 8) & 0xff;
- header[index] = (flags) & 0xff;
- index = index + 4;
-
- // Write collection name
- index = index + header.write(this.ns, index, 'utf8') + 1;
- header[index - 1] = 0;
-
- // Write header information flags numberToSkip
- header[index + 3] = (this.numberToSkip >> 24) & 0xff;
- header[index + 2] = (this.numberToSkip >> 16) & 0xff;
- header[index + 1] = (this.numberToSkip >> 8) & 0xff;
- header[index] = (this.numberToSkip) & 0xff;
- index = index + 4;
-
- // Write header information flags numberToReturn
- header[index + 3] = (this.numberToReturn >> 24) & 0xff;
- header[index + 2] = (this.numberToReturn >> 16) & 0xff;
- header[index + 1] = (this.numberToReturn >> 8) & 0xff;
- header[index] = (this.numberToReturn) & 0xff;
- index = index + 4;
-
- // Return the buffers
- return buffers;
-}
-
-Query.getRequestId = function() {
- return ++_requestId;
-}
-
-/**************************************************************
- * GETMORE
- **************************************************************/
-var GetMore = function(bson, ns, cursorId, opts) {
- opts = opts || {};
- this.numberToReturn = opts.numberToReturn || 0;
- this.requestId = _requestId++;
- this.bson = bson;
- this.ns = ns;
- this.cursorId = cursorId;
-}
-
-//
-// Uses a single allocated buffer for the process, avoiding multiple memory allocations
-GetMore.prototype.toBin = function() {
- var length = 4 + Buffer.byteLength(this.ns) + 1 + 4 + 8 + (4 * 4);
- // Create command buffer
- var index = 0;
- // Allocate buffer
- var _buffer = new Buffer(length);
-
- // Write header information
- // index = write32bit(index, _buffer, length);
- _buffer[index + 3] = (length >> 24) & 0xff;
- _buffer[index + 2] = (length >> 16) & 0xff;
- _buffer[index + 1] = (length >> 8) & 0xff;
- _buffer[index] = (length) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, requestId);
- _buffer[index + 3] = (this.requestId >> 24) & 0xff;
- _buffer[index + 2] = (this.requestId >> 16) & 0xff;
- _buffer[index + 1] = (this.requestId >> 8) & 0xff;
- _buffer[index] = (this.requestId) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, 0);
- _buffer[index + 3] = (0 >> 24) & 0xff;
- _buffer[index + 2] = (0 >> 16) & 0xff;
- _buffer[index + 1] = (0 >> 8) & 0xff;
- _buffer[index] = (0) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, OP_GETMORE);
- _buffer[index + 3] = (OP_GETMORE >> 24) & 0xff;
- _buffer[index + 2] = (OP_GETMORE >> 16) & 0xff;
- _buffer[index + 1] = (OP_GETMORE >> 8) & 0xff;
- _buffer[index] = (OP_GETMORE) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, 0);
- _buffer[index + 3] = (0 >> 24) & 0xff;
- _buffer[index + 2] = (0 >> 16) & 0xff;
- _buffer[index + 1] = (0 >> 8) & 0xff;
- _buffer[index] = (0) & 0xff;
- index = index + 4;
-
- // Write collection name
- index = index + _buffer.write(this.ns, index, 'utf8') + 1;
- _buffer[index - 1] = 0;
-
- // Write batch size
- // index = write32bit(index, _buffer, numberToReturn);
- _buffer[index + 3] = (this.numberToReturn >> 24) & 0xff;
- _buffer[index + 2] = (this.numberToReturn >> 16) & 0xff;
- _buffer[index + 1] = (this.numberToReturn >> 8) & 0xff;
- _buffer[index] = (this.numberToReturn) & 0xff;
- index = index + 4;
-
- // Write cursor id
- // index = write32bit(index, _buffer, cursorId.getLowBits());
- _buffer[index + 3] = (this.cursorId.getLowBits() >> 24) & 0xff;
- _buffer[index + 2] = (this.cursorId.getLowBits() >> 16) & 0xff;
- _buffer[index + 1] = (this.cursorId.getLowBits() >> 8) & 0xff;
- _buffer[index] = (this.cursorId.getLowBits()) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, cursorId.getHighBits());
- _buffer[index + 3] = (this.cursorId.getHighBits() >> 24) & 0xff;
- _buffer[index + 2] = (this.cursorId.getHighBits() >> 16) & 0xff;
- _buffer[index + 1] = (this.cursorId.getHighBits() >> 8) & 0xff;
- _buffer[index] = (this.cursorId.getHighBits()) & 0xff;
- index = index + 4;
-
- // Return buffer
- return _buffer;
-}
-
-/**************************************************************
- * KILLCURSOR
- **************************************************************/
-var KillCursor = function(bson, cursorIds) {
- this.requestId = _requestId++;
- this.cursorIds = cursorIds;
-}
-
-//
-// Uses a single allocated buffer for the process, avoiding multiple memory allocations
-KillCursor.prototype.toBin = function() {
- var length = 4 + 4 + (4 * 4) + (this.cursorIds.length * 8);
-
- // Create command buffer
- var index = 0;
- var _buffer = new Buffer(length);
-
- // Write header information
- // index = write32bit(index, _buffer, length);
- _buffer[index + 3] = (length >> 24) & 0xff;
- _buffer[index + 2] = (length >> 16) & 0xff;
- _buffer[index + 1] = (length >> 8) & 0xff;
- _buffer[index] = (length) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, requestId);
- _buffer[index + 3] = (this.requestId >> 24) & 0xff;
- _buffer[index + 2] = (this.requestId >> 16) & 0xff;
- _buffer[index + 1] = (this.requestId >> 8) & 0xff;
- _buffer[index] = (this.requestId) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, 0);
- _buffer[index + 3] = (0 >> 24) & 0xff;
- _buffer[index + 2] = (0 >> 16) & 0xff;
- _buffer[index + 1] = (0 >> 8) & 0xff;
- _buffer[index] = (0) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, OP_KILL_CURSORS);
- _buffer[index + 3] = (OP_KILL_CURSORS >> 24) & 0xff;
- _buffer[index + 2] = (OP_KILL_CURSORS >> 16) & 0xff;
- _buffer[index + 1] = (OP_KILL_CURSORS >> 8) & 0xff;
- _buffer[index] = (OP_KILL_CURSORS) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, 0);
- _buffer[index + 3] = (0 >> 24) & 0xff;
- _buffer[index + 2] = (0 >> 16) & 0xff;
- _buffer[index + 1] = (0 >> 8) & 0xff;
- _buffer[index] = (0) & 0xff;
- index = index + 4;
-
- // Write batch size
- // index = write32bit(index, _buffer, this.cursorIds.length);
- _buffer[index + 3] = (this.cursorIds.length >> 24) & 0xff;
- _buffer[index + 2] = (this.cursorIds.length >> 16) & 0xff;
- _buffer[index + 1] = (this.cursorIds.length >> 8) & 0xff;
- _buffer[index] = (this.cursorIds.length) & 0xff;
- index = index + 4;
-
- // Write all the cursor ids into the array
- for(var i = 0; i < this.cursorIds.length; i++) {
- // Write cursor id
- // index = write32bit(index, _buffer, cursorIds[i].getLowBits());
- _buffer[index + 3] = (this.cursorIds[i].getLowBits() >> 24) & 0xff;
- _buffer[index + 2] = (this.cursorIds[i].getLowBits() >> 16) & 0xff;
- _buffer[index + 1] = (this.cursorIds[i].getLowBits() >> 8) & 0xff;
- _buffer[index] = (this.cursorIds[i].getLowBits()) & 0xff;
- index = index + 4;
-
- // index = write32bit(index, _buffer, cursorIds[i].getHighBits());
- _buffer[index + 3] = (this.cursorIds[i].getHighBits() >> 24) & 0xff;
- _buffer[index + 2] = (this.cursorIds[i].getHighBits() >> 16) & 0xff;
- _buffer[index + 1] = (this.cursorIds[i].getHighBits() >> 8) & 0xff;
- _buffer[index] = (this.cursorIds[i].getHighBits()) & 0xff;
- index = index + 4;
- }
-
- // Return buffer
- return _buffer;
-}
-
-var Response = function(bson, data, opts) {
- opts = opts || {promoteLongs: true};
- this.parsed = false;
-
- //
- // Parse Header
- //
- this.index = 0;
- this.raw = data;
- this.data = data;
- this.bson = bson;
- this.opts = opts;
-
- // Read the message length
- this.length = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
-
- // Fetch the request id for this reply
- this.requestId = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
-
- // Fetch the id of the request that triggered the response
- this.responseTo = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
-
- // Skip op-code field
- this.index = this.index + 4;
-
- // Unpack flags
- this.responseFlags = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
-
- // Unpack the cursor
- var lowBits = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
- var highBits = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
- // Create long object
- this.cursorId = new Long(lowBits, highBits);
-
- // Unpack the starting from
- this.startingFrom = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
-
- // Unpack the number of objects returned
- this.numberReturned = data[this.index] | data[this.index + 1] << 8 | data[this.index + 2] << 16 | data[this.index + 3] << 24;
- this.index = this.index + 4;
-
- // Preallocate document array
- this.documents = new Array(this.numberReturned);
-
- // Flag values
- this.cursorNotFound = (this.responseFlags & CURSOR_NOT_FOUND) != 0;
- this.queryFailure = (this.responseFlags & QUERY_FAILURE) != 0;
- this.shardConfigStale = (this.responseFlags & SHARD_CONFIG_STALE) != 0;
- this.awaitCapable = (this.responseFlags & AWAIT_CAPABLE) != 0;
- this.promoteLongs = typeof opts.promoteLongs == 'boolean' ? opts.promoteLongs : true;
-}
-
-Response.prototype.isParsed = function() {
- return this.parsed;
-}
-
-// Validation buffers
-var firstBatch = new Buffer('firstBatch', 'utf8');
-var nextBatch = new Buffer('nextBatch', 'utf8');
-var cursorId = new Buffer('id', 'utf8').toString('hex');
-
-var documentBuffers = {
- firstBatch: firstBatch.toString('hex'),
- nextBatch: nextBatch.toString('hex')
-};
-
-Response.prototype.parse = function(options) {
- // Don't parse again if not needed
- if(this.parsed) return;
- options = options || {};
-
- // Allow the return of raw documents instead of parsing
- var raw = options.raw || false;
- var documentsReturnedIn = options.documentsReturnedIn || null;
-
- //
- // Single document and documentsReturnedIn set
- //
- if(this.numberReturned == 1 && documentsReturnedIn != null && raw) {
- // Calculate the bson size
- var bsonSize = this.data[this.index] | this.data[this.index + 1] << 8 | this.data[this.index + 2] << 16 | this.data[this.index + 3] << 24;
- // Slice out the buffer containing the command result document
- var document = this.data.slice(this.index, this.index + bsonSize);
- // Set up field we wish to keep as raw
- var fieldsAsRaw = {}
- fieldsAsRaw[documentsReturnedIn] = true;
- // Set up the options
- var _options = {promoteLongs: this.opts.promoteLongs, fieldsAsRaw: fieldsAsRaw};
-
- // Deserialize but keep the array of documents in non-parsed form
- var doc = this.bson.deserialize(document, _options);
-
- // Get the documents
- this.documents = doc.cursor[documentsReturnedIn];
- this.numberReturned = this.documents.length;
- // Ensure we have a Long valie cursor id
- this.cursorId = typeof doc.cursor.id == 'number'
- ? Long.fromNumber(doc.cursor.id)
- : doc.cursor.id;
-
- // Adjust the index
- this.index = this.index + bsonSize;
-
- // Set as parsed
- this.parsed = true
- return;
- }
-
- //
- // Parse Body
- //
- for(var i = 0; i < this.numberReturned; i++) {
- var bsonSize = this.data[this.index] | this.data[this.index + 1] << 8 | this.data[this.index + 2] << 16 | this.data[this.index + 3] << 24;
- // Parse options
- var _options = {promoteLongs: this.opts.promoteLongs};
-
- // If we have raw results specified slice the return document
- if(raw) {
- this.documents[i] = this.data.slice(this.index, this.index + bsonSize);
- } else {
- this.documents[i] = this.bson.deserialize(this.data.slice(this.index, this.index + bsonSize), _options);
- }
-
- // Adjust the index
- this.index = this.index + bsonSize;
- }
-
- // Set parsed
- this.parsed = true;
-}
-
-module.exports = {
- Query: Query
- , GetMore: GetMore
- , Response: Response
- , KillCursor: KillCursor
-}
diff --git a/node_modules/mongodb-core/lib/connection/connection.js b/node_modules/mongodb-core/lib/connection/connection.js
deleted file mode 100644
index b44c4e0..0000000
--- a/node_modules/mongodb-core/lib/connection/connection.js
+++ /dev/null
@@ -1,531 +0,0 @@
-"use strict";
-
-var inherits = require('util').inherits
- , EventEmitter = require('events').EventEmitter
- , net = require('net')
- , tls = require('tls')
- , f = require('util').format
- , debugOptions = require('./utils').debugOptions
- , Response = require('./commands').Response
- , MongoError = require('../error')
- , Logger = require('./logger');
-
-var _id = 0;
-var debugFields = ['host', 'port', 'size', 'keepAlive', 'keepAliveInitialDelay', 'noDelay'
- , 'connectionTimeout', 'socketTimeout', 'singleBufferSerializtion', 'ssl', 'ca', 'cert'
- , 'rejectUnauthorized', 'promoteLongs', 'checkServerIdentity'];
-var connectionAccounting = false;
-var connections = {};
-
-/**
- * Creates a new Connection instance
- * @class
- * @param {string} options.host The server host
- * @param {number} options.port The server port
- * @param {boolean} [options.keepAlive=true] TCP Connection keep alive enabled
- * @param {number} [options.keepAliveInitialDelay=0] Initial delay before TCP keep alive enabled
- * @param {boolean} [options.noDelay=true] TCP Connection no delay
- * @param {number} [options.connectionTimeout=0] TCP Connection timeout setting
- * @param {number} [options.socketTimeout=0] TCP Socket timeout setting
- * @param {boolean} [options.singleBufferSerializtion=true] Serialize into single buffer, trade of peak memory for serialization speed
- * @param {boolean} [options.ssl=false] Use SSL for connection
- * @param {boolean|function} [options.checkServerIdentity=true] Ensure we check server identify during SSL, set to false to disable checking. Only works for Node 0.12.x or higher. You can pass in a boolean or your own checkServerIdentity override function.
- * @param {Buffer} [options.ca] SSL Certificate store binary buffer
- * @param {Buffer} [options.cert] SSL Certificate binary buffer
- * @param {Buffer} [options.key] SSL Key file binary buffer
- * @param {string} [options.passphrase] SSL Certificate pass phrase
- * @param {boolean} [options.rejectUnauthorized=true] Reject unauthorized server certificates
- * @param {boolean} [options.promoteLongs=true] Convert Long values from the db into Numbers if they fit into 53 bits
- * @fires Connection#connect
- * @fires Connection#close
- * @fires Connection#error
- * @fires Connection#timeout
- * @fires Connection#parseError
- * @return {Connection} A cursor instance
- */
-var Connection = function(messageHandler, options) {
- // Add event listener
- EventEmitter.call(this);
- // Set empty if no options passed
- this.options = options || {};
- // Identification information
- this.id = _id++;
- // Logger instance
- this.logger = Logger('Connection', options);
- // No bson parser passed in
- if(!options.bson) throw new Error("must pass in valid bson parser");
- // Get bson parser
- this.bson = options.bson;
- // Grouping tag used for debugging purposes
- this.tag = options.tag;
- // Message handler
- this.messageHandler = messageHandler;
-
- // Max BSON message size
- this.maxBsonMessageSize = options.maxBsonMessageSize || (1024 * 1024 * 16 * 4);
- // Debug information
- if(this.logger.isDebug()) this.logger.debug(f('creating connection %s with options [%s]', this.id, JSON.stringify(debugOptions(debugFields, options))));
-
- // Default options
- this.port = options.port || 27017;
- this.host = options.host || 'localhost';
- this.keepAlive = typeof options.keepAlive == 'boolean' ? options.keepAlive : true;
- this.keepAliveInitialDelay = options.keepAliveInitialDelay || 0;
- this.noDelay = typeof options.noDelay == 'boolean' ? options.noDelay : true;
- this.connectionTimeout = options.connectionTimeout || 0;
- this.socketTimeout = options.socketTimeout || 0;
-
- // If connection was destroyed
- this.destroyed = false;
-
- // Check if we have a domain socket
- this.domainSocket = this.host.indexOf('\/') != -1;
-
- // Serialize commands using function
- this.singleBufferSerializtion = typeof options.singleBufferSerializtion == 'boolean' ? options.singleBufferSerializtion : true;
- this.serializationFunction = this.singleBufferSerializtion ? 'toBinUnified' : 'toBin';
-
- // SSL options
- this.ca = options.ca || null;
- this.cert = options.cert || null;
- this.key = options.key || null;
- this.passphrase = options.passphrase || null;
- this.ssl = typeof options.ssl == 'boolean' ? options.ssl : false;
- this.rejectUnauthorized = typeof options.rejectUnauthorized == 'boolean' ? options.rejectUnauthorized : true;
- this.checkServerIdentity = typeof options.checkServerIdentity == 'boolean'
- || typeof options.checkServerIdentity == 'function' ? options.checkServerIdentity : true;
-
- // If ssl not enabled
- if(!this.ssl) this.rejectUnauthorized = false;
-
- // Response options
- this.responseOptions = {
- promoteLongs: typeof options.promoteLongs == 'boolean' ? options.promoteLongs : true
- }
-
- // Flushing
- this.flushing = false;
- this.queue = [];
-
- // Internal state
- this.connection = null;
- this.writeStream = null;
-}
-
-inherits(Connection, EventEmitter);
-
-Connection.prototype.setSocketTimeout = function(value) {
- if(this.connection) {
- this.connection.setTimeout(value);
- }
-}
-
-Connection.prototype.resetSocketTimeout = function(value) {
- if(this.connection) {
- this.connection.setTimeout(this.socketTimeout);;
- }
-}
-
-Connection.enableConnectionAccounting = function() {
- connectionAccounting = true;
- connections = {};
-}
-
-Connection.disableConnectionAccounting = function() {
- connectionAccounting = false;
-}
-
-Connection.connections = function() {
- return connections;
-}
-
-//
-// Connection handlers
-var errorHandler = function(self) {
- return function(err) {
- if(connectionAccounting) delete connections[self.id];
- // Debug information
- if(self.logger.isDebug()) self.logger.debug(f('connection %s for [%s:%s] errored out with [%s]', self.id, self.host, self.port, JSON.stringify(err)));
- // Emit the error
- if(self.listeners('error').length > 0) self.emit("error", MongoError.create(err), self);
- }
-}
-
-var timeoutHandler = function(self) {
- return function(err) {
- if(connectionAccounting) delete connections[self.id];
- // Debug information
- if(self.logger.isDebug()) self.logger.debug(f('connection %s for [%s:%s] timed out', self.id, self.host, self.port));
- // Emit timeout error
- self.emit("timeout"
- , MongoError.create(f("connection %s to %s:%s timed out", self.id, self.host, self.port))
- , self);
- }
-}
-
-var closeHandler = function(self) {
- return function(hadError) {
- if(connectionAccounting) delete connections[self.id];
- // Debug information
- if(self.logger.isDebug()) self.logger.debug(f('connection %s with for [%s:%s] closed', self.id, self.host, self.port));
-
- // Emit close event
- if(!hadError) {
- self.emit("close"
- , MongoError.create(f("connection %s to %s:%s closed", self.id, self.host, self.port))
- , self);
- }
- }
-}
-
-var dataHandler = function(self) {
- return function(data) {
- // Parse until we are done with the data
- while(data.length > 0) {
- // If we still have bytes to read on the current message
- if(self.bytesRead > 0 && self.sizeOfMessage > 0) {
- // Calculate the amount of remaining bytes
- var remainingBytesToRead = self.sizeOfMessage - self.bytesRead;
- // Check if the current chunk contains the rest of the message
- if(remainingBytesToRead > data.length) {
- // Copy the new data into the exiting buffer (should have been allocated when we know the message size)
- data.copy(self.buffer, self.bytesRead);
- // Adjust the number of bytes read so it point to the correct index in the buffer
- self.bytesRead = self.bytesRead + data.length;
-
- // Reset state of buffer
- data = new Buffer(0);
- } else {
- // Copy the missing part of the data into our current buffer
- data.copy(self.buffer, self.bytesRead, 0, remainingBytesToRead);
- // Slice the overflow into a new buffer that we will then re-parse
- data = data.slice(remainingBytesToRead);
-
- // Emit current complete message
- try {
- var emitBuffer = self.buffer;
- // Reset state of buffer
- self.buffer = null;
- self.sizeOfMessage = 0;
- self.bytesRead = 0;
- self.stubBuffer = null;
- // Emit the buffer
- self.messageHandler(new Response(self.bson, emitBuffer, self.responseOptions), self);
- } catch(err) {
- var errorObject = {err:"socketHandler", trace:err, bin:self.buffer, parseState:{
- sizeOfMessage:self.sizeOfMessage,
- bytesRead:self.bytesRead,
- stubBuffer:self.stubBuffer}};
- // We got a parse Error fire it off then keep going
- self.emit("parseError", errorObject, self);
- }
- }
- } else {
- // Stub buffer is kept in case we don't get enough bytes to determine the
- // size of the message (< 4 bytes)
- if(self.stubBuffer != null && self.stubBuffer.length > 0) {
- // If we have enough bytes to determine the message size let's do it
- if(self.stubBuffer.length + data.length > 4) {
- // Prepad the data
- var newData = new Buffer(self.stubBuffer.length + data.length);
- self.stubBuffer.copy(newData, 0);
- data.copy(newData, self.stubBuffer.length);
- // Reassign for parsing
- data = newData;
-
- // Reset state of buffer
- self.buffer = null;
- self.sizeOfMessage = 0;
- self.bytesRead = 0;
- self.stubBuffer = null;
-
- } else {
-
- // Add the the bytes to the stub buffer
- var newStubBuffer = new Buffer(self.stubBuffer.length + data.length);
- // Copy existing stub buffer
- self.stubBuffer.copy(newStubBuffer, 0);
- // Copy missing part of the data
- data.copy(newStubBuffer, self.stubBuffer.length);
- // Exit parsing loop
- data = new Buffer(0);
- }
- } else {
- if(data.length > 4) {
- // Retrieve the message size
- // var sizeOfMessage = data.readUInt32LE(0);
- var sizeOfMessage = data[0] | data[1] << 8 | data[2] << 16 | data[3] << 24;
- // If we have a negative sizeOfMessage emit error and return
- if(sizeOfMessage < 0 || sizeOfMessage > self.maxBsonMessageSize) {
- var errorObject = {err:"socketHandler", trace:'', bin:self.buffer, parseState:{
- sizeOfMessage: sizeOfMessage,
- bytesRead: self.bytesRead,
- stubBuffer: self.stubBuffer}};
- // We got a parse Error fire it off then keep going
- self.emit("parseError", errorObject, self);
- return;
- }
-
- // Ensure that the size of message is larger than 0 and less than the max allowed
- if(sizeOfMessage > 4 && sizeOfMessage < self.maxBsonMessageSize && sizeOfMessage > data.length) {
- self.buffer = new Buffer(sizeOfMessage);
- // Copy all the data into the buffer
- data.copy(self.buffer, 0);
- // Update bytes read
- self.bytesRead = data.length;
- // Update sizeOfMessage
- self.sizeOfMessage = sizeOfMessage;
- // Ensure stub buffer is null
- self.stubBuffer = null;
- // Exit parsing loop
- data = new Buffer(0);
-
- } else if(sizeOfMessage > 4 && sizeOfMessage < self.maxBsonMessageSize && sizeOfMessage == data.length) {
- try {
- var emitBuffer = data;
- // Reset state of buffer
- self.buffer = null;
- self.sizeOfMessage = 0;
- self.bytesRead = 0;
- self.stubBuffer = null;
- // Exit parsing loop
- data = new Buffer(0);
- // Emit the message
- self.messageHandler(new Response(self.bson, emitBuffer, self.responseOptions), self);
- } catch (err) {
- self.emit("parseError", err, self);
- }
- } else if(sizeOfMessage <= 4 || sizeOfMessage > self.maxBsonMessageSize) {
- var errorObject = {err:"socketHandler", trace:null, bin:data, parseState:{
- sizeOfMessage:sizeOfMessage,
- bytesRead:0,
- buffer:null,
- stubBuffer:null}};
- // We got a parse Error fire it off then keep going
- self.emit("parseError", errorObject, self);
-
- // Clear out the state of the parser
- self.buffer = null;
- self.sizeOfMessage = 0;
- self.bytesRead = 0;
- self.stubBuffer = null;
- // Exit parsing loop
- data = new Buffer(0);
- } else {
- var emitBuffer = data.slice(0, sizeOfMessage);
- // Reset state of buffer
- self.buffer = null;
- self.sizeOfMessage = 0;
- self.bytesRead = 0;
- self.stubBuffer = null;
- // Copy rest of message
- data = data.slice(sizeOfMessage);
- // Emit the message
- self.messageHandler(new Response(self.bson, emitBuffer, self.responseOptions), self);
- }
- } else {
- // Create a buffer that contains the space for the non-complete message
- self.stubBuffer = new Buffer(data.length)
- // Copy the data to the stub buffer
- data.copy(self.stubBuffer, 0);
- // Exit parsing loop
- data = new Buffer(0);
- }
- }
- }
- }
- }
-}
-
-/**
- * Connect
- * @method
- */
-Connection.prototype.connect = function(_options) {
- var self = this;
- _options = _options || {};
- // Set the connections
- if(connectionAccounting) connections[this.id] = this;
- // Check if we are overriding the promoteLongs
- if(typeof _options.promoteLongs == 'boolean') {
- self.responseOptions.promoteLongs = _options.promoteLongs;
- }
-
- // Create new connection instance
- self.connection = self.domainSocket
- ? net.createConnection(self.host)
- : net.createConnection(self.port, self.host);
-
- // Set the options for the connection
- self.connection.setKeepAlive(self.keepAlive, self.keepAliveInitialDelay);
- self.connection.setTimeout(self.connectionTimeout);
- self.connection.setNoDelay(self.noDelay);
-
- // If we have ssl enabled
- if(self.ssl) {
- var sslOptions = {
- socket: self.connection
- , rejectUnauthorized: self.rejectUnauthorized
- }
-
- if(self.ca) sslOptions.ca = self.ca;
- if(self.cert) sslOptions.cert = self.cert;
- if(self.key) sslOptions.key = self.key;
- if(self.passphrase) sslOptions.passphrase = self.passphrase;
-
- // Override checkServerIdentity behavior
- if(self.checkServerIdentity == false) {
- // Skip the identiy check by retuning undefined as per node documents
- // https://nodejs.org/api/tls.html#tls_tls_connect_options_callback
- sslOptions.checkServerIdentity = function(servername, cert) {
- return undefined;
- }
- } else if(typeof self.checkServerIdentity == 'function') {
- sslOptions.checkServerIdentity = self.checkServerIdentity;
- }
-
- // Attempt SSL connection
- self.connection = tls.connect(self.port, self.host, sslOptions, function() {
- // Error on auth or skip
- if(self.connection.authorizationError && self.rejectUnauthorized) {
- return self.emit("error", self.connection.authorizationError, self, {ssl:true});
- }
-
- // Set socket timeout instead of connection timeout
- self.connection.setTimeout(self.socketTimeout);
- // We are done emit connect
- self.emit('connect', self);
- });
- self.connection.setTimeout(self.connectionTimeout);
- } else {
- self.connection.on('connect', function() {
- // Set socket timeout instead of connection timeout
- self.connection.setTimeout(self.socketTimeout);
- // Emit connect event
- self.emit('connect', self);
- });
- }
-
- // Add handlers for events
- self.connection.once('error', errorHandler(self));
- self.connection.once('timeout', timeoutHandler(self));
- self.connection.once('close', closeHandler(self));
- self.connection.on('data', dataHandler(self));
-}
-
-/**
- * Unref this connection
- * @method
- * @return {boolean}
- */
-Connection.prototype.unref = function() {
- if (this.connection) this.connection.unref();
- else {
- var self = this;
- this.once('connect', function() {
- self.connection.unref();
- });
- }
-}
-
-/**
- * Destroy connection
- * @method
- */
-Connection.prototype.destroy = function() {
- // Set the connections
- if(connectionAccounting) delete connections[this.id];
- if(this.connection) {
- this.connection.end();
- this.connection.destroy();
- }
-
- this.destroyed = true;
-}
-
-/**
- * Write to connection
- * @method
- * @param {Command} command Command to write out need to implement toBin and toBinUnified
- */
-Connection.prototype.write = function(buffer) {
- // Debug Log
- if(this.logger.isDebug()) {
- if(!Array.isArray(buffer)) {
- this.logger.debug(f('writing buffer [%s] to %s:%s', buffer.toString('hex'), this.host, this.port));
- } else {
- for(var i = 0; i < buffer.length; i++)
- this.logger.debug(f('writing buffer [%s] to %s:%s', buffer[i].toString('hex'), this.host, this.port));
- }
- }
-
- // Write out the command
- if(!Array.isArray(buffer)) return this.connection.write(buffer, 'binary');
- // Iterate over all buffers and write them in order to the socket
- for(var i = 0; i < buffer.length; i++) this.connection.write(buffer[i], 'binary');
-}
-
-/**
- * Return id of connection as a string
- * @method
- * @return {string}
- */
-Connection.prototype.toString = function() {
- return "" + this.id;
-}
-
-/**
- * Return json object of connection
- * @method
- * @return {object}
- */
-Connection.prototype.toJSON = function() {
- return {id: this.id, host: this.host, port: this.port};
-}
-
-/**
- * Is the connection connected
- * @method
- * @return {boolean}
- */
-Connection.prototype.isConnected = function() {
- if(this.destroyed) return false;
- return !this.connection.destroyed && this.connection.writable;
-}
-
-/**
- * A server connect event, used to verify that the connection is up and running
- *
- * @event Connection#connect
- * @type {Connection}
- */
-
-/**
- * The server connection closed, all pool connections closed
- *
- * @event Connection#close
- * @type {Connection}
- */
-
-/**
- * The server connection caused an error, all pool connections closed
- *
- * @event Connection#error
- * @type {Connection}
- */
-
-/**
- * The server connection timed out, all pool connections closed
- *
- * @event Connection#timeout
- * @type {Connection}
- */
-
-/**
- * The driver experienced an invalid message, all pool connections closed
- *
- * @event Connection#parseError
- * @type {Connection}
- */
-
-module.exports = Connection;
diff --git a/node_modules/mongodb-core/lib/connection/logger.js b/node_modules/mongodb-core/lib/connection/logger.js
deleted file mode 100644
index cba8954..0000000
--- a/node_modules/mongodb-core/lib/connection/logger.js
+++ /dev/null
@@ -1,229 +0,0 @@
-"use strict";
-
-var f = require('util').format
- , MongoError = require('../error');
-
-// Filters for classes
-var classFilters = {};
-var filteredClasses = {};
-var level = null;
-// Save the process id
-var pid = process.pid;
-// current logger
-var currentLogger = null;
-
-/**
- * Creates a new Logger instance
- * @class
- * @param {string} className The Class name associated with the logging instance
- * @param {object} [options=null] Optional settings.
- * @param {Function} [options.logger=null] Custom logger function;
- * @param {string} [options.loggerLevel=error] Override default global log level.
- * @return {Logger} a Logger instance.
- */
-var Logger = function(className, options) {
- if(!(this instanceof Logger)) return new Logger(className, options);
- options = options || {};
-
- // Current reference
- var self = this;
- this.className = className;
-
- // Current logger
- if(options.logger) {
- currentLogger = options.logger;
- } else if(currentLogger == null) {
- currentLogger = console.log;
- }
-
- // Set level of logging, default is error
- if(options.loggerLevel) {
- level = options.loggerLevel || 'error';
- }
-
- // Add all class names
- if(filteredClasses[this.className] == null) classFilters[this.className] = true;
-}
-
-/**
- * Log a message at the debug level
- * @method
- * @param {string} message The message to log
- * @param {object} object additional meta data to log
- * @return {null}
- */
-Logger.prototype.debug = function(message, object) {
- if(this.isDebug()
- && ((Object.keys(filteredClasses).length > 0 && filteredClasses[this.className])
- || (Object.keys(filteredClasses).length == 0 && classFilters[this.className]))) {
- var dateTime = new Date().getTime();
- var msg = f("[%s-%s:%s] %s %s", 'DEBUG', this.className, pid, dateTime, message);
- var state = {
- type: 'debug', message: message, className: this.className, pid: pid, date: dateTime
- };
- if(object) state.meta = object;
- currentLogger(msg, state);
- }
-}
-
-/**
- * Log a message at the warn level
- * @method
- * @param {string} message The message to log
- * @param {object} object additional meta data to log
- * @return {null}
- */
-Logger.prototype.warn = function(message, object) {
- if(this.isWarn()
- && ((Object.keys(filteredClasses).length > 0 && filteredClasses[this.className])
- || (Object.keys(filteredClasses).length == 0 && classFilters[this.className]))) {
- var dateTime = new Date().getTime();
- var msg = f("[%s-%s:%s] %s %s", 'WARN', this.className, pid, dateTime, message);
- var state = {
- type: 'warn', message: message, className: this.className, pid: pid, date: dateTime
- };
- if(object) state.meta = object;
- currentLogger(msg, state);
- }
-},
-
-/**
- * Log a message at the info level
- * @method
- * @param {string} message The message to log
- * @param {object} object additional meta data to log
- * @return {null}
- */
-Logger.prototype.info = function(message, object) {
- if(this.isInfo()
- && ((Object.keys(filteredClasses).length > 0 && filteredClasses[this.className])
- || (Object.keys(filteredClasses).length == 0 && classFilters[this.className]))) {
- var dateTime = new Date().getTime();
- var msg = f("[%s-%s:%s] %s %s", 'INFO', this.className, pid, dateTime, message);
- var state = {
- type: 'info', message: message, className: this.className, pid: pid, date: dateTime
- };
- if(object) state.meta = object;
- currentLogger(msg, state);
- }
-},
-
-/**
- * Log a message at the error level
- * @method
- * @param {string} message The message to log
- * @param {object} object additional meta data to log
- * @return {null}
- */
-Logger.prototype.error = function(message, object) {
- if(this.isError()
- && ((Object.keys(filteredClasses).length > 0 && filteredClasses[this.className])
- || (Object.keys(filteredClasses).length == 0 && classFilters[this.className]))) {
- var dateTime = new Date().getTime();
- var msg = f("[%s-%s:%s] %s %s", 'ERROR', this.className, pid, dateTime, message);
- var state = {
- type: 'error', message: message, className: this.className, pid: pid, date: dateTime
- };
- if(object) state.meta = object;
- currentLogger(msg, state);
- }
-},
-
-/**
- * Is the logger set at info level
- * @method
- * @return {boolean}
- */
-Logger.prototype.isInfo = function() {
- return level == 'info' || level == 'debug';
-},
-
-/**
- * Is the logger set at error level
- * @method
- * @return {boolean}
- */
-Logger.prototype.isError = function() {
- return level == 'error' || level == 'info' || level == 'debug';
-},
-
-/**
- * Is the logger set at error level
- * @method
- * @return {boolean}
- */
-Logger.prototype.isWarn = function() {
- return level == 'error' || level == 'warn' || level == 'info' || level == 'debug';
-},
-
-/**
- * Is the logger set at debug level
- * @method
- * @return {boolean}
- */
-Logger.prototype.isDebug = function() {
- return level == 'debug';
-}
-
-/**
- * Resets the logger to default settings, error and no filtered classes
- * @method
- * @return {null}
- */
-Logger.reset = function() {
- level = 'error';
- filteredClasses = {};
-}
-
-/**
- * Get the current logger function
- * @method
- * @return {function}
- */
-Logger.currentLogger = function() {
- return currentLogger;
-}
-
-/**
- * Set the current logger function
- * @method
- * @param {function} logger Logger function.
- * @return {null}
- */
-Logger.setCurrentLogger = function(logger) {
- if(typeof logger != 'function') throw new MongoError("current logger must be a function");
- currentLogger = logger;
-}
-
-/**
- * Set what classes to log.
- * @method
- * @param {string} type The type of filter (currently only class)
- * @param {string[]} values The filters to apply
- * @return {null}
- */
-Logger.filter = function(type, values) {
- if(type == 'class' && Array.isArray(values)) {
- filteredClasses = {};
-
- values.forEach(function(x) {
- filteredClasses[x] = true;
- });
- }
-}
-
-/**
- * Set the current log level
- * @method
- * @param {string} level Set current log level (debug, info, error)
- * @return {null}
- */
-Logger.setLevel = function(_level) {
- if(_level != 'info' && _level != 'error' && _level != 'debug' && _level != 'warn') {
- throw new Error(f("%s is an illegal logging level", _level));
- }
-
- level = _level;
-}
-
-module.exports = Logger;
diff --git a/node_modules/mongodb-core/lib/connection/pool.js b/node_modules/mongodb-core/lib/connection/pool.js
deleted file mode 100644
index 4f1c83b..0000000
--- a/node_modules/mongodb-core/lib/connection/pool.js
+++ /dev/null
@@ -1,1122 +0,0 @@
-"use strict";
-
-var inherits = require('util').inherits,
- EventEmitter = require('events').EventEmitter,
- Connection = require('./connection'),
- MongoError = require('../error'),
- Logger = require('./logger'),
- f = require('util').format,
- Query = require('./commands').Query,
- CommandResult = require('./command_result'),
- assign = require('../topologies/shared').assign;
-
-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 DESTROYING = 'destroying';
-var DESTROYED = 'destroyed';
-
-var _id = 0;
-
-/**
- * Creates a new Pool instance
- * @class
- * @param {string} options.host The server host
- * @param {number} options.port The server port
- * @param {number} [options.size=1] Max server connection pool size
- * @param {boolean} [options.reconnect=true] Server will attempt to reconnect on loss of connection
- * @param {number} [options.reconnectTries=30] Server attempt to reconnect #times
- * @param {number} [options.reconnectInterval=1000] Server will wait # milliseconds between retries
- * @param {boolean} [options.keepAlive=true] TCP Connection keep alive enabled
- * @param {number} [options.keepAliveInitialDelay=0] Initial delay before TCP keep alive enabled
- * @param {boolean} [options.noDelay=true] TCP Connection no delay
- * @param {number} [options.connectionTimeout=0] TCP Connection timeout setting
- * @param {number} [options.socketTimeout=0] TCP Socket timeout setting
- * @param {number} [options.monitoringSocketTimeout=30000] TCP Socket timeout setting for replicaset monitoring socket
- * @param {boolean} [options.ssl=false] Use SSL for connection
- * @param {boolean|function} [options.checkServerIdentity=true] Ensure we check server identify during SSL, set to false to disable checking. Only works for Node 0.12.x or higher. You can pass in a boolean or your own checkServerIdentity override function.
- * @param {Buffer} [options.ca] SSL Certificate store binary buffer
- * @param {Buffer} [options.cert] SSL Certificate binary buffer
- * @param {Buffer} [options.key] SSL Key file binary buffer
- * @param {string} [options.passPhrase] SSL Certificate pass phrase
- * @param {boolean} [options.rejectUnauthorized=false] Reject unauthorized server certificates
- * @param {boolean} [options.promoteLongs=true] Convert Long values from the db into Numbers if they fit into 53 bits
- * @fires Pool#connect
- * @fires Pool#close
- * @fires Pool#error
- * @fires Pool#timeout
- * @fires Pool#parseError
- * @return {Pool} A cursor instance
- */
-var Pool = function(options) {
- var self = this;
- // Add event listener
- EventEmitter.call(this);
- // Add the options
- this.options = 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);
-
- // Identification information
- this.id = _id++;
- // 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 = options.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.loggingout = false;
- this.nonAuthenticatedConnections = [];
- this.authenticatingTimestamp = null;
- // Number of consecutive timeouts caught
- this.numberOfConsecutiveTimeouts = 0;
-}
-
-inherits(Pool, EventEmitter);
-
-Object.defineProperty(Pool.prototype, 'size', {
- enumerable:true,
- get: function() { return this.options.size; }
-});
-
-Object.defineProperty(Pool.prototype, 'connectionTimeout', {
- enumerable:true,
- get: function() { return this.options.connectionTimeout; }
-});
-
-Object.defineProperty(Pool.prototype, 'socketTimeout', {
- enumerable:true,
- get: function() { return this.options.socketTimeout; }
-});
-
-function stateTransition(self, newState) {
- var legalTransitions = {
- 'disconnected': [CONNECTING, DESTROYING, DISCONNECTED],
- 'connecting': [CONNECTING, DESTROYING, CONNECTED, DISCONNECTED],
- 'connected': [CONNECTED, DISCONNECTED, DESTROYING],
- 'destroying': [DESTROYING, DESTROYED],
- 'destroyed': [DESTROYED]
- }
-
- // Get current state
- var legalStates = legalTransitions[self.state];
- if(legalStates && legalStates.indexOf(newState) != -1) {
- self.state = newState;
- } else {
- self.logger.error(f('Pool with id [%s] failed attempted illegal state transition from [%s] to [%s] only following state allowed [%s]'
- , self.id, self.state, newState, legalStates));
- }
-}
-
-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];
-
- // Authenticate using the provided mechanism
- provider.auth.apply(provider, [write(pool), [connection], db].concat(auth.slice(2)).concat([cb]));
-}
-
-// The write function used by the authentication mechanism (bypasses external)
-function write(self) {
- return function(connection, buffer, callback) {
- // Ensure we stop auth if pool was destroyed
- if(self.state == DESTROYED || self.state == DESTROYING) {
- return callback(new MongoError('pool destroyed'));
- }
- // Set the connection workItem callback
- connection.workItem = {cb: callback, command: true};
- // Write the buffer out to the connection
- connection.write(buffer);
- };
-}
-
-
-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()];
-
- // Auth provider
- provider.reauthenticate(write(pool), [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) {
- removeConnection(self, this);
-
- // Flush out the callback if there is one
- if(this.workItem && this.workItem.cb) {
- var workItem = this.workItem;
- this.workItem = null;
- workItem.cb(err);
- }
-
- // Did we catch a timeout, increment the numberOfConsecutiveTimeouts
- if(event == 'timeout') {
- self.numberOfConsecutiveTimeouts = self.numberOfConsecutiveTimeouts + 1;
-
- // Have we timed out more than reconnectTries in a row ?
- // Force close the pool as we are trying to connect to tcp sink hole
- if(self.numberOfConsecutiveTimeouts > self.options.reconnectTries) {
- self.numberOfConsecutiveTimeouts = 0;
- // Destroy all connections and pool
- self.destroy(true);
- // Emit close event
- return self.emit('close', self);
- }
- }
-
- // No more socket available propegate the event
- if(self.socketCount() == 0) {
- if(self.state != DESTROYED && self.state != DESTROYING) {
- stateTransition(self, 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.reconnectId && self.options.reconnect) {
- self.reconnectId = setTimeout(attemptReconnect(self), self.options.reconnectInterval);
- }
- };
-}
-
-function attemptReconnect(self) {
- return function() {
- self.emit('attemptReconnect', self);
- if(self.state == DESTROYED || self.state == DESTROYING) return;
-
- // We are connected do not try again
- if(self.isConnected()) {
- self.reconnectId = null;
- 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('reconnectFailed'
- , new MongoError(f('failed to reconnect after %s attempts with interval %s ms', self.options.reconnectTries, self.options.reconnectInterval)));
- } else {
- self.reconnectId = setTimeout(attemptReconnect(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 || self.state == DESTROYING) {
- 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.on('error', connectionFailureHandler(self, 'error'));
- connection.on('close', connectionFailureHandler(self, 'close'));
- connection.on('timeout', connectionFailureHandler(self, 'timeout'));
- connection.on('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
- self.emit('reconnect', self);
- // Trigger execute to start everything up again
- _execute(self)();
- });
- }
- }
-
- // Create a connection
- var connection = new Connection(messageHandler(self), self.options);
- // Add handlers
- connection.on('close', _connectionFailureHandler(self, 'close'));
- connection.on('error', _connectionFailureHandler(self, 'error'));
- connection.on('timeout', _connectionFailureHandler(self, 'timeout'));
- connection.on('parseError', _connectionFailureHandler(self, 'parseError'));
- // On connection
- connection.on('connect', _connectHandler(self));
- // Attempt connection
- connection.connect();
- }
-}
-
-function moveConnectionBetween(connection, from, to) {
- var index = from.indexOf(connection);
- // Move the connection from connecting to available
- if(index != -1) {
- from.splice(index, 1);
- to.push(connection);
- }
-}
-
-function messageHandler(self) {
- return function(message, connection) {
- // Get the callback
- var workItem = connection.workItem;
- // Reset timeout counter
- self.numberOfConsecutiveTimeouts = 0;
-
- // Reset the connection timeout if we modified it for
- // this operation
- if(workItem.socketTimeout) {
- connection.resetSocketTimeout();
- }
-
- // Log if debug enabled
- if(self.logger.isDebug()) {
- self.logger.debug(f('message [%s] received from %s:%s'
- , message.raw.toString('hex'), self.options.host, self.options.port));
- }
-
- // Authenticate any straggler connections
- function authenticateStragglers(self, connection, callback) {
- // Get any non authenticated connections
- var connections = self.nonAuthenticatedConnections.slice(0);
- var nonAuthenticatedConnections = self.nonAuthenticatedConnections;
- self.nonAuthenticatedConnections = [];
-
- // Establish if the connection need to be authenticated
- // Add to authentication list if
- // 1. we were in an authentication process when the operation was executed
- // 2. our current authentication timestamp is from the workItem one, meaning an auth has happened
- if(connection.workItem.authenticating == true
- || (typeof connection.workItem.authenticatingTimestamp == 'number'
- && connection.workItem.authenticatingTimestamp != self.authenticatingTimestamp)) {
- // Add connection to the list
- connections.push(connection);
- }
-
- // Clear out workItem
- connection.workItem = null;
-
- // No connections need to be re-authenticated
- if(connections.length == 0) {
- // Release the connection back to the pool
- moveConnectionBetween(connection, self.inUseConnections, self.availableConnections);
- // Finish
- return callback();
- }
-
- // Apply re-authentication to all connections before releasing back to pool
- var connectionCount = connections.length;
- // Authenticate all connections
- for(var i = 0; i < connectionCount; i++) {
- reauthenticate(self, connections[i], function(err) {
- connectionCount = connectionCount - 1;
-
- if(connectionCount == 0) {
- // Put non authenticated connections in available connections
- self.availableConnections = self.availableConnections.concat(nonAuthenticatedConnections);
- // Release the connection back to the pool
- moveConnectionBetween(connection, self.inUseConnections, self.availableConnections);
- // Return
- callback();
- }
- });
- }
- }
-
- authenticateStragglers(self, connection, function(err) {
- // Keep executing, ensure current message handler does not stop execution
- process.nextTick(function() {
- _execute(self)();
- });
-
- // Time to dispatch the message if we have a callback
- if(!workItem.immediateRelease) {
- try {
- // Parse the message according to the provided options
- message.parse(workItem);
- } catch(err) {
- return workItem.cb(MongoError.create(err));
- }
-
- // Establish if we have an error
- if(workItem.command && message.documents[0] && (message.documents[0].ok == 0 || message.documents[0]['$err']
- || message.documents[0]['errmsg'] || message.documents[0]['code'])) {
- return workItem.cb(MongoError.create(message.documents[0]));
- }
-
- // Return the documents
- workItem.cb(null, new CommandResult(message.documents[0], connection, message));
- }
- });
- }
-}
-
-/**
- * Return the total socket count in the pool.
- * @method
- * @return {Number} The number of socket available.
- */
-Pool.prototype.socketCount = function() {
- return this.availableConnections.length
- + this.inUseConnections.length
- + this.connectingConnections.length;
-}
-
-/**
- * Return all pool connections
- * @method
- * @return {Connectio[]} The pool connections
- */
-Pool.prototype.allConnections = function() {
- return this.availableConnections
- .concat(this.inUseConnections)
- .concat(this.connectingConnections);
-}
-
-/**
- * Get a pool connection (round-robin)
- * @method
- * @return {Connection}
- */
-Pool.prototype.get = function() {
- return this.allConnections()[0];
-}
-
-/**
- * Is the pool connected
- * @method
- * @return {boolean}
- */
-Pool.prototype.isConnected = function() {
- // We are in a destroyed state
- if(this.state == DESTROYED || this.state == DESTROYING) {
- return false;
- }
-
- // Get connections
- var connections = this.availableConnections
- .concat(this.inUseConnections)
- .concat(this.connectingConnections);
- for(var i = 0; i < connections.length; i++) {
- if(connections[i].isConnected()) return true;
- }
-
- // Might be authenticating, but we are still connected
- if(connections.length == 0 && this.authenticating) {
- return true
- }
-
- // Not connected
- return false;
-}
-
-/**
- * Was the pool destroyed
- * @method
- * @return {boolean}
- */
-Pool.prototype.isDestroyed = function() {
- return this.state == DESTROYED || this.state == DESTROYING;
-}
-
-/**
- * Is the pool in a disconnected state
- * @method
- * @return {boolean}
- */
-Pool.prototype.isDisconnected = function() {
- return this.state == DISCONNECTED;
-}
-
-/**
- * Connect pool
- * @method
- */
-Pool.prototype.connect = function(auth) {
- if(this.state != DISCONNECTED) throw new MongoError('connection in unlawful state ' + this.state);
- var self = this;
- // Transition to connecting state
- stateTransition(this, CONNECTING);
- // Create an array of the arguments
- var args = Array.prototype.slice.call(arguments, 0);
- // Create a connection
- var connection = new Connection(messageHandler(self), this.options);
- // Add to list of connections
- this.connectingConnections.push(connection);
- // Add listeners to the connection
- connection.once('connect', function(connection) {
- if(self.state == DESTROYED || self.state == DESTROYING) return self.destroy();
-
- // Apply any store credentials
- reauthenticate(self, connection, function(err) {
- if(self.state == DESTROYED || self.state == DESTROYING) return self.destroy();
-
- // We have an error emit it
- if(err) {
- // Destroy the pool
- self.destroy();
- // Emit the error
- return self.emit('error', err);
- }
-
- // Authenticate
- authenticate(self, args, connection, function(err) {
- if(self.state == DESTROYED || self.state == DESTROYING) return self.destroy();
-
- // We have an error emit it
- if(err) {
- // Destroy the pool
- self.destroy();
- // Emit the error
- return self.emit('error', err);
- }
- // Set connected mode
- stateTransition(self, CONNECTED);
-
- // Move the active connection
- moveConnectionBetween(connection, self.connectingConnections, self.availableConnections);
-
- // Emit the connect event
- self.emit('connect', self);
- });
- });
- });
-
- // Add error handlers
- connection.once('error', connectionFailureHandler(this, 'error'));
- connection.once('close', connectionFailureHandler(this, 'close'));
- connection.once('timeout', connectionFailureHandler(this, 'timeout'));
- connection.once('parseError', connectionFailureHandler(this, 'parseError'));
-
- try {
- connection.connect();
- } catch(err) {
- // SSL or something threw on connect
- self.emit('error', err);
- }
-}
-
-/**
- * Authenticate using a specified mechanism
- * @method
- * @param {string} mechanism The Auth mechanism we are invoking
- * @param {string} db The db we are invoking the mechanism against
- * @param {...object} param Parameters for the specific mechanism
- * @param {authResultCallback} callback A callback function
- */
-Pool.prototype.auth = function(mechanism, db) {
- var self = this;
- var args = Array.prototype.slice.call(arguments, 0);
- var callback = args.pop();
- // If we are not connected don't allow additonal authentications to happen
- // if(this.state != CONNECTED) throw new MongoError('connection in unlawful state ' + this.state);
-
- // If we don't have the mechanism fail
- if(self.authProviders[mechanism] == null && mechanism != 'default') {
- throw new MongoError(f("auth provider %s does not exist", mechanism));
- }
-
- // Signal that we are authenticating a new set of credentials
- this.authenticating = true;
- this.authenticatingTimestamp = new Date().getTime();
-
- // Authenticate all live connections
- function authenticateLiveConnections(self, args, cb) {
- // Get the current viable connections
- var connections = self.availableConnections;
- // Allow nothing else to use the connections while we authenticate them
- self.availableConnections = [];
-
- var connectionsCount = connections.length;
- var error = null;
- // No connections available, return
- if(connectionsCount == 0) return callback(null);
- // Authenticate the connections
- for(var i = 0; i < connections.length; i++) {
- authenticate(self, args, connections[i], function(err) {
- connectionsCount = connectionsCount - 1;
-
- // Store the error
- if(err) error = err;
-
- // Processed all connections
- if(connectionsCount == 0) {
- // Auth finished
- self.authenticating = false;
- // Add the connections back to available connections
- self.availableConnections = self.availableConnections.concat(connections);
- // We had an error, return it
- if(error) {
- // Log the error
- if(self.logger.isError()) {
- self.logger.error(f('[%s] failed to authenticate against server %s:%s'
- , self.id, self.options.host, self.options.port));
- }
-
- return cb(error);
- }
- cb(null);
- }
- });
- }
- }
-
- // Wait for a logout in process to happen
- function waitForLogout(self, cb) {
- if(!self.loggingout) return cb();
- setTimeout(function() {
- waitForLogout(self, cb);
- }, 1)
- }
-
- // Wait for loggout to finish
- waitForLogout(self, function() {
- // Authenticate all live connections
- authenticateLiveConnections(self, args, function(err) {
- // Credentials correctly stored in auth provider if successful
- // Any new connections will now reauthenticate correctly
- self.authenticating = false;
- // Return after authentication connections
- callback(err);
- });
- });
-}
-
-/**
- * Logout all users against a database
- * @method
- * @param {string} dbName The database name
- * @param {authResultCallback} callback A callback function
- */
-Pool.prototype.logout = function(dbName, callback) {
- var self = this;
- if(typeof dbName != 'string') throw new MongoError('logout method requires a db name as first argument');
- if(typeof callback != 'function') throw new MongoError('logout method requires a callback');
-
- // Indicate logout in process
- this.loggingout = true;
-
- // Get all relevant connections
- var connections = self.availableConnections.concat(self.inUseConnections);
- var count = connections.length;
- // Store any error
- var error = null;
-
- // Send logout command over all the connections
- for(var i = 0; i < connections.length; i++) {
- var query = new Query(this.options.bson
- , f('%s.$cmd', dbName)
- , {logout:1}, {numberToSkip: 0, numberToReturn: 1});
- write(self)(connections[i], query.toBin(), function(err, r) {
- count = count - 1;
- if(err) error = err;
-
- if(count == 0) {
- self.loggingout = false;
- callback(error);
- };
- });
- }
-}
-
-/**
- * Unref the pool
- * @method
- */
-Pool.prototype.unref = function() {
- // Get all the known connections
- var connections = this.availableConnections
- .concat(this.inUseConnections)
- .concat(this.connectingConnections);
- connections.forEach(function(c) {
- c.unref();
- });
-}
-
-// Events
-var events = ['error', 'close', 'timeout', 'parseError', 'connect'];
-
-// Destroy the connections
-function destroy(self, connections) {
- // Destroy all connections
- connections.forEach(function(c) {
- // Remove all listeners
- for(var i = 0; i < events.length; i++) {
- c.removeAllListeners(events[i]);
- }
- // Destroy connection
- c.destroy();
- });
-
- // Zero out all connections
- self.inUseConnections = [];
- self.availableConnections = [];
- self.nonAuthenticatedConnections = [];
- self.connectingConnections = [];
-
- // Set state to destroyed
- stateTransition(self, DESTROYED);
-}
-
-/**
- * Destroy pool
- * @method
- */
-Pool.prototype.destroy = function(force) {
- var self = this;
- // Do not try again if the pool is already dead
- if(this.state == DESTROYED || self.state == DESTROYING) return;
- // Set state to destroyed
- stateTransition(this, DESTROYING);
-
- // Are we force closing
- if(force) {
- // Get all the known connections
- var connections = self.availableConnections
- .concat(self.inUseConnections)
- .concat(self.nonAuthenticatedConnections)
- .concat(self.connectingConnections);
- return destroy(self, connections);
- }
-
- // Wait for the operations to drain before we close the pool
- function checkStatus() {
- if(self.queue.length == 0) {
- // Get all the known connections
- var connections = self.availableConnections
- .concat(self.inUseConnections)
- .concat(self.nonAuthenticatedConnections)
- .concat(self.connectingConnections);
-
- // Check if we have any in flight operations
- for(var i = 0; i < connections.length; i++) {
- // There is an operation still in flight, reschedule a
- // check waiting for it to drain
- if(connections[i].workItem) {
- return setTimeout(checkStatus, 1);
- }
- }
-
- destroy(self, connections);
- } else {
- setTimeout(checkStatus, 1);
- }
- }
-
- // Initiate drain of operations
- checkStatus();
-}
-
-/**
- * Write a message to MongoDB
- * @method
- * @return {Connection}
- */
-Pool.prototype.write = function(buffer, options, cb) {
- // Ensure we have a callback
- if(typeof options == 'function') {
- cb = options;
- }
-
- // Always have options
- options = options || {};
-
- // Pool was destroyed error out
- if(this.state == DESTROYED || this.state == DESTROYING) {
- // Callback with an error
- if(cb) cb(new MongoError('pool destroyed'));
- return;
- }
-
- // Do we have an operation
- var operation = {
- buffer:buffer, cb: cb, raw: false, promoteLongs: true
- };
-
- // Set the options for the parsing
- operation.promoteLongs = typeof options.promoteLongs == 'boolean' ? options.promoteLongs : true;
- operation.raw = typeof options.raw == 'boolean' ? options.raw : false;
- operation.immediateRelease = typeof options.immediateRelease == 'boolean' ? options.immediateRelease : false;
- operation.documentsReturnedIn = options.documentsReturnedIn;
- operation.command = typeof options.command == 'boolean' ? options.command : false;
- // Optional per operation socketTimeout
- operation.socketTimeout = options.socketTimeout;
- operation.monitoring = options.monitoring;
- // // debug
- // operation.cmd = options.cmd;
-
- // We need to have a callback function unless the message returns no response
- if(!(typeof cb == 'function') && !options.noResponse) {
- throw new MongoError('write method must provide a callback');
- }
-
- // If we have a monitoring operation schedule as the very first operation
- // Otherwise add to back of queue
- if(options.monitoring) {
- this.queue.unshift(operation);
- } else {
- this.queue.push(operation);
- }
-
- // Attempt to execute the operation
- _execute(this)();
-}
-
-// Remove connection method
-function remove(connection, connections) {
- for(var i = 0; i < connections.length; i++) {
- if(connections[i] === connection) {
- connections.splice(i, 1);
- return true;
- }
- }
-}
-
-function removeConnection(self, connection) {
- if(remove(connection, self.availableConnections)) return;
- if(remove(connection, self.inUseConnections)) return;
- if(remove(connection, self.connectingConnections)) return;
- if(remove(connection, self.nonAuthenticatedConnections)) return;
-}
-
-// All event handlers
-var handlers = ["close", "message", "error", "timeout", "parseError", "connect"];
-
-function _createConnection(self) {
- var connection = new Connection(messageHandler(self), self.options);
-
- // Push the connection
- self.connectingConnections.push(connection);
-
- // Handle any errors
- var tempErrorHandler = function(_connection) {
- return function(err) {
- // Destroy the connection
- _connection.destroy();
- // Remove the connection from the connectingConnections list
- removeConnection(self, _connection);
- // Start reconnection attempts
- if(!self.reconnectId && self.options.reconnect) {
- self.reconnectId = setTimeout(attemptReconnect(self), self.options.reconnectInterval);
- }
- }
- }
-
- // Handle successful connection
- var tempConnectHandler = function(_connection) {
- return function() {
- // Destroyed state return
- if(self.state == DESTROYED || self.state == DESTROYING) {
- // Remove the connection from the list
- removeConnection(self, _connection);
- return _connection.destroy();
- }
-
- // Destroy all event emitters
- handlers.forEach(function(e) {
- _connection.removeAllListeners(e);
- });
-
- // Add the final handlers
- _connection.once('close', connectionFailureHandler(self, 'close'));
- _connection.once('error', connectionFailureHandler(self, 'error'));
- _connection.once('timeout', connectionFailureHandler(self, 'timeout'));
- _connection.once('parseError', connectionFailureHandler(self, 'parseError'));
-
- // Signal
- reauthenticate(self, _connection, function(err) {
- if(self.state == DESTROYED || self.state == DESTROYING) {
- return _connection.destroy();
- }
- // Remove the connection from the connectingConnections list
- removeConnection(self, _connection);
-
- // Handle error
- if(err) {
- return _connection.destroy();
- }
-
- // If we are authenticating at the moment
- // Do not automatially put in available connections
- // As we need to apply the credentials first
- if(self.authenticating) {
- self.nonAuthenticatedConnections.push(_connection);
- } else {
- // Push to available
- self.availableConnections.push(_connection);
- // Execute any work waiting
- _execute(self)();
- }
- });
- }
- }
-
- // Add all handlers
- connection.once('close', tempErrorHandler(connection));
- connection.once('error', tempErrorHandler(connection));
- connection.once('timeout', tempErrorHandler(connection));
- connection.once('parseError', tempErrorHandler(connection));
- connection.once('connect', tempConnectHandler(connection));
-
- // Start connection
- connection.connect();
-}
-
-function flushMonitoringOperations(queue) {
- for(var i = 0; i < queue.length; i++) {
- if(queue[i].monitoring) {
- var workItem = queue[i];
- queue.splice(i, 1);
- workItem.cb(new MongoError('no connection available for monitoring'));
- }
- }
-}
-
-function _execute(self) {
- return function() {
- if(self.state == DESTROYED) return;
- // Already executing, skip
- if(self.executing) return;
- // Set pool as executing
- self.executing = true;
-
- // Wait for auth to clear before continuing
- function waitForAuth(cb) {
- if(!self.authenticating) return cb();
- // Wait for a milisecond and try again
- setTimeout(function() {
- waitForAuth(cb);
- }, 1);
- }
-
- // Block on any auth in process
- waitForAuth(function() {
- // As long as we have available connections
- while(true) {
- // Total availble connections
- var totalConnections = self.availableConnections.length
- + self.connectingConnections.length
- + self.inUseConnections.length;
-
- // Have we not reached the max connection size yet
- if(self.availableConnections.length == 0
- && self.connectingConnections.length == 0
- && totalConnections < self.options.size
- && self.queue.length > 0) {
- // // Flush any monitoring operations
- // flushMonitoringOperations(self.queue);
- // Create a new connection
- _createConnection(self);
- // Attempt to execute again
- self.executing = false;
- return;
- }
-
- // No available connections available, flush any monitoring ops
- if(self.availableConnections.length == 0) {
- // Flush any monitoring operations
- flushMonitoringOperations(self.queue);
- break;
- }
-
- // No queue break
- if(self.queue.length == 0) {
- break;
- }
-
- // Get a connection
- var connection = self.availableConnections.pop();
- if(connection.isConnected()) {
- // Get the next work item
- var workItem = self.queue.shift();
-
- // Get actual binary commands
- var buffer = workItem.buffer;
-
- // Add connection to workers in flight
- self.inUseConnections.push(connection);
-
- // Set current status of authentication process
- workItem.authenticating = self.authenticating;
- workItem.authenticatingTimestamp = self.authenticatingTimestamp;
-
- // Add current associated callback to the connection
- connection.workItem = workItem
-
- // We have a custom socketTimeout
- if(!workItem.immediateRelease && typeof workItem.socketTimeout == 'number') {
- connection.setSocketTimeout(workItem.socketTimeout);
- }
-
- // Put operation on the wire
- if(Array.isArray(buffer)) {
- for(var i = 0; i < buffer.length; i++) {
- connection.write(buffer[i])
- }
- } else {
- connection.write(buffer);
- }
-
- // Fire and forgot message, release the socket
- if(workItem.immediateRelease && !self.authenticating) {
- self.inUseConnections.pop();
- self.availableConnections.push(connection);
- } else if(workItem.immediateRelease && self.authenticating) {
- self.inUseConnections.pop();
- self.nonAuthenticatedConnections.push(connection);
- }
- } else {
- flushMonitoringOperations(self.queue);
- }
- }
- });
-
- self.executing = false;
- }
-}
-
-/**
- * A server connect event, used to verify that the connection is up and running
- *
- * @event Pool#connect
- * @type {Pool}
- */
-
-/**
- * A server reconnect event, used to verify that pool reconnected.
- *
- * @event Pool#reconnect
- * @type {Pool}
- */
-
-/**
- * The server connection closed, all pool connections closed
- *
- * @event Pool#close
- * @type {Pool}
- */
-
-/**
- * The server connection caused an error, all pool connections closed
- *
- * @event Pool#error
- * @type {Pool}
- */
-
-/**
- * The server connection timed out, all pool connections closed
- *
- * @event Pool#timeout
- * @type {Pool}
- */
-
-/**
- * The driver experienced an invalid message, all pool connections closed
- *
- * @event Pool#parseError
- * @type {Pool}
- */
-
-/**
- * The driver attempted to reconnect
- *
- * @event Pool#attemptReconnect
- * @type {Pool}
- */
-
-/**
- * The driver exhausted all reconnect attempts
- *
- * @event Pool#reconnectFailed
- * @type {Pool}
- */
-
-module.exports = Pool;
diff --git a/node_modules/mongodb-core/lib/connection/utils.js b/node_modules/mongodb-core/lib/connection/utils.js
deleted file mode 100644
index 019ef19..0000000
--- a/node_modules/mongodb-core/lib/connection/utils.js
+++ /dev/null
@@ -1,67 +0,0 @@
-"use strict";
-
-// Set property function
-var setProperty = function(obj, prop, flag, values) {
- Object.defineProperty(obj, prop.name, {
- enumerable:true,
- set: function(value) {
- if(typeof value != 'boolean') throw new Error(f("%s required a boolean", prop.name));
- // Flip the bit to 1
- if(value == true) values.flags |= flag;
- // Flip the bit to 0 if it's set, otherwise ignore
- if(value == false && (values.flags & flag) == flag) values.flags ^= flag;
- prop.value = value;
- }
- , get: function() { return prop.value; }
- });
-}
-
-// Set property function
-var getProperty = function(obj, propName, fieldName, values, func) {
- Object.defineProperty(obj, propName, {
- enumerable:true,
- get: function() {
- // Not parsed yet, parse it
- if(values[fieldName] == null && obj.isParsed && !obj.isParsed()) {
- obj.parse();
- }
-
- // Do we have a post processing function
- if(typeof func == 'function') return func(values[fieldName]);
- // Return raw value
- return values[fieldName];
- }
- });
-}
-
-// Set simple property
-var getSingleProperty = function(obj, name, value) {
- Object.defineProperty(obj, name, {
- enumerable:true,
- get: function() {
- return value
- }
- });
-}
-
-// Shallow copy
-var copy = function(fObj, tObj) {
- tObj = tObj || {};
- for(var name in fObj) tObj[name] = fObj[name];
- return tObj;
-}
-
-var debugOptions = function(debugFields, options) {
- var finaloptions = {};
- debugFields.forEach(function(n) {
- finaloptions[n] = options[n];
- });
-
- return finaloptions;
-}
-
-exports.setProperty = setProperty;
-exports.getProperty = getProperty;
-exports.getSingleProperty = getSingleProperty;
-exports.copy = copy;
-exports.debugOptions = debugOptions;