wip

Author: nbbeeken

src/connection_string.ts

@@ -34,13 +34,7 @@ import { PromiseProvider } from './promise_provider';
 import { Encrypter } from './encrypter';
 import { Compressor, CompressorName } from './cmap/wire_protocol/compression';
 
-const VALID_TXT_RECORDS = [
-  'authSource',
-  'replicaSet',
-  'loadBalanced',
-  'srvMaxHosts',
-  'srvServiceName'
-];
+const VALID_TXT_RECORDS = ['authSource', 'replicaSet', 'loadBalanced', 'srvMaxHosts'];
 
 const LB_SINGLE_HOST_ERROR = 'loadBalanced option only supported with a single host in the URI';
 const LB_REPLICA_SET_ERROR = 'loadBalanced option not supported with a replicaSet option';
@@ -69,7 +63,10 @@ function matchesParentDomain(srvAddress: string, parentDomain: string): boolean
  * @param uri - The connection string to parse
  * @param options - Optional user provided connection string options
  */
-export function resolveSRVRecord(options: MongoOptions, callback: Callback<HostAddress[]>): void {
+export function resolveSRVRecord(
+  options: MongoOptions,
+  callback: Callback<{ hosts: HostAddress[]; records: dns.SrvRecord[] }>
+): void {
   if (typeof options.srvHost !== 'string') {
     return callback(new MongoAPIError('Option "srvHost" must not be empty'));
   }
@@ -130,7 +127,6 @@ export function resolveSRVRecord(options: MongoOptions, callback: Callback<HostA
         const replicaSet = txtRecordOptions.get('replicaSet') ?? undefined;
         const loadBalanced = txtRecordOptions.get('loadBalanced') ?? undefined;
         const srvMaxHostsString = txtRecordOptions.get('srvMaxHosts') ?? undefined;
-        // const srvServiceName = txtRecordOptions.get('srvServiceName') ?? undefined;
 
         if (srvMaxHostsString) {
           try {
@@ -166,7 +162,7 @@ export function resolveSRVRecord(options: MongoOptions, callback: Callback<HostA
         }
       }
 
-      callback(undefined, hostAddresses);
+      callback(undefined, { hosts: hostAddresses, records: addresses });
     });
   });
 }

src/operations/connect.ts

@@ -1,3 +1,4 @@
+import type * as dns from 'dns';
 import { MongoRuntimeError, MongoInvalidArgumentError } from '../error';
 import { Topology, TOPOLOGY_EVENTS } from '../sdam/topology';
 import { resolveSRVRecord } from '../connection_string';
@@ -49,8 +50,10 @@ export function connect(
   };
 
   if (typeof options.srvHost === 'string') {
-    return resolveSRVRecord(options, (err, hosts) => {
-      if (err || !hosts) return callback(err);
+    return resolveSRVRecord(options, (err, res) => {
+      if (err || !res) return callback(err);
+
+      const { hosts, records } = res;
 
       const selectedHosts =
         options.srvMaxHosts === 0 || options.srvMaxHosts >= hosts.length
@@ -61,7 +64,11 @@ export function connect(
         options.hosts[index] = host;
       }
 
-      return createTopology(mongoClient, options, connectCallback);
+      return createTopology(
+        mongoClient,
+        { ...options, initialSrvResults: records },
+        connectCallback
+      );
     });
   }
 
@@ -70,7 +77,7 @@ export function connect(
 
 function createTopology(
   mongoClient: MongoClient,
-  options: MongoOptions,
+  options: MongoOptions & { initialSrvResults?: dns.SrvRecord[] },
   callback: Callback<Topology>
 ) {
   // Create the topology

src/sdam/srv_polling.ts

@@ -32,6 +32,21 @@ export class SrvPollingEvent {
   hostnames(): Set<string> {
     return new Set(this.srvRecords.map(r => HostAddress.fromSrvRecord(r).toString()));
   }
+
+  equals(other: SrvPollingEvent): boolean {
+    return (
+      this.srvRecords.length !== 0 &&
+      other.srvRecords.length !== 0 &&
+      this.srvRecords.length === other.srvRecords.length &&
+      this.srvRecords.every(
+        (record, index) =>
+          record.name === other.srvRecords[index].name &&
+          record.port === other.srvRecords[index].port &&
+          record.weight === other.srvRecords[index].weight &&
+          record.priority === other.srvRecords[index].priority
+      )
+    );
+  }
 }
 
 /** @internal */
@@ -41,6 +56,8 @@ export interface SrvPollerOptions extends LoggerOptions {
   srvMaxHosts: number;
   srvHost: string;
   heartbeatFrequencyMS: number;
+
+  initialSrvResults?: dns.SrvRecord[];
 }
 
 /** @internal */
@@ -58,6 +75,7 @@ export class SrvPoller extends TypedEventEmitter<SrvPollerEvents> {
   generation: number;
   srvMaxHosts: number;
   srvServiceName: string;
+  lastSrvPollingEvent?: SrvPollingEvent;
   _timeout?: NodeJS.Timeout;
 
   /** @event */
@@ -75,6 +93,9 @@ export class SrvPoller extends TypedEventEmitter<SrvPollerEvents> {
     this.srvServiceName = options.srvServiceName ?? 'mongodb';
     this.rescanSrvIntervalMS = options.rescanSrvIntervalMS ?? 60000;
     this.heartbeatFrequencyMS = options.heartbeatFrequencyMS ?? 10000;
+    this.lastSrvPollingEvent = Array.isArray(options.initialSrvResults)
+      ? new SrvPollingEvent(options.initialSrvResults)
+      : undefined;
     this.logger = new Logger('srvPoller', options);
 
     this.haMode = false;
@@ -116,7 +137,11 @@ export class SrvPoller extends TypedEventEmitter<SrvPollerEvents> {
   success(srvRecords: dns.SrvRecord[]): void {
     this.haMode = false;
     this.schedule();
-    this.emit(SrvPoller.SRV_RECORD_DISCOVERY, new SrvPollingEvent(srvRecords));
+    const event = new SrvPollingEvent(srvRecords);
+    if (!this.lastSrvPollingEvent?.equals(event)) {
+      this.emit(SrvPoller.SRV_RECORD_DISCOVERY, event);
+    }
+    this.lastSrvPollingEvent = event;
   }
 
   failure(message: string, obj?: NodeJS.ErrnoException): void {

src/sdam/topology.ts

@@ -1,4 +1,5 @@
 import Denque = require('denque');
+import type * as dns from 'dns';
 import { ReadPreference, ReadPreferenceLike } from '../read_preference';
 import { compareTopologyVersion, ServerDescription } from './server_description';
 import { TopologyDescription } from './topology_description';
@@ -143,6 +144,7 @@ export interface TopologyOptions extends BSONSerializeOptions, ServerOptions {
   srvMaxHosts: number;
   srvServiceName: string;
   rescanSrvIntervalMS: number;
+  initialSrvResults?: dns.SrvRecord[];
   hosts: HostAddress[];
   retryWrites: boolean;
   retryReads: boolean;
@@ -345,7 +347,8 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
           srvHost: options.srvHost,
           srvMaxHosts: options.srvMaxHosts,
           srvServiceName: options.srvServiceName,
-          rescanSrvIntervalMS: options.rescanSrvIntervalMS
+          rescanSrvIntervalMS: options.rescanSrvIntervalMS,
+          initialSrvResults: options.initialSrvResults
         });
 
       this.on(Topology.TOPOLOGY_DESCRIPTION_CHANGED, this.s.detectShardedTopology);