replset.js 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489
  1. 'use strict';
  2. const Server = require('./server');
  3. const Cursor = require('../cursor');
  4. const MongoError = require('../core').MongoError;
  5. const TopologyBase = require('./topology_base').TopologyBase;
  6. const Store = require('./topology_base').Store;
  7. const CReplSet = require('../core').ReplSet;
  8. const MAX_JS_INT = require('../utils').MAX_JS_INT;
  9. const translateOptions = require('../utils').translateOptions;
  10. const filterOptions = require('../utils').filterOptions;
  11. const mergeOptions = require('../utils').mergeOptions;
  12. /**
  13. * @fileOverview The **ReplSet** class is a class that represents a Replicaset topology and is
  14. * used to construct connections.
  15. *
  16. * **ReplSet Should not be used, use MongoClient.connect**
  17. */
  18. // Allowed parameters
  19. var legalOptionNames = [
  20. 'ha',
  21. 'haInterval',
  22. 'replicaSet',
  23. 'rs_name',
  24. 'secondaryAcceptableLatencyMS',
  25. 'connectWithNoPrimary',
  26. 'poolSize',
  27. 'ssl',
  28. 'checkServerIdentity',
  29. 'sslValidate',
  30. 'sslCA',
  31. 'sslCert',
  32. 'ciphers',
  33. 'ecdhCurve',
  34. 'sslCRL',
  35. 'sslKey',
  36. 'sslPass',
  37. 'socketOptions',
  38. 'bufferMaxEntries',
  39. 'store',
  40. 'auto_reconnect',
  41. 'autoReconnect',
  42. 'emitError',
  43. 'keepAlive',
  44. 'keepAliveInitialDelay',
  45. 'noDelay',
  46. 'connectTimeoutMS',
  47. 'socketTimeoutMS',
  48. 'strategy',
  49. 'debug',
  50. 'family',
  51. 'loggerLevel',
  52. 'logger',
  53. 'reconnectTries',
  54. 'appname',
  55. 'domainsEnabled',
  56. 'servername',
  57. 'promoteLongs',
  58. 'promoteValues',
  59. 'promoteBuffers',
  60. 'maxStalenessSeconds',
  61. 'promiseLibrary',
  62. 'minSize',
  63. 'monitorCommands'
  64. ];
  65. /**
  66. * Creates a new ReplSet instance
  67. * @class
  68. * @deprecated
  69. * @param {Server[]} servers A seedlist of servers participating in the replicaset.
  70. * @param {object} [options] Optional settings.
  71. * @param {boolean} [options.ha=true] Turn on high availability monitoring.
  72. * @param {number} [options.haInterval=10000] Time between each replicaset status check.
  73. * @param {string} [options.replicaSet] The name of the replicaset to connect to.
  74. * @param {number} [options.secondaryAcceptableLatencyMS=15] Sets the range of servers to pick when using NEAREST (lowest ping ms + the latency fence, ex: range of 1 to (1 + 15) ms)
  75. * @param {boolean} [options.connectWithNoPrimary=false] Sets if the driver should connect even if no primary is available
  76. * @param {number} [options.poolSize=5] Number of connections in the connection pool for each server instance, set to 5 as default for legacy reasons.
  77. * @param {boolean} [options.ssl=false] Use ssl connection (needs to have a mongod server with ssl support)
  78. * @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.
  79. * @param {boolean} [options.sslValidate=false] Validate mongod server certificate against ca (needs to have a mongod server with ssl support, 2.4 or higher)
  80. * @param {array} [options.sslCA] Array of valid certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher)
  81. * @param {array} [options.sslCRL] Array of revocation certificates either as Buffers or Strings (needs to have a mongod server with ssl support, 2.4 or higher)
  82. * @param {(Buffer|string)} [options.sslCert] String or buffer containing the certificate we wish to present (needs to have a mongod server with ssl support, 2.4 or higher.
  83. * @param {string} [options.ciphers] Passed directly through to tls.createSecureContext. See https://nodejs.org/dist/latest-v9.x/docs/api/tls.html#tls_tls_createsecurecontext_options for more info.
  84. * @param {string} [options.ecdhCurve] Passed directly through to tls.createSecureContext. See https://nodejs.org/dist/latest-v9.x/docs/api/tls.html#tls_tls_createsecurecontext_options for more info.
  85. * @param {(Buffer|string)} [options.sslKey] String or buffer containing the certificate private key we wish to present (needs to have a mongod server with ssl support, 2.4 or higher)
  86. * @param {(Buffer|string)} [options.sslPass] String or buffer containing the certificate password (needs to have a mongod server with ssl support, 2.4 or higher)
  87. * @param {string} [options.servername] String containing the server name requested via TLS SNI.
  88. * @param {object} [options.socketOptions] Socket options
  89. * @param {boolean} [options.socketOptions.noDelay=true] TCP Socket NoDelay option.
  90. * @param {boolean} [options.socketOptions.keepAlive=true] TCP Connection keep alive enabled
  91. * @param {number} [options.socketOptions.keepAliveInitialDelay=30000] The number of milliseconds to wait before initiating keepAlive on the TCP socket
  92. * @param {number} [options.socketOptions.connectTimeoutMS=10000] How long to wait for a connection to be established before timing out
  93. * @param {number} [options.socketOptions.socketTimeoutMS=360000] How long a send or receive on a socket can take before timing out
  94. * @param {boolean} [options.domainsEnabled=false] Enable the wrapping of the callback in the current domain, disabled by default to avoid perf hit.
  95. * @param {number} [options.maxStalenessSeconds=undefined] The max staleness to secondary reads (values under 10 seconds cannot be guaranteed);
  96. * @param {boolean} [options.monitorCommands=false] Enable command monitoring for this topology
  97. * @fires ReplSet#connect
  98. * @fires ReplSet#ha
  99. * @fires ReplSet#joined
  100. * @fires ReplSet#left
  101. * @fires ReplSet#fullsetup
  102. * @fires ReplSet#open
  103. * @fires ReplSet#close
  104. * @fires ReplSet#error
  105. * @fires ReplSet#timeout
  106. * @fires ReplSet#parseError
  107. * @fires ReplSet#commandStarted
  108. * @fires ReplSet#commandSucceeded
  109. * @fires ReplSet#commandFailed
  110. * @property {string} parserType the parser type used (c++ or js).
  111. * @return {ReplSet} a ReplSet instance.
  112. */
  113. class ReplSet extends TopologyBase {
  114. constructor(servers, options) {
  115. super();
  116. options = options || {};
  117. var self = this;
  118. // Filter the options
  119. options = filterOptions(options, legalOptionNames);
  120. // Ensure all the instances are Server
  121. for (var i = 0; i < servers.length; i++) {
  122. if (!(servers[i] instanceof Server)) {
  123. throw MongoError.create({
  124. message: 'all seed list instances must be of the Server type',
  125. driver: true
  126. });
  127. }
  128. }
  129. // Stored options
  130. var storeOptions = {
  131. force: false,
  132. bufferMaxEntries:
  133. typeof options.bufferMaxEntries === 'number' ? options.bufferMaxEntries : MAX_JS_INT
  134. };
  135. // Shared global store
  136. var store = options.store || new Store(self, storeOptions);
  137. // Build seed list
  138. var seedlist = servers.map(function(x) {
  139. return { host: x.host, port: x.port };
  140. });
  141. // Clone options
  142. var clonedOptions = mergeOptions(
  143. {},
  144. {
  145. disconnectHandler: store,
  146. cursorFactory: Cursor,
  147. reconnect: false,
  148. emitError: typeof options.emitError === 'boolean' ? options.emitError : true,
  149. size: typeof options.poolSize === 'number' ? options.poolSize : 5,
  150. monitorCommands:
  151. typeof options.monitorCommands === 'boolean' ? options.monitorCommands : false
  152. }
  153. );
  154. // Translate any SSL options and other connectivity options
  155. clonedOptions = translateOptions(clonedOptions, options);
  156. // Socket options
  157. var socketOptions =
  158. options.socketOptions && Object.keys(options.socketOptions).length > 0
  159. ? options.socketOptions
  160. : options;
  161. // Translate all the options to the core types
  162. clonedOptions = translateOptions(clonedOptions, socketOptions);
  163. // Create the ReplSet
  164. var coreTopology = new CReplSet(seedlist, clonedOptions);
  165. // Listen to reconnect event
  166. coreTopology.on('reconnect', function() {
  167. self.emit('reconnect');
  168. store.execute();
  169. });
  170. // Internal state
  171. this.s = {
  172. // Replicaset
  173. coreTopology: coreTopology,
  174. // Server capabilities
  175. sCapabilities: null,
  176. // Debug tag
  177. tag: options.tag,
  178. // Store options
  179. storeOptions: storeOptions,
  180. // Cloned options
  181. clonedOptions: clonedOptions,
  182. // Store
  183. store: store,
  184. // Options
  185. options: options,
  186. // Server Session Pool
  187. sessionPool: null,
  188. // Active client sessions
  189. sessions: new Set(),
  190. // Promise library
  191. promiseLibrary: options.promiseLibrary || Promise
  192. };
  193. // Debug
  194. if (clonedOptions.debug) {
  195. // Last ismaster
  196. Object.defineProperty(this, 'replset', {
  197. enumerable: true,
  198. get: function() {
  199. return coreTopology;
  200. }
  201. });
  202. }
  203. }
  204. // Connect method
  205. connect(_options, callback) {
  206. var self = this;
  207. if ('function' === typeof _options) (callback = _options), (_options = {});
  208. if (_options == null) _options = {};
  209. if (!('function' === typeof callback)) callback = null;
  210. _options = Object.assign({}, this.s.clonedOptions, _options);
  211. self.s.options = _options;
  212. // Update bufferMaxEntries
  213. self.s.storeOptions.bufferMaxEntries =
  214. typeof _options.bufferMaxEntries === 'number' ? _options.bufferMaxEntries : -1;
  215. // Actual handler
  216. var errorHandler = function(event) {
  217. return function(err) {
  218. if (event !== 'error') {
  219. self.emit(event, err);
  220. }
  221. };
  222. };
  223. // Clear out all the current handlers left over
  224. var events = [
  225. 'timeout',
  226. 'error',
  227. 'close',
  228. 'serverOpening',
  229. 'serverDescriptionChanged',
  230. 'serverHeartbeatStarted',
  231. 'serverHeartbeatSucceeded',
  232. 'serverHeartbeatFailed',
  233. 'serverClosed',
  234. 'topologyOpening',
  235. 'topologyClosed',
  236. 'topologyDescriptionChanged',
  237. 'commandStarted',
  238. 'commandSucceeded',
  239. 'commandFailed',
  240. 'joined',
  241. 'left',
  242. 'ping',
  243. 'ha'
  244. ];
  245. events.forEach(function(e) {
  246. self.s.coreTopology.removeAllListeners(e);
  247. });
  248. // relay the event
  249. var relay = function(event) {
  250. return function(t, server) {
  251. self.emit(event, t, server);
  252. };
  253. };
  254. // Replset events relay
  255. var replsetRelay = function(event) {
  256. return function(t, server) {
  257. self.emit(event, t, server.lastIsMaster(), server);
  258. };
  259. };
  260. // Relay ha
  261. var relayHa = function(t, state) {
  262. self.emit('ha', t, state);
  263. if (t === 'start') {
  264. self.emit('ha_connect', t, state);
  265. } else if (t === 'end') {
  266. self.emit('ha_ismaster', t, state);
  267. }
  268. };
  269. // Set up serverConfig listeners
  270. self.s.coreTopology.on('joined', replsetRelay('joined'));
  271. self.s.coreTopology.on('left', relay('left'));
  272. self.s.coreTopology.on('ping', relay('ping'));
  273. self.s.coreTopology.on('ha', relayHa);
  274. // Set up SDAM listeners
  275. self.s.coreTopology.on('serverDescriptionChanged', relay('serverDescriptionChanged'));
  276. self.s.coreTopology.on('serverHeartbeatStarted', relay('serverHeartbeatStarted'));
  277. self.s.coreTopology.on('serverHeartbeatSucceeded', relay('serverHeartbeatSucceeded'));
  278. self.s.coreTopology.on('serverHeartbeatFailed', relay('serverHeartbeatFailed'));
  279. self.s.coreTopology.on('serverOpening', relay('serverOpening'));
  280. self.s.coreTopology.on('serverClosed', relay('serverClosed'));
  281. self.s.coreTopology.on('topologyOpening', relay('topologyOpening'));
  282. self.s.coreTopology.on('topologyClosed', relay('topologyClosed'));
  283. self.s.coreTopology.on('topologyDescriptionChanged', relay('topologyDescriptionChanged'));
  284. self.s.coreTopology.on('commandStarted', relay('commandStarted'));
  285. self.s.coreTopology.on('commandSucceeded', relay('commandSucceeded'));
  286. self.s.coreTopology.on('commandFailed', relay('commandFailed'));
  287. self.s.coreTopology.on('fullsetup', function() {
  288. self.emit('fullsetup', self, self);
  289. });
  290. self.s.coreTopology.on('all', function() {
  291. self.emit('all', null, self);
  292. });
  293. // Connect handler
  294. var connectHandler = function() {
  295. // Set up listeners
  296. self.s.coreTopology.once('timeout', errorHandler('timeout'));
  297. self.s.coreTopology.once('error', errorHandler('error'));
  298. self.s.coreTopology.once('close', errorHandler('close'));
  299. // Emit open event
  300. self.emit('open', null, self);
  301. // Return correctly
  302. try {
  303. callback(null, self);
  304. } catch (err) {
  305. process.nextTick(function() {
  306. throw err;
  307. });
  308. }
  309. };
  310. // Error handler
  311. var connectErrorHandler = function() {
  312. return function(err) {
  313. ['timeout', 'error', 'close'].forEach(function(e) {
  314. self.s.coreTopology.removeListener(e, connectErrorHandler);
  315. });
  316. self.s.coreTopology.removeListener('connect', connectErrorHandler);
  317. // Destroy the replset
  318. self.s.coreTopology.destroy();
  319. // Try to callback
  320. try {
  321. callback(err);
  322. } catch (err) {
  323. if (!self.s.coreTopology.isConnected())
  324. process.nextTick(function() {
  325. throw err;
  326. });
  327. }
  328. };
  329. };
  330. // Set up listeners
  331. self.s.coreTopology.once('timeout', connectErrorHandler('timeout'));
  332. self.s.coreTopology.once('error', connectErrorHandler('error'));
  333. self.s.coreTopology.once('close', connectErrorHandler('close'));
  334. self.s.coreTopology.once('connect', connectHandler);
  335. // Start connection
  336. self.s.coreTopology.connect(_options);
  337. }
  338. close(forceClosed, callback) {
  339. ['timeout', 'error', 'close', 'joined', 'left'].forEach(e => this.removeAllListeners(e));
  340. super.close(forceClosed, callback);
  341. }
  342. }
  343. Object.defineProperty(ReplSet.prototype, 'haInterval', {
  344. enumerable: true,
  345. get: function() {
  346. return this.s.coreTopology.s.haInterval;
  347. }
  348. });
  349. /**
  350. * A replset connect event, used to verify that the connection is up and running
  351. *
  352. * @event ReplSet#connect
  353. * @type {ReplSet}
  354. */
  355. /**
  356. * The replset high availability event
  357. *
  358. * @event ReplSet#ha
  359. * @type {function}
  360. * @param {string} type The stage in the high availability event (start|end)
  361. * @param {boolean} data.norepeat This is a repeating high availability process or a single execution only
  362. * @param {number} data.id The id for this high availability request
  363. * @param {object} data.state An object containing the information about the current replicaset
  364. */
  365. /**
  366. * A server member left the replicaset
  367. *
  368. * @event ReplSet#left
  369. * @type {function}
  370. * @param {string} type The type of member that left (primary|secondary|arbiter)
  371. * @param {Server} server The server object that left
  372. */
  373. /**
  374. * A server member joined the replicaset
  375. *
  376. * @event ReplSet#joined
  377. * @type {function}
  378. * @param {string} type The type of member that joined (primary|secondary|arbiter)
  379. * @param {Server} server The server object that joined
  380. */
  381. /**
  382. * ReplSet open event, emitted when replicaset can start processing commands.
  383. *
  384. * @event ReplSet#open
  385. * @type {Replset}
  386. */
  387. /**
  388. * ReplSet fullsetup event, emitted when all servers in the topology have been connected to.
  389. *
  390. * @event ReplSet#fullsetup
  391. * @type {Replset}
  392. */
  393. /**
  394. * ReplSet close event
  395. *
  396. * @event ReplSet#close
  397. * @type {object}
  398. */
  399. /**
  400. * ReplSet error event, emitted if there is an error listener.
  401. *
  402. * @event ReplSet#error
  403. * @type {MongoError}
  404. */
  405. /**
  406. * ReplSet timeout event
  407. *
  408. * @event ReplSet#timeout
  409. * @type {object}
  410. */
  411. /**
  412. * ReplSet parseError event
  413. *
  414. * @event ReplSet#parseError
  415. * @type {object}
  416. */
  417. /**
  418. * An event emitted indicating a command was started, if command monitoring is enabled
  419. *
  420. * @event ReplSet#commandStarted
  421. * @type {object}
  422. */
  423. /**
  424. * An event emitted indicating a command succeeded, if command monitoring is enabled
  425. *
  426. * @event ReplSet#commandSucceeded
  427. * @type {object}
  428. */
  429. /**
  430. * An event emitted indicating a command failed, if command monitoring is enabled
  431. *
  432. * @event ReplSet#commandFailed
  433. * @type {object}
  434. */
  435. module.exports = ReplSet;