refactor(sdam): introduce a state machine for unified topology

The `Topology` type gains a state machine for tracking its current
connection state.

NODE-2280

Author: mbroadst

lib/core/sdam/topology.js

@@ -59,6 +59,33 @@ const LOCAL_SERVER_EVENTS = SERVER_RELAY_EVENTS.concat([
   'ended'
 ]);
 
+const STATE_CLOSING = 'closing';
+const STATE_CLOSED = 'closed';
+const STATE_CONNECTING = 'connecting';
+const STATE_CONNECTED = 'connected';
+
+function stateTransition(topology, newState) {
+  const legalTransitions = {
+    [STATE_CLOSED]: [STATE_CLOSED, STATE_CONNECTING],
+    [STATE_CONNECTING]: [STATE_CONNECTING, STATE_CLOSING, STATE_CONNECTED, STATE_CLOSED],
+    [STATE_CONNECTED]: [STATE_CONNECTED, STATE_CLOSING, STATE_CLOSED],
+    [STATE_CLOSING]: [STATE_CLOSING, STATE_CLOSED]
+  };
+
+  const legalStates = legalTransitions[topology.s.state];
+  if (legalStates && legalStates.indexOf(newState) < 0) {
+    console.log('throwing an error');
+    throw new MongoError(
+      `illegal state transition from [${
+        topology.s.state
+      }] => [${newState}], allowed: [${legalStates}]`
+    );
+  }
+
+  topology.emit('stateChanged', topology.s.state, newState);
+  topology.s.state = newState;
+}
+
 /**
  * A container of server instances representing a connection to a MongoDB topology.
  *
@@ -117,6 +144,8 @@ class Topology extends EventEmitter {
       options,
       // initial seedlist of servers to connect to
       seedlist: seedlist,
+      // initial state
+      state: STATE_CLOSED,
       // the topology description
       description: new TopologyDescription(
         topologyType,
@@ -229,6 +258,15 @@ class Topology extends EventEmitter {
   connect(options, callback) {
     if (typeof options === 'function') (callback = options), (options = {});
     options = options || {};
+    if (this.s.state === STATE_CONNECTED) {
+      if (typeof callback === 'function') {
+        callback();
+      }
+
+      return;
+    }
+
+    stateTransition(this, STATE_CONNECTING);
 
     // emit SDAM monitoring events
     this.emit('topologyOpening', new monitoring.TopologyOpeningEvent(this.s.id));
@@ -243,17 +281,15 @@ class Topology extends EventEmitter {
       )
     );
 
+    // connect all known servers, then attempt server selection to connect
     connectServers(this, Array.from(this.s.description.servers.values()));
-    this.s.connected = true;
-
-    // otherwise, wait for a server to properly connect based on user provided read preference,
-    // or primary.
 
     translateReadPreference(options);
     const readPreference = options.readPreference || ReadPreference.primary;
 
     this.selectServer(readPreferenceServerSelector(readPreference), options, (err, server) => {
       if (err) {
+        stateTransition(this, STATE_CLOSED);
         if (typeof callback === 'function') {
           callback(err, null);
         } else {
@@ -264,11 +300,13 @@ class Topology extends EventEmitter {
       }
 
       const errorHandler = err => {
+        stateTransition(this, STATE_CLOSED);
         server.removeListener('connect', connectHandler);
         if (typeof callback === 'function') callback(err, null);
       };
 
       const connectHandler = (_, err) => {
+        stateTransition(this, STATE_CONNECTED);
         server.removeListener('error', errorHandler);
         this.emit('open', err, this);
         this.emit('connect', this);
@@ -301,6 +339,13 @@ class Topology extends EventEmitter {
     }
 
     options = options || {};
+    if (this.s.state === STATE_CLOSED) {
+      if (typeof callback === 'function') {
+        callback();
+      }
+
+      return;
+    }
 
     // clear all existing monitor timers
     this.s.monitorTimers.map(timer => clearTimeout(timer));
@@ -327,9 +372,12 @@ class Topology extends EventEmitter {
       delete this.s.detectTopologyDescriptionChange;
     }
 
+    // defer state transition because we may need to send an `endSessions` command above
+    stateTransition(this, STATE_CLOSING);
+
     const servers = this.s.servers;
     if (servers.size === 0) {
-      this.s.connected = false;
+      stateTransition(this, STATE_CLOSED);
       if (typeof callback === 'function') {
         callback(null, null);
       }
@@ -346,7 +394,7 @@ class Topology extends EventEmitter {
           // emit an event for close
           this.emit('topologyClosed', new monitoring.TopologyClosedEvent(this.s.id));
 
-          this.s.connected = false;
+          stateTransition(this, STATE_CLOSED);
           if (typeof callback === 'function') {
             callback(null, null);
           }
@@ -807,7 +855,7 @@ function selectServers(topology, selector, timeout, start, callback) {
   }
 
   // ensure we are connected
-  if (!topology.s.connected) {
+  if (topology.s.state !== STATE_CONNECTED && topology.s.state !== STATE_CONNECTING) {
     topology.connect();
 
     // we want to make sure we're still within the requested timeout window

test/functional/core/topology_tests.js

@@ -0,0 +1,19 @@
+'use strict';
+const expect = require('chai').expect;
+
+describe('Topology', { requires: { unifiedTopology: true } }, function() {
+  it('should correctly track states of a topology', function(done) {
+    const topology = this.configuration.newTopology();
+
+    const states = [];
+    topology.on('stateChanged', (_, newState) => states.push(newState));
+    topology.connect(err => {
+      expect(err).to.not.exist;
+      topology.destroy(err => {
+        expect(err).to.not.exist;
+        expect(states).to.eql(['connecting', 'connected', 'closing', 'closed']);
+        done();
+      });
+    });
+  });
+});

test/runner/filters/unified_filter.js

@@ -0,0 +1,23 @@
+'use strict';
+
+/**
+ * Filter for tests that require the unified topology
+ *
+ * example:
+ * metadata: {
+ *    requires: {
+ *      unifiedTopology: <boolean>
+ *    }
+ * }
+ */
+class UnifiedTopologyFilter {
+  filter(test) {
+    if (!test.metadata) return true;
+    if (!test.metadata.requires) return true;
+    if (!test.metadata.requires.unifiedTopology) return true;
+
+    return !!process.env.MONGODB_UNIFIED_TOPOLOGY;
+  }
+}
+
+module.exports = UnifiedTopologyFilter;