1/*!
  2 * Module dependencies.
  3 */
  4
  5var MongooseConnection = require('../../connection');
  6var mongo = require('mongodb');
  7var Db = mongo.Db;
  8var Server = mongo.Server;
  9var Mongos = mongo.Mongos;
 10var STATES = require('../../connectionstate');
 11var ReplSetServers = mongo.ReplSet;
 12var DisconnectedError = require('../../error/disconnected');
 13
 14/**
 15 * A [node-mongodb-native](https://github.com/mongodb/node-mongodb-native) connection implementation.
 16 *
 17 * @inherits Connection
 18 * @api private
 19 */
 20
 21function NativeConnection() {
 22  MongooseConnection.apply(this, arguments);
 23  this._listening = false;
 24}
 25
 26/**
 27 * Expose the possible connection states.
 28 * @api public
 29 */
 30
 31NativeConnection.STATES = STATES;
 32
 33/*!
 34 * Inherits from Connection.
 35 */
 36
 37NativeConnection.prototype.__proto__ = MongooseConnection.prototype;
 38
 39/**
 40 * Opens the connection to MongoDB.
 41 *
 42 * @param {Function} fn
 43 * @return {Connection} this
 44 * @api private
 45 */
 46
 47NativeConnection.prototype.doOpen = function(fn) {
 48  var _this = this;
 49  var server = new Server(this.host, this.port, this.options.server);
 50
 51  if (this.options && this.options.mongos) {
 52    var mongos = new Mongos([server], this.options.mongos);
 53    this.db = new Db(this.name, mongos, this.options.db);
 54  } else {
 55    this.db = new Db(this.name, server, this.options.db);
 56  }
 57
 58  this.db.open(function(err) {
 59    listen(_this);
 60
 61    if (!mongos) {
 62      server.s.server.on('error', function(error) {
 63        if (/after \d+ retries/.test(error.message)) {
 64          _this.emit('error', new DisconnectedError(server.s.server.name));
 65        }
 66      });
 67    }
 68
 69    if (err) return fn(err);
 70
 71    fn();
 72  });
 73
 74  return this;
 75};
 76
 77/**
 78 * Switches to a different database using the same connection pool.
 79 *
 80 * Returns a new connection object, with the new db.
 81 *
 82 * @param {String} name The database name
 83 * @return {Connection} New Connection Object
 84 * @api public
 85 */
 86
 87NativeConnection.prototype.useDb = function(name) {
 88  // we have to manually copy all of the attributes...
 89  var newConn = new this.constructor();
 90  newConn.name = name;
 91  newConn.base = this.base;
 92  newConn.collections = {};
 93  newConn.models = {};
 94  newConn.replica = this.replica;
 95  newConn.hosts = this.hosts;
 96  newConn.host = this.host;
 97  newConn.port = this.port;
 98  newConn.user = this.user;
 99  newConn.pass = this.pass;
100  newConn.options = this.options;
101  newConn._readyState = this._readyState;
102  newConn._closeCalled = this._closeCalled;
103  newConn._hasOpened = this._hasOpened;
104  newConn._listening = false;
105
106  // First, when we create another db object, we are not guaranteed to have a
107  // db object to work with. So, in the case where we have a db object and it
108  // is connected, we can just proceed with setting everything up. However, if
109  // we do not have a db or the state is not connected, then we need to wait on
110  // the 'open' event of the connection before doing the rest of the setup
111  // the 'connected' event is the first time we'll have access to the db object
112
113  var _this = this;
114
115  if (this.db && this._readyState === STATES.connected) {
116    wireup();
117  } else {
118    this.once('connected', wireup);
119  }
120
121  function wireup() {
122    newConn.db = _this.db.db(name);
123    newConn.onOpen();
124    // setup the events appropriately
125    listen(newConn);
126  }
127
128  newConn.name = name;
129
130  // push onto the otherDbs stack, this is used when state changes
131  this.otherDbs.push(newConn);
132  newConn.otherDbs.push(this);
133
134  return newConn;
135};
136
137/*!
138 * Register listeners for important events and bubble appropriately.
139 */
140
141function listen(conn) {
142  if (conn.db._listening) {
143    return;
144  }
145  conn.db._listening = true;
146
147  conn.db.on('close', function() {
148    if (conn._closeCalled) return;
149
150    // the driver never emits an `open` event. auto_reconnect still
151    // emits a `close` event but since we never get another
152    // `open` we can't emit close
153    if (conn.db.serverConfig.autoReconnect) {
154      conn.readyState = STATES.disconnected;
155      conn.emit('close');
156      return;
157    }
158    conn.onClose();
159  });
160  conn.db.on('error', function(err) {
161    conn.emit('error', err);
162  });
163  conn.db.on('reconnect', function() {
164    conn.readyState = STATES.connected;
165    conn.emit('reconnected');
166  });
167  conn.db.on('timeout', function(err) {
168    var error = new Error(err && err.err || 'connection timeout');
169    conn.emit('error', error);
170  });
171  conn.db.on('open', function(err, db) {
172    if (STATES.disconnected === conn.readyState && db && db.databaseName) {
173      conn.readyState = STATES.connected;
174      conn.emit('reconnected');
175    }
176  });
177  conn.db.on('parseError', function(err) {
178    conn.emit('parseError', err);
179  });
180}
181
182/**
183 * Opens a connection to a MongoDB ReplicaSet.
184 *
185 * See description of [doOpen](#NativeConnection-doOpen) for server options. In this case `options.replset` is also passed to ReplSetServers.
186 *
187 * @param {Function} fn
188 * @api private
189 * @return {Connection} this
190 */
191
192NativeConnection.prototype.doOpenSet = function(fn) {
193  var servers = [],
194      _this = this;
195
196  this.hosts.forEach(function(server) {
197    var host = server.host || server.ipc;
198    var port = server.port || 27017;
199    servers.push(new Server(host, port, _this.options.server));
200  });
201
202  var server = this.options.mongos
203    ? new Mongos(servers, this.options.mongos)
204    : new ReplSetServers(servers, this.options.replset || this.options.replSet);
205  this.db = new Db(this.name, server, this.options.db);
206
207  this.db.on('fullsetup', function() {
208    _this.emit('fullsetup');
209  });
210
211  this.db.on('all', function() {
212    _this.emit('all');
213  });
214
215  this.db.open(function(err) {
216    if (err) return fn(err);
217    fn();
218    listen(_this);
219  });
220
221  return this;
222};
223
224/**
225 * Closes the connection
226 *
227 * @param {Function} fn
228 * @return {Connection} this
229 * @api private
230 */
231
232NativeConnection.prototype.doClose = function(fn) {
233  this.db.close(fn);
234  return this;
235};
236
237/**
238 * Prepares default connection options for the node-mongodb-native driver.
239 *
240 * _NOTE: `passed` options take precedence over connection string options._
241 *
242 * @param {Object} passed options that were passed directly during connection
243 * @param {Object} [connStrOptions] options that were passed in the connection string
244 * @api private
245 */
246
247NativeConnection.prototype.parseOptions = function(passed, connStrOpts) {
248  var o = passed || {};
249  o.db || (o.db = {});
250  o.auth || (o.auth = {});
251  o.server || (o.server = {});
252  o.replset || (o.replset = o.replSet) || (o.replset = {});
253  o.server.socketOptions || (o.server.socketOptions = {});
254  o.replset.socketOptions || (o.replset.socketOptions = {});
255  o.mongos || (o.mongos = (connStrOpts && connStrOpts.mongos));
256  (o.mongos === true) && (o.mongos = {});
257
258  var opts = connStrOpts || {};
259  Object.keys(opts).forEach(function(name) {
260    switch (name) {
261      case 'ssl':
262        o.server.ssl = opts.ssl;
263        o.replset.ssl = opts.ssl;
264        o.mongos && (o.mongos.ssl = opts.ssl);
265        break;
266      case 'poolSize':
267        if (typeof o.server[name] === 'undefined') {
268          o.server[name] = o.replset[name] = opts[name];
269        }
270        break;
271      case 'slaveOk':
272        if (typeof o.server.slave_ok === 'undefined') {
273          o.server.slave_ok = opts[name];
274        }
275        break;
276      case 'autoReconnect':
277        if (typeof o.server.auto_reconnect === 'undefined') {
278          o.server.auto_reconnect = opts[name];
279        }
280        break;
281      case 'socketTimeoutMS':
282      case 'connectTimeoutMS':
283        if (typeof o.server.socketOptions[name] === 'undefined') {
284          o.server.socketOptions[name] = o.replset.socketOptions[name] = opts[name];
285        }
286        break;
287      case 'authdb':
288        if (typeof o.auth.authdb === 'undefined') {
289          o.auth.authdb = opts[name];
290        }
291        break;
292      case 'authSource':
293        if (typeof o.auth.authSource === 'undefined') {
294          o.auth.authSource = opts[name];
295        }
296        break;
297      case 'retries':
298      case 'reconnectWait':
299      case 'rs_name':
300        if (typeof o.replset[name] === 'undefined') {
301          o.replset[name] = opts[name];
302        }
303        break;
304      case 'replicaSet':
305        if (typeof o.replset.rs_name === 'undefined') {
306          o.replset.rs_name = opts[name];
307        }
308        break;
309      case 'readSecondary':
310        if (typeof o.replset.read_secondary === 'undefined') {
311          o.replset.read_secondary = opts[name];
312        }
313        break;
314      case 'nativeParser':
315        if (typeof o.db.native_parser === 'undefined') {
316          o.db.native_parser = opts[name];
317        }
318        break;
319      case 'w':
320      case 'safe':
321      case 'fsync':
322      case 'journal':
323      case 'wtimeoutMS':
324        if (typeof o.db[name] === 'undefined') {
325          o.db[name] = opts[name];
326        }
327        break;
328      case 'readPreference':
329        if (typeof o.db.readPreference === 'undefined') {
330          o.db.readPreference = opts[name];
331        }
332        break;
333      case 'readPreferenceTags':
334        if (typeof o.db.read_preference_tags === 'undefined') {
335          o.db.read_preference_tags = opts[name];
336        }
337        break;
338      case 'sslValidate':
339        o.server.sslValidate = opts.sslValidate;
340        o.replset.sslValidate = opts.sslValidate;
341        o.mongos && (o.mongos.sslValidate = opts.sslValidate);
342    }
343  });
344
345  if (!('auto_reconnect' in o.server)) {
346    o.server.auto_reconnect = true;
347  }
348
349  // mongoose creates its own ObjectIds
350  o.db.forceServerObjectId = false;
351
352  // default safe using new nomenclature
353  if (!('journal' in o.db || 'j' in o.db ||
354        'fsync' in o.db || 'safe' in o.db || 'w' in o.db)) {
355    o.db.w = 1;
356  }
357
358  if (o.promiseLibrary) {
359    o.db.promiseLibrary = o.promiseLibrary;
360  }
361
362  validate(o);
363  return o;
364};
365
366/*!
367 * Validates the driver db options.
368 *
369 * @param {Object} o
370 */
371
372function validate(o) {
373  if (o.db.w === -1 || o.db.w === 0) {
374    if (o.db.journal || o.db.fsync || o.db.safe) {
375      throw new Error(
376          'Invalid writeConcern: '
377        + 'w set to -1 or 0 cannot be combined with safe|fsync|journal');
378    }
379  }
380}
381
382/*!
383 * Module exports.
384 */
385
386module.exports = NativeConnection;