Implement the serverMonitoringMode logic

This change is in accordance with source/server-discovery-and-monitoring/server-monitoring.rst. JAVA-4936
mongodb · Feb 21, 2024 · 78f0a48 · 78f0a48
1 parent 86064ae
commit 78f0a48
Show file tree

Hide file tree

Showing 10 changed files with 505 additions and 9 deletions.
diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java
@@ -110,7 +110,7 @@ public Cluster createCluster(final ClusterSettings originalClusterSettings, fina
                     connectionPoolSettings, internalConnectionPoolSettings,
                     streamFactory, heartbeatStreamFactory, credential, loggerSettings, commandListener, applicationName,
                     mongoDriverInformation != null ? mongoDriverInformation : MongoDriverInformation.builder().build(), compressorList,
-                    serverApi);
+                    serverApi, FaasEnvironment.getFaasEnvironment() != FaasEnvironment.UNKNOWN);
 
             if (clusterSettings.getMode() == ClusterConnectionMode.SINGLE) {
                 return new SingleServerCluster(clusterId, clusterSettings, serverFactory);

diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterableServerFactory.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterableServerFactory.java
@@ -53,6 +53,7 @@ public class DefaultClusterableServerFactory implements ClusterableServerFactory
     private final List<MongoCompressor> compressorList;
     @Nullable
     private final ServerApi serverApi;
+    private final boolean faas;
 
     public DefaultClusterableServerFactory(
             final ServerSettings serverSettings, final ConnectionPoolSettings connectionPoolSettings,
@@ -62,7 +63,7 @@ public DefaultClusterableServerFactory(
             final LoggerSettings loggerSettings,
             @Nullable final CommandListener commandListener,
             @Nullable final String applicationName, @Nullable final MongoDriverInformation mongoDriverInformation,
-            final List<MongoCompressor> compressorList, @Nullable final ServerApi serverApi) {
+            final List<MongoCompressor> compressorList, @Nullable final ServerApi serverApi, final boolean faas) {
         this.serverSettings = serverSettings;
         this.connectionPoolSettings = connectionPoolSettings;
         this.internalConnectionPoolSettings = internalConnectionPoolSettings;
@@ -75,6 +76,7 @@ public DefaultClusterableServerFactory(
         this.mongoDriverInformation = mongoDriverInformation;
         this.compressorList = compressorList;
         this.serverApi = serverApi;
+        this.faas = faas;
     }
 
     @Override
@@ -86,7 +88,7 @@ public ClusterableServer create(final Cluster cluster, final ServerAddress serve
                 // no credentials, compressor list, or command listener for the server monitor factory
                 new InternalStreamConnectionFactory(clusterMode, true, heartbeatStreamFactory, null, applicationName,
                         mongoDriverInformation, emptyList(), loggerSettings, null, serverApi),
-                clusterMode, serverApi, sdamProvider);
+                clusterMode, serverApi, faas, sdamProvider);
         ConnectionPool connectionPool = new DefaultConnectionPool(serverId,
                 new InternalStreamConnectionFactory(clusterMode, streamFactory, credential, applicationName,
                         mongoDriverInformation, compressorList, loggerSettings, commandListener, serverApi),

diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultServerMonitor.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultServerMonitor.java
@@ -50,6 +50,7 @@
 import static com.mongodb.MongoNamespace.COMMAND_COLLECTION_NAME;
 import static com.mongodb.ReadPreference.primary;
 import static com.mongodb.assertions.Assertions.assertNotNull;
+import static com.mongodb.assertions.Assertions.fail;
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.connection.ServerType.UNKNOWN;
 import static com.mongodb.internal.Locks.checkedWithLock;
@@ -76,6 +77,7 @@ class DefaultServerMonitor implements ServerMonitor {
     private final ClusterConnectionMode clusterConnectionMode;
     @Nullable
     private final ServerApi serverApi;
+    private final boolean faas;
     private final ServerSettings serverSettings;
     private final ServerMonitorRunnable monitor;
     private final Thread monitorThread;
@@ -90,13 +92,15 @@ class DefaultServerMonitor implements ServerMonitor {
             final InternalConnectionFactory internalConnectionFactory,
                          final ClusterConnectionMode clusterConnectionMode,
                          @Nullable final ServerApi serverApi,
+                         final boolean faas,
                          final Provider<SdamServerDescriptionManager> sdamProvider) {
         this.serverSettings = notNull("serverSettings", serverSettings);
         this.serverId = notNull("serverId", serverId);
         this.serverMonitorListener = singleServerMonitorListener(serverSettings);
         this.internalConnectionFactory = notNull("internalConnectionFactory", internalConnectionFactory);
         this.clusterConnectionMode = notNull("clusterConnectionMode", clusterConnectionMode);
         this.serverApi = serverApi;
+        this.faas = faas;
         this.sdamProvider = sdamProvider;
         monitor = new ServerMonitorRunnable();
         monitorThread = new Thread(monitor, "cluster-" + this.serverId.getClusterId() + "-" + this.serverId.getAddress());
@@ -251,7 +255,21 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
         }
 
         private boolean shouldStreamResponses(final ServerDescription currentServerDescription) {
-            return currentServerDescription.getTopologyVersion() != null;
+            boolean serverSupportsStreaming = currentServerDescription.getTopologyVersion() != null;
+            switch (serverSettings.getServerMonitoringMode()) {
+                case STREAM: {
+                    return serverSupportsStreaming;
+                }
+                case POLL: {
+                    return false;
+                }
+                case AUTO: {
+                    return !faas && serverSupportsStreaming;
+                }
+                default: {
+                    throw fail();
+                }
+            }
         }
 
         private CommandMessage createCommandMessage(final BsonDocument command, final InternalConnection connection,

diff --git a/...ore/src/test/functional/com/mongodb/internal/connection/ServerMonitorSpecification.groovy b/...ore/src/test/functional/com/mongodb/internal/connection/ServerMonitorSpecification.groovy
@@ -224,7 +224,7 @@ class ServerMonitorSpecification extends OperationFunctionalSpecification {
                         SocketSettings.builder().connectTimeout(500, TimeUnit.MILLISECONDS).build(), getSslSettings()),
                         getCredentialWithCache(), null, null, [], LoggerSettings.builder().build(), null,
                         getServerApi()),
-                getClusterConnectionMode(), getServerApi(), SameObjectProvider.initialized(sdam))
+                getClusterConnectionMode(), getServerApi(), false, SameObjectProvider.initialized(sdam))
         serverMonitor.start()
         serverMonitor
     }

diff --git a/driver-core/src/test/functional/com/mongodb/internal/connection/SingleServerClusterTest.java b/driver-core/src/test/functional/com/mongodb/internal/connection/SingleServerClusterTest.java
@@ -69,7 +69,7 @@ private void setUpCluster(final ServerAddress serverAddress) {
                         streamFactory, streamFactory, getCredential(),
 
                         LoggerSettings.builder().build(), null, null, null,
-                        Collections.emptyList(), getServerApi()));
+                        Collections.emptyList(), getServerApi(), false));
     }
 
     @After