1/*!
  2 * Module dependencies.
  3 */
  4
  5var MongooseConnection = require('../../connection')
  6  , mongo = require('mongodb')
  7  , Db = mongo.Db
  8  , Server = mongo.Server
  9  , Mongos = mongo.Mongos
 10  , STATES = require('../../connectionstate')
 11  , ReplSetServers = mongo.ReplSetServers;
 12
 13/**
 14 * A [node-mongodb-native](https://github.com/mongodb/node-mongodb-native) connection implementation.
 15 *
 16 * @inherits Connection
 17 * @api private
 18 */
 19
 20function NativeConnection() {
 21  MongooseConnection.apply(this, arguments);
 22  this._listening = false;
 23};
 24
 25/**
 26 * Expose the possible connection states.
 27 * @api public
 28 */
 29
 30NativeConnection.STATES = STATES;
 31
 32/*!
 33 * Inherits from Connection.
 34 */
 35
 36NativeConnection.prototype.__proto__ = MongooseConnection.prototype;
 37
 38/**
 39 * Opens the connection to MongoDB.
 40 *
 41 * @param {Function} fn
 42 * @return {Connection} this
 43 * @api private
 44 */
 45
 46NativeConnection.prototype.doOpen = function (fn) {
 47  if (this.db) {
 48    mute(this);
 49  }
 50
 51  var server = new Server(this.host, this.port, this.options.server);
 52  this.db = new Db(this.name, server, this.options.db);
 53
 54  var self = this;
 55  this.db.open(function (err) {
 56    if (err) return fn(err);
 57    listen(self);
 58    fn();
 59  });
 60
 61  return this;
 62};
 63
 64/*!
 65 * Register listeners for important events and bubble appropriately.
 66 */
 67
 68function listen (conn) {
 69  if (conn._listening) return;
 70  conn._listening = true;
 71
 72  conn.db.on('close', function(){
 73    if (conn._closeCalled) return;
 74
 75    // the driver never emits an `open` event. auto_reconnect still
 76    // emits a `close` event but since we never get another
 77    // `open` we can't emit close
 78    if (conn.db.serverConfig.autoReconnect) {
 79      conn.readyState = STATES.disconnected;
 80      conn.emit('close');
 81      return;
 82    }
 83    conn.onClose();
 84  });
 85  conn.db.on('error', function(err){
 86    conn.emit('error', err);
 87  });
 88  conn.db.on('timeout', function(err){
 89    var error = new Error(err && err.err || 'connection timeout');
 90    conn.emit('error', error);
 91  });
 92  conn.db.on('open', function (err, db) {
 93    if (STATES.disconnected === conn.readyState && db && db.databaseName) {
 94      conn.readyState = STATES.connected;
 95      conn.emit('reconnected')
 96    }
 97  })
 98}
 99
100/*!
101 * Remove listeners registered in `listen`
102 */
103
104function mute (conn) {
105  if (!conn.db) throw new Error('missing db');
106  conn.db.removeAllListeners("close");
107  conn.db.removeAllListeners("error");
108  conn.db.removeAllListeners("timeout");
109  conn.db.removeAllListeners("open");
110  conn.db.removeAllListeners("fullsetup");
111  conn._listening = false;
112}
113
114/**
115 * Opens a connection to a MongoDB ReplicaSet.
116 *
117 * See description of [doOpen](#NativeConnection-doOpen) for server options. In this case `options.replset` is also passed to ReplSetServers.
118 *
119 * @param {Function} fn
120 * @api private
121 * @return {Connection} this
122 */
123
124NativeConnection.prototype.doOpenSet = function (fn) {
125  if (this.db) {
126    mute(this);
127  }
128
129  var servers = []
130    , self = this;
131
132  this.hosts.forEach(function (server) {
133    var host = server.host || server.ipc;
134    var port = server.port || 27017;
135    servers.push(new Server(host, port, self.options.server));
136  })
137
138  var server = this.options.mongos
139    ? new Mongos(servers, this.options.mongos)
140    : new ReplSetServers(servers, this.options.replset);
141  this.db = new Db(this.name, server, this.options.db);
142
143  this.db.on('fullsetup', function () {
144    self.emit('fullsetup')
145  });
146
147  this.db.open(function (err) {
148    if (err) return fn(err);
149    fn();
150    listen(self);
151  });
152
153  return this;
154};
155
156/**
157 * Closes the connection
158 *
159 * @param {Function} fn
160 * @return {Connection} this
161 * @api private
162 */
163
164NativeConnection.prototype.doClose = function (fn) {
165  this.db.close();
166  if (fn) fn();
167  return this;
168}
169
170/**
171 * Prepares default connection options for the node-mongodb-native driver.
172 *
173 * _NOTE: `passed` options take precedence over connection string options._
174 *
175 * @param {Object} passed options that were passed directly during connection
176 * @param {Object} [connStrOptions] options that were passed in the connection string
177 * @api private
178 */
179
180NativeConnection.prototype.parseOptions = function (passed, connStrOpts) {
181  var o = passed || {};
182  o.db || (o.db = {});
183  o.auth || (o.auth = {});
184  o.server || (o.server = {});
185  o.replset || (o.replset = {});
186  o.server.socketOptions || (o.server.socketOptions = {});
187  o.replset.socketOptions || (o.replset.socketOptions = {});
188
189  var opts = connStrOpts || {};
190  Object.keys(opts).forEach(function (name) {
191    switch (name) {
192      case 'poolSize':
193        if ('undefined' == typeof o.server.poolSize) {
194          o.server.poolSize = o.replset.poolSize = opts[name];
195        }
196        break;
197      case 'slaveOk':
198        if ('undefined' == typeof o.server.slave_ok) {
199          o.server.slave_ok = opts[name];
200        }
201        break;
202      case 'autoReconnect':
203        if ('undefined' == typeof o.server.auto_reconnect) {
204          o.server.auto_reconnect = opts[name];
205        }
206        break;
207      case 'ssl':
208      case 'socketTimeoutMS':
209      case 'connectTimeoutMS':
210        if ('undefined' == typeof o.server.socketOptions[name]) {
211          o.server.socketOptions[name] = o.replset.socketOptions[name] = opts[name];
212        }
213        break;
214      case 'authdb':
215        if ('undefined' == typeof o.auth.authdb) {
216          o.auth.authdb = opts[name];
217        }
218        break;
219      case 'authSource':
220        if ('undefined' == typeof o.auth.authSource) {
221          o.auth.authSource = opts[name];
222        }
223        break;
224      case 'retries':
225      case 'reconnectWait':
226      case 'rs_name':
227        if ('undefined' == typeof o.replset[name]) {
228          o.replset[name] = opts[name];
229        }
230        break;
231      case 'replicaSet':
232        if ('undefined' == typeof o.replset.rs_name) {
233          o.replset.rs_name = opts[name];
234        }
235        break;
236      case 'readSecondary':
237        if ('undefined' == typeof o.replset.read_secondary) {
238          o.replset.read_secondary = opts[name];
239        }
240        break;
241      case 'nativeParser':
242        if ('undefined' == typeof o.db.native_parser) {
243          o.db.native_parser = opts[name];
244        }
245        break;
246      case 'w':
247      case 'safe':
248      case 'fsync':
249      case 'journal':
250      case 'wtimeoutMS':
251        if ('undefined' == typeof o.db[name]) {
252          o.db[name] = opts[name];
253        }
254        break;
255      case 'readPreference':
256        if ('undefined' == typeof o.db.read_preference) {
257          o.db.read_preference = opts[name];
258        }
259        break;
260      case 'readPreferenceTags':
261        if ('undefined' == typeof o.db.read_preference_tags) {
262          o.db.read_preference_tags = opts[name];
263        }
264        break;
265    }
266  })
267
268  if (!('auto_reconnect' in o.server)) {
269    o.server.auto_reconnect = true;
270  }
271
272  if (!o.db.read_preference) {
273    // read from primaries by default
274    o.db.read_preference = 'primary';
275  }
276
277  // mongoose creates its own ObjectIds
278  o.db.forceServerObjectId = false;
279
280  // default safe using new nomenclature
281  if (!('journal' in o.db || 'j' in o.db ||
282        'fsync' in o.db || 'safe' in o.db || 'w' in o.db)) {
283    o.db.w = 1;
284  }
285
286  validate(o);
287  return o;
288}
289
290/*!
291 * Validates the driver db options.
292 *
293 * @param {Object} o
294 */
295
296function validate (o) {
297  if (-1 === o.db.w || 0 === o.db.w) {
298    if (o.db.journal || o.db.fsync || o.db.safe) {
299      throw new Error(
300          'Invalid writeConcern: '
301        + 'w set to -1 or 0 cannot be combined with safe|fsync|journal');
302    }
303  }
304}
305
306/*!
307 * Module exports.
308 */
309
310module.exports = NativeConnection;