topology.js 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694
  1. "use strict";
  2. Object.defineProperty(exports, "__esModule", { value: true });
  3. exports.ServerCapabilities = exports.Topology = void 0;
  4. const connection_string_1 = require("../connection_string");
  5. const constants_1 = require("../constants");
  6. const error_1 = require("../error");
  7. const mongo_logger_1 = require("../mongo_logger");
  8. const mongo_types_1 = require("../mongo_types");
  9. const read_preference_1 = require("../read_preference");
  10. const timeout_1 = require("../timeout");
  11. const utils_1 = require("../utils");
  12. const common_1 = require("./common");
  13. const events_1 = require("./events");
  14. const server_1 = require("./server");
  15. const server_description_1 = require("./server_description");
  16. const server_selection_1 = require("./server_selection");
  17. const server_selection_events_1 = require("./server_selection_events");
  18. const srv_polling_1 = require("./srv_polling");
  19. const topology_description_1 = require("./topology_description");
  20. // Global state
  21. let globalTopologyCounter = 0;
  22. const stateTransition = (0, utils_1.makeStateMachine)({
  23. [common_1.STATE_CLOSED]: [common_1.STATE_CLOSED, common_1.STATE_CONNECTING],
  24. [common_1.STATE_CONNECTING]: [common_1.STATE_CONNECTING, common_1.STATE_CLOSING, common_1.STATE_CONNECTED, common_1.STATE_CLOSED],
  25. [common_1.STATE_CONNECTED]: [common_1.STATE_CONNECTED, common_1.STATE_CLOSING, common_1.STATE_CLOSED],
  26. [common_1.STATE_CLOSING]: [common_1.STATE_CLOSING, common_1.STATE_CLOSED]
  27. });
  28. /**
  29. * A container of server instances representing a connection to a MongoDB topology.
  30. * @internal
  31. */
  32. class Topology extends mongo_types_1.TypedEventEmitter {
  33. /**
  34. * @param seedlist - a list of HostAddress instances to connect to
  35. */
  36. constructor(client, seeds, options) {
  37. super();
  38. this.on('error', utils_1.noop);
  39. this.client = client;
  40. // Options should only be undefined in tests, MongoClient will always have defined options
  41. options = options ?? {
  42. hosts: [utils_1.HostAddress.fromString('localhost:27017')],
  43. ...Object.fromEntries(connection_string_1.DEFAULT_OPTIONS.entries())
  44. };
  45. if (typeof seeds === 'string') {
  46. seeds = [utils_1.HostAddress.fromString(seeds)];
  47. }
  48. else if (!Array.isArray(seeds)) {
  49. seeds = [seeds];
  50. }
  51. const seedlist = [];
  52. for (const seed of seeds) {
  53. if (typeof seed === 'string') {
  54. seedlist.push(utils_1.HostAddress.fromString(seed));
  55. }
  56. else if (seed instanceof utils_1.HostAddress) {
  57. seedlist.push(seed);
  58. }
  59. else {
  60. // FIXME(NODE-3483): May need to be a MongoParseError
  61. throw new error_1.MongoRuntimeError(`Topology cannot be constructed from ${JSON.stringify(seed)}`);
  62. }
  63. }
  64. const topologyType = topologyTypeFromOptions(options);
  65. const topologyId = globalTopologyCounter++;
  66. const selectedHosts = options.srvMaxHosts == null ||
  67. options.srvMaxHosts === 0 ||
  68. options.srvMaxHosts >= seedlist.length
  69. ? seedlist
  70. : (0, utils_1.shuffle)(seedlist, options.srvMaxHosts);
  71. const serverDescriptions = new Map();
  72. for (const hostAddress of selectedHosts) {
  73. serverDescriptions.set(hostAddress.toString(), new server_description_1.ServerDescription(hostAddress));
  74. }
  75. this.waitQueue = new utils_1.List();
  76. this.s = {
  77. // the id of this topology
  78. id: topologyId,
  79. // passed in options
  80. options,
  81. // initial seedlist of servers to connect to
  82. seedlist,
  83. // initial state
  84. state: common_1.STATE_CLOSED,
  85. // the topology description
  86. description: new topology_description_1.TopologyDescription(topologyType, serverDescriptions, options.replicaSet, undefined, undefined, undefined, options),
  87. serverSelectionTimeoutMS: options.serverSelectionTimeoutMS,
  88. heartbeatFrequencyMS: options.heartbeatFrequencyMS,
  89. minHeartbeatFrequencyMS: options.minHeartbeatFrequencyMS,
  90. // a map of server instances to normalized addresses
  91. servers: new Map(),
  92. credentials: options?.credentials,
  93. clusterTime: undefined,
  94. detectShardedTopology: ev => this.detectShardedTopology(ev),
  95. detectSrvRecords: ev => this.detectSrvRecords(ev)
  96. };
  97. this.mongoLogger = client.mongoLogger;
  98. this.component = 'topology';
  99. if (options.srvHost && !options.loadBalanced) {
  100. this.s.srvPoller =
  101. options.srvPoller ??
  102. new srv_polling_1.SrvPoller({
  103. heartbeatFrequencyMS: this.s.heartbeatFrequencyMS,
  104. srvHost: options.srvHost,
  105. srvMaxHosts: options.srvMaxHosts,
  106. srvServiceName: options.srvServiceName
  107. });
  108. this.on(Topology.TOPOLOGY_DESCRIPTION_CHANGED, this.s.detectShardedTopology);
  109. }
  110. this.connectionLock = undefined;
  111. }
  112. detectShardedTopology(event) {
  113. const previousType = event.previousDescription.type;
  114. const newType = event.newDescription.type;
  115. const transitionToSharded = previousType !== common_1.TopologyType.Sharded && newType === common_1.TopologyType.Sharded;
  116. const srvListeners = this.s.srvPoller?.listeners(srv_polling_1.SrvPoller.SRV_RECORD_DISCOVERY);
  117. const listeningToSrvPolling = !!srvListeners?.includes(this.s.detectSrvRecords);
  118. if (transitionToSharded && !listeningToSrvPolling) {
  119. this.s.srvPoller?.on(srv_polling_1.SrvPoller.SRV_RECORD_DISCOVERY, this.s.detectSrvRecords);
  120. this.s.srvPoller?.start();
  121. }
  122. }
  123. detectSrvRecords(ev) {
  124. const previousTopologyDescription = this.s.description;
  125. this.s.description = this.s.description.updateFromSrvPollingEvent(ev, this.s.options.srvMaxHosts);
  126. if (this.s.description === previousTopologyDescription) {
  127. // Nothing changed, so return
  128. return;
  129. }
  130. updateServers(this);
  131. this.emitAndLog(Topology.TOPOLOGY_DESCRIPTION_CHANGED, new events_1.TopologyDescriptionChangedEvent(this.s.id, previousTopologyDescription, this.s.description));
  132. }
  133. /**
  134. * @returns A `TopologyDescription` for this topology
  135. */
  136. get description() {
  137. return this.s.description;
  138. }
  139. get loadBalanced() {
  140. return this.s.options.loadBalanced;
  141. }
  142. get serverApi() {
  143. return this.s.options.serverApi;
  144. }
  145. get capabilities() {
  146. return new ServerCapabilities(this.lastHello());
  147. }
  148. /** Initiate server connect */
  149. async connect(options) {
  150. this.connectionLock ??= this._connect(options);
  151. try {
  152. await this.connectionLock;
  153. return this;
  154. }
  155. finally {
  156. this.connectionLock = undefined;
  157. }
  158. }
  159. async _connect(options) {
  160. options = options ?? {};
  161. if (this.s.state === common_1.STATE_CONNECTED) {
  162. return this;
  163. }
  164. stateTransition(this, common_1.STATE_CONNECTING);
  165. // emit SDAM monitoring events
  166. this.emitAndLog(Topology.TOPOLOGY_OPENING, new events_1.TopologyOpeningEvent(this.s.id));
  167. // emit an event for the topology change
  168. this.emitAndLog(Topology.TOPOLOGY_DESCRIPTION_CHANGED, new events_1.TopologyDescriptionChangedEvent(this.s.id, new topology_description_1.TopologyDescription(common_1.TopologyType.Unknown), // initial is always Unknown
  169. this.s.description));
  170. // connect all known servers, then attempt server selection to connect
  171. const serverDescriptions = Array.from(this.s.description.servers.values());
  172. this.s.servers = new Map(serverDescriptions.map(serverDescription => [
  173. serverDescription.address,
  174. createAndConnectServer(this, serverDescription)
  175. ]));
  176. // In load balancer mode we need to fake a server description getting
  177. // emitted from the monitor, since the monitor doesn't exist.
  178. if (this.s.options.loadBalanced) {
  179. for (const description of serverDescriptions) {
  180. const newDescription = new server_description_1.ServerDescription(description.hostAddress, undefined, {
  181. loadBalanced: this.s.options.loadBalanced
  182. });
  183. this.serverUpdateHandler(newDescription);
  184. }
  185. }
  186. const serverSelectionTimeoutMS = this.client.s.options.serverSelectionTimeoutMS;
  187. const readPreference = options.readPreference ?? read_preference_1.ReadPreference.primary;
  188. const timeoutContext = timeout_1.TimeoutContext.create({
  189. // TODO(NODE-6448): auto-connect ignores timeoutMS; potential future feature
  190. timeoutMS: undefined,
  191. serverSelectionTimeoutMS,
  192. waitQueueTimeoutMS: this.client.s.options.waitQueueTimeoutMS
  193. });
  194. const selectServerOptions = {
  195. operationName: 'handshake',
  196. ...options,
  197. timeoutContext
  198. };
  199. try {
  200. const server = await this.selectServer((0, server_selection_1.readPreferenceServerSelector)(readPreference), selectServerOptions);
  201. const skipPingOnConnect = this.s.options.__skipPingOnConnect === true;
  202. if (!skipPingOnConnect && this.s.credentials) {
  203. const connection = await server.pool.checkOut({ timeoutContext: timeoutContext });
  204. server.pool.checkIn(connection);
  205. stateTransition(this, common_1.STATE_CONNECTED);
  206. this.emit(Topology.OPEN, this);
  207. this.emit(Topology.CONNECT, this);
  208. return this;
  209. }
  210. stateTransition(this, common_1.STATE_CONNECTED);
  211. this.emit(Topology.OPEN, this);
  212. this.emit(Topology.CONNECT, this);
  213. return this;
  214. }
  215. catch (error) {
  216. this.close();
  217. throw error;
  218. }
  219. }
  220. closeCheckedOutConnections() {
  221. for (const server of this.s.servers.values()) {
  222. return server.closeCheckedOutConnections();
  223. }
  224. }
  225. /** Close this topology */
  226. close() {
  227. if (this.s.state === common_1.STATE_CLOSED || this.s.state === common_1.STATE_CLOSING) {
  228. return;
  229. }
  230. for (const server of this.s.servers.values()) {
  231. closeServer(server, this);
  232. }
  233. this.s.servers.clear();
  234. stateTransition(this, common_1.STATE_CLOSING);
  235. drainWaitQueue(this.waitQueue, new error_1.MongoTopologyClosedError());
  236. if (this.s.srvPoller) {
  237. this.s.srvPoller.stop();
  238. this.s.srvPoller.removeListener(srv_polling_1.SrvPoller.SRV_RECORD_DISCOVERY, this.s.detectSrvRecords);
  239. }
  240. this.removeListener(Topology.TOPOLOGY_DESCRIPTION_CHANGED, this.s.detectShardedTopology);
  241. stateTransition(this, common_1.STATE_CLOSED);
  242. // emit an event for close
  243. this.emitAndLog(Topology.TOPOLOGY_CLOSED, new events_1.TopologyClosedEvent(this.s.id));
  244. }
  245. /**
  246. * Selects a server according to the selection predicate provided
  247. *
  248. * @param selector - An optional selector to select servers by, defaults to a random selection within a latency window
  249. * @param options - Optional settings related to server selection
  250. * @param callback - The callback used to indicate success or failure
  251. * @returns An instance of a `Server` meeting the criteria of the predicate provided
  252. */
  253. async selectServer(selector, options) {
  254. let serverSelector;
  255. if (typeof selector !== 'function') {
  256. if (typeof selector === 'string') {
  257. serverSelector = (0, server_selection_1.readPreferenceServerSelector)(read_preference_1.ReadPreference.fromString(selector));
  258. }
  259. else {
  260. let readPreference;
  261. if (selector instanceof read_preference_1.ReadPreference) {
  262. readPreference = selector;
  263. }
  264. else {
  265. read_preference_1.ReadPreference.translate(options);
  266. readPreference = options.readPreference || read_preference_1.ReadPreference.primary;
  267. }
  268. serverSelector = (0, server_selection_1.readPreferenceServerSelector)(readPreference);
  269. }
  270. }
  271. else {
  272. serverSelector = selector;
  273. }
  274. options = { serverSelectionTimeoutMS: this.s.serverSelectionTimeoutMS, ...options };
  275. if (this.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  276. this.client.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionStartedEvent(selector, this.description, options.operationName));
  277. }
  278. let timeout;
  279. if (options.timeoutContext)
  280. timeout = options.timeoutContext.serverSelectionTimeout;
  281. else {
  282. timeout = timeout_1.Timeout.expires(options.serverSelectionTimeoutMS ?? 0);
  283. }
  284. const isSharded = this.description.type === common_1.TopologyType.Sharded;
  285. const session = options.session;
  286. const transaction = session && session.transaction;
  287. if (isSharded && transaction && transaction.server) {
  288. if (this.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  289. this.client.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionSucceededEvent(selector, this.description, transaction.server.pool.address, options.operationName));
  290. }
  291. if (options.timeoutContext?.clearServerSelectionTimeout)
  292. timeout?.clear();
  293. return transaction.server;
  294. }
  295. const { promise: serverPromise, resolve, reject } = (0, utils_1.promiseWithResolvers)();
  296. const waitQueueMember = {
  297. serverSelector,
  298. topologyDescription: this.description,
  299. mongoLogger: this.client.mongoLogger,
  300. transaction,
  301. resolve,
  302. reject,
  303. cancelled: false,
  304. startTime: (0, utils_1.now)(),
  305. operationName: options.operationName,
  306. waitingLogged: false,
  307. previousServer: options.previousServer
  308. };
  309. const abortListener = (0, utils_1.addAbortListener)(options.signal, function () {
  310. waitQueueMember.cancelled = true;
  311. reject(this.reason);
  312. });
  313. this.waitQueue.push(waitQueueMember);
  314. processWaitQueue(this);
  315. try {
  316. timeout?.throwIfExpired();
  317. const server = await (timeout ? Promise.race([serverPromise, timeout]) : serverPromise);
  318. if (options.timeoutContext?.csotEnabled() && server.description.minRoundTripTime !== 0) {
  319. options.timeoutContext.minRoundTripTime = server.description.minRoundTripTime;
  320. }
  321. return server;
  322. }
  323. catch (error) {
  324. if (timeout_1.TimeoutError.is(error)) {
  325. // Timeout
  326. waitQueueMember.cancelled = true;
  327. const timeoutError = new error_1.MongoServerSelectionError(`Server selection timed out after ${timeout?.duration} ms`, this.description);
  328. if (this.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  329. this.client.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionFailedEvent(selector, this.description, timeoutError, options.operationName));
  330. }
  331. if (options.timeoutContext?.csotEnabled()) {
  332. throw new error_1.MongoOperationTimeoutError('Timed out during server selection', {
  333. cause: timeoutError
  334. });
  335. }
  336. throw timeoutError;
  337. }
  338. // Other server selection error
  339. throw error;
  340. }
  341. finally {
  342. abortListener?.[utils_1.kDispose]();
  343. if (options.timeoutContext?.clearServerSelectionTimeout)
  344. timeout?.clear();
  345. }
  346. }
  347. /**
  348. * Update the internal TopologyDescription with a ServerDescription
  349. *
  350. * @param serverDescription - The server to update in the internal list of server descriptions
  351. */
  352. serverUpdateHandler(serverDescription) {
  353. if (!this.s.description.hasServer(serverDescription.address)) {
  354. return;
  355. }
  356. // ignore this server update if its from an outdated topologyVersion
  357. if (isStaleServerDescription(this.s.description, serverDescription)) {
  358. return;
  359. }
  360. // these will be used for monitoring events later
  361. const previousTopologyDescription = this.s.description;
  362. const previousServerDescription = this.s.description.servers.get(serverDescription.address);
  363. if (!previousServerDescription) {
  364. return;
  365. }
  366. // Driver Sessions Spec: "Whenever a driver receives a cluster time from
  367. // a server it MUST compare it to the current highest seen cluster time
  368. // for the deployment. If the new cluster time is higher than the
  369. // highest seen cluster time it MUST become the new highest seen cluster
  370. // time. Two cluster times are compared using only the BsonTimestamp
  371. // value of the clusterTime embedded field."
  372. const clusterTime = serverDescription.$clusterTime;
  373. if (clusterTime) {
  374. (0, common_1._advanceClusterTime)(this, clusterTime);
  375. }
  376. // If we already know all the information contained in this updated description, then
  377. // we don't need to emit SDAM events, but still need to update the description, in order
  378. // to keep client-tracked attributes like last update time and round trip time up to date
  379. const equalDescriptions = previousServerDescription && previousServerDescription.equals(serverDescription);
  380. // first update the TopologyDescription
  381. this.s.description = this.s.description.update(serverDescription);
  382. if (this.s.description.compatibilityError) {
  383. this.emit(Topology.ERROR, new error_1.MongoCompatibilityError(this.s.description.compatibilityError));
  384. return;
  385. }
  386. // emit monitoring events for this change
  387. if (!equalDescriptions) {
  388. const newDescription = this.s.description.servers.get(serverDescription.address);
  389. if (newDescription) {
  390. this.emit(Topology.SERVER_DESCRIPTION_CHANGED, new events_1.ServerDescriptionChangedEvent(this.s.id, serverDescription.address, previousServerDescription, newDescription));
  391. }
  392. }
  393. // update server list from updated descriptions
  394. updateServers(this, serverDescription);
  395. // attempt to resolve any outstanding server selection attempts
  396. if (this.waitQueue.length > 0) {
  397. processWaitQueue(this);
  398. }
  399. if (!equalDescriptions) {
  400. this.emitAndLog(Topology.TOPOLOGY_DESCRIPTION_CHANGED, new events_1.TopologyDescriptionChangedEvent(this.s.id, previousTopologyDescription, this.s.description));
  401. }
  402. }
  403. auth(credentials, callback) {
  404. if (typeof credentials === 'function')
  405. ((callback = credentials), (credentials = undefined));
  406. if (typeof callback === 'function')
  407. callback(undefined, true);
  408. }
  409. get clientMetadata() {
  410. return this.s.options.metadata;
  411. }
  412. isConnected() {
  413. return this.s.state === common_1.STATE_CONNECTED;
  414. }
  415. isDestroyed() {
  416. return this.s.state === common_1.STATE_CLOSED;
  417. }
  418. // NOTE: There are many places in code where we explicitly check the last hello
  419. // to do feature support detection. This should be done any other way, but for
  420. // now we will just return the first hello seen, which should suffice.
  421. lastHello() {
  422. const serverDescriptions = Array.from(this.description.servers.values());
  423. if (serverDescriptions.length === 0)
  424. return {};
  425. const sd = serverDescriptions.filter((sd) => sd.type !== common_1.ServerType.Unknown)[0];
  426. const result = sd || { maxWireVersion: this.description.commonWireVersion };
  427. return result;
  428. }
  429. get commonWireVersion() {
  430. return this.description.commonWireVersion;
  431. }
  432. get logicalSessionTimeoutMinutes() {
  433. return this.description.logicalSessionTimeoutMinutes;
  434. }
  435. get clusterTime() {
  436. return this.s.clusterTime;
  437. }
  438. set clusterTime(clusterTime) {
  439. this.s.clusterTime = clusterTime;
  440. }
  441. }
  442. exports.Topology = Topology;
  443. /** @event */
  444. Topology.SERVER_OPENING = constants_1.SERVER_OPENING;
  445. /** @event */
  446. Topology.SERVER_CLOSED = constants_1.SERVER_CLOSED;
  447. /** @event */
  448. Topology.SERVER_DESCRIPTION_CHANGED = constants_1.SERVER_DESCRIPTION_CHANGED;
  449. /** @event */
  450. Topology.TOPOLOGY_OPENING = constants_1.TOPOLOGY_OPENING;
  451. /** @event */
  452. Topology.TOPOLOGY_CLOSED = constants_1.TOPOLOGY_CLOSED;
  453. /** @event */
  454. Topology.TOPOLOGY_DESCRIPTION_CHANGED = constants_1.TOPOLOGY_DESCRIPTION_CHANGED;
  455. /** @event */
  456. Topology.ERROR = constants_1.ERROR;
  457. /** @event */
  458. Topology.OPEN = constants_1.OPEN;
  459. /** @event */
  460. Topology.CONNECT = constants_1.CONNECT;
  461. /** @event */
  462. Topology.CLOSE = constants_1.CLOSE;
  463. /** @event */
  464. Topology.TIMEOUT = constants_1.TIMEOUT;
  465. /** Destroys a server, and removes all event listeners from the instance */
  466. function closeServer(server, topology) {
  467. for (const event of constants_1.LOCAL_SERVER_EVENTS) {
  468. server.removeAllListeners(event);
  469. }
  470. server.close();
  471. topology.emitAndLog(Topology.SERVER_CLOSED, new events_1.ServerClosedEvent(topology.s.id, server.description.address));
  472. for (const event of constants_1.SERVER_RELAY_EVENTS) {
  473. server.removeAllListeners(event);
  474. }
  475. }
  476. /** Predicts the TopologyType from options */
  477. function topologyTypeFromOptions(options) {
  478. if (options?.directConnection) {
  479. return common_1.TopologyType.Single;
  480. }
  481. if (options?.replicaSet) {
  482. return common_1.TopologyType.ReplicaSetNoPrimary;
  483. }
  484. if (options?.loadBalanced) {
  485. return common_1.TopologyType.LoadBalanced;
  486. }
  487. return common_1.TopologyType.Unknown;
  488. }
  489. /**
  490. * Creates new server instances and attempts to connect them
  491. *
  492. * @param topology - The topology that this server belongs to
  493. * @param serverDescription - The description for the server to initialize and connect to
  494. */
  495. function createAndConnectServer(topology, serverDescription) {
  496. topology.emitAndLog(Topology.SERVER_OPENING, new events_1.ServerOpeningEvent(topology.s.id, serverDescription.address));
  497. const server = new server_1.Server(topology, serverDescription, topology.s.options);
  498. for (const event of constants_1.SERVER_RELAY_EVENTS) {
  499. server.on(event, (e) => topology.emit(event, e));
  500. }
  501. server.on(server_1.Server.DESCRIPTION_RECEIVED, description => topology.serverUpdateHandler(description));
  502. server.connect();
  503. return server;
  504. }
  505. /**
  506. * @param topology - Topology to update.
  507. * @param incomingServerDescription - New server description.
  508. */
  509. function updateServers(topology, incomingServerDescription) {
  510. // update the internal server's description
  511. if (incomingServerDescription && topology.s.servers.has(incomingServerDescription.address)) {
  512. const server = topology.s.servers.get(incomingServerDescription.address);
  513. if (server) {
  514. server.s.description = incomingServerDescription;
  515. if (incomingServerDescription.error instanceof error_1.MongoError &&
  516. incomingServerDescription.error.hasErrorLabel(error_1.MongoErrorLabel.ResetPool)) {
  517. const interruptInUseConnections = incomingServerDescription.error.hasErrorLabel(error_1.MongoErrorLabel.InterruptInUseConnections);
  518. server.pool.clear({ interruptInUseConnections });
  519. }
  520. else if (incomingServerDescription.error == null) {
  521. const newTopologyType = topology.s.description.type;
  522. const shouldMarkPoolReady = incomingServerDescription.isDataBearing ||
  523. (incomingServerDescription.type !== common_1.ServerType.Unknown &&
  524. newTopologyType === common_1.TopologyType.Single);
  525. if (shouldMarkPoolReady) {
  526. server.pool.ready();
  527. }
  528. }
  529. }
  530. }
  531. // add new servers for all descriptions we currently don't know about locally
  532. for (const serverDescription of topology.description.servers.values()) {
  533. if (!topology.s.servers.has(serverDescription.address)) {
  534. const server = createAndConnectServer(topology, serverDescription);
  535. topology.s.servers.set(serverDescription.address, server);
  536. }
  537. }
  538. // for all servers no longer known, remove their descriptions and destroy their instances
  539. for (const entry of topology.s.servers) {
  540. const serverAddress = entry[0];
  541. if (topology.description.hasServer(serverAddress)) {
  542. continue;
  543. }
  544. if (!topology.s.servers.has(serverAddress)) {
  545. continue;
  546. }
  547. const server = topology.s.servers.get(serverAddress);
  548. topology.s.servers.delete(serverAddress);
  549. // prepare server for garbage collection
  550. if (server) {
  551. closeServer(server, topology);
  552. }
  553. }
  554. }
  555. function drainWaitQueue(queue, drainError) {
  556. while (queue.length) {
  557. const waitQueueMember = queue.shift();
  558. if (!waitQueueMember) {
  559. continue;
  560. }
  561. if (!waitQueueMember.cancelled) {
  562. if (waitQueueMember.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  563. waitQueueMember.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionFailedEvent(waitQueueMember.serverSelector, waitQueueMember.topologyDescription, drainError, waitQueueMember.operationName));
  564. }
  565. waitQueueMember.reject(drainError);
  566. }
  567. }
  568. }
  569. function processWaitQueue(topology) {
  570. if (topology.s.state === common_1.STATE_CLOSED) {
  571. drainWaitQueue(topology.waitQueue, new error_1.MongoTopologyClosedError());
  572. return;
  573. }
  574. const isSharded = topology.description.type === common_1.TopologyType.Sharded;
  575. const serverDescriptions = Array.from(topology.description.servers.values());
  576. const membersToProcess = topology.waitQueue.length;
  577. for (let i = 0; i < membersToProcess; ++i) {
  578. const waitQueueMember = topology.waitQueue.shift();
  579. if (!waitQueueMember) {
  580. continue;
  581. }
  582. if (waitQueueMember.cancelled) {
  583. continue;
  584. }
  585. let selectedDescriptions;
  586. try {
  587. const serverSelector = waitQueueMember.serverSelector;
  588. const previousServer = waitQueueMember.previousServer;
  589. selectedDescriptions = serverSelector
  590. ? serverSelector(topology.description, serverDescriptions, previousServer ? [previousServer] : [])
  591. : serverDescriptions;
  592. }
  593. catch (selectorError) {
  594. if (topology.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  595. topology.client.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionFailedEvent(waitQueueMember.serverSelector, topology.description, selectorError, waitQueueMember.operationName));
  596. }
  597. waitQueueMember.reject(selectorError);
  598. continue;
  599. }
  600. let selectedServer;
  601. if (selectedDescriptions.length === 0) {
  602. if (!waitQueueMember.waitingLogged) {
  603. if (topology.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.INFORMATIONAL)) {
  604. topology.client.mongoLogger?.info(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.WaitingForSuitableServerEvent(waitQueueMember.serverSelector, topology.description, topology.s.serverSelectionTimeoutMS !== 0
  605. ? topology.s.serverSelectionTimeoutMS - ((0, utils_1.now)() - waitQueueMember.startTime)
  606. : -1, waitQueueMember.operationName));
  607. }
  608. waitQueueMember.waitingLogged = true;
  609. }
  610. topology.waitQueue.push(waitQueueMember);
  611. continue;
  612. }
  613. else if (selectedDescriptions.length === 1) {
  614. selectedServer = topology.s.servers.get(selectedDescriptions[0].address);
  615. }
  616. else {
  617. const descriptions = (0, utils_1.shuffle)(selectedDescriptions, 2);
  618. const server1 = topology.s.servers.get(descriptions[0].address);
  619. const server2 = topology.s.servers.get(descriptions[1].address);
  620. selectedServer =
  621. server1 && server2 && server1.s.operationCount < server2.s.operationCount
  622. ? server1
  623. : server2;
  624. }
  625. if (!selectedServer) {
  626. const serverSelectionError = new error_1.MongoServerSelectionError('server selection returned a server description but the server was not found in the topology', topology.description);
  627. if (topology.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  628. topology.client.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionFailedEvent(waitQueueMember.serverSelector, topology.description, serverSelectionError, waitQueueMember.operationName));
  629. }
  630. waitQueueMember.reject(serverSelectionError);
  631. return;
  632. }
  633. const transaction = waitQueueMember.transaction;
  634. if (isSharded && transaction && transaction.isActive && selectedServer) {
  635. transaction.pinServer(selectedServer);
  636. }
  637. if (topology.client.mongoLogger?.willLog(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, mongo_logger_1.SeverityLevel.DEBUG)) {
  638. topology.client.mongoLogger?.debug(mongo_logger_1.MongoLoggableComponent.SERVER_SELECTION, new server_selection_events_1.ServerSelectionSucceededEvent(waitQueueMember.serverSelector, waitQueueMember.topologyDescription, selectedServer.pool.address, waitQueueMember.operationName));
  639. }
  640. waitQueueMember.resolve(selectedServer);
  641. }
  642. if (topology.waitQueue.length > 0) {
  643. // ensure all server monitors attempt monitoring soon
  644. for (const [, server] of topology.s.servers) {
  645. process.nextTick(function scheduleServerCheck() {
  646. return server.requestCheck();
  647. });
  648. }
  649. }
  650. }
  651. function isStaleServerDescription(topologyDescription, incomingServerDescription) {
  652. const currentServerDescription = topologyDescription.servers.get(incomingServerDescription.address);
  653. const currentTopologyVersion = currentServerDescription?.topologyVersion;
  654. return ((0, server_description_1.compareTopologyVersion)(currentTopologyVersion, incomingServerDescription.topologyVersion) > 0);
  655. }
  656. /**
  657. * @public
  658. * @deprecated This class will be removed as dead code in the next major version.
  659. */
  660. class ServerCapabilities {
  661. constructor(hello) {
  662. this.minWireVersion = hello.minWireVersion || 0;
  663. this.maxWireVersion = hello.maxWireVersion || 0;
  664. }
  665. get hasAggregationCursor() {
  666. return true;
  667. }
  668. get hasWriteCommands() {
  669. return true;
  670. }
  671. get hasTextSearch() {
  672. return true;
  673. }
  674. get hasAuthCommands() {
  675. return true;
  676. }
  677. get hasListCollectionsCommand() {
  678. return true;
  679. }
  680. get hasListIndexesCommand() {
  681. return true;
  682. }
  683. get supportsSnapshotReads() {
  684. return this.maxWireVersion >= 13;
  685. }
  686. get commandsTakeWriteConcern() {
  687. return true;
  688. }
  689. get commandsTakeCollation() {
  690. return true;
  691. }
  692. }
  693. exports.ServerCapabilities = ServerCapabilities;
  694. //# sourceMappingURL=topology.js.map