diff --git a/driver-core/src/main/com/mongodb/ConnectionString.java b/driver-core/src/main/com/mongodb/ConnectionString.java
index 5e6a5b7d81..e715b8983f 100644
--- a/driver-core/src/main/com/mongodb/ConnectionString.java
+++ b/driver-core/src/main/com/mongodb/ConnectionString.java
@@ -18,12 +18,15 @@
 
 import com.mongodb.connection.ClusterSettings;
 import com.mongodb.connection.ConnectionPoolSettings;
+import com.mongodb.connection.ServerMonitoringMode;
+import com.mongodb.connection.ServerSettings;
 import com.mongodb.connection.SocketSettings;
 import com.mongodb.event.ConnectionCheckOutStartedEvent;
 import com.mongodb.event.ConnectionCheckedInEvent;
 import com.mongodb.event.ConnectionCheckedOutEvent;
 import com.mongodb.event.ConnectionCreatedEvent;
 import com.mongodb.event.ConnectionReadyEvent;
+import com.mongodb.internal.connection.ServerMonitoringModeUtil;
 import com.mongodb.internal.diagnostics.logging.Logger;
 import com.mongodb.internal.diagnostics.logging.Loggers;
 import com.mongodb.internal.dns.DefaultDnsResolver;
@@ -111,6 +114,13 @@
  * <ul>
  * <li>{@code heartbeatFrequencyMS=ms}: The frequency that the driver will attempt to determine the current state of each server in the
  * cluster.</li>
+ * <li>{@code serverMonitoringMode=enum}: The server monitoring mode, which defines the monitoring protocol to use. Enumerated values:
+ *  <ul>
+ *      <li>{@code stream};</li>
+ *      <li>{@code poll};</li>
+ *      <li>{@code auto} - the default.</li>
+ *  </ul>
+ * </li>
  * </ul>
  * <p>Replica set configuration:</p>
  * <ul>
@@ -307,6 +317,7 @@ public class ConnectionString {
     private Integer serverSelectionTimeout;
     private Integer localThreshold;
     private Integer heartbeatFrequency;
+    private ServerMonitoringMode serverMonitoringMode;
     private String applicationName;
     private List<MongoCompressor> compressorList;
     private UuidRepresentation uuidRepresentation;
@@ -529,6 +540,7 @@ public ConnectionString(final String connectionString, @Nullable final DnsClient
         GENERAL_OPTIONS_KEYS.add("serverselectiontimeoutms");
         GENERAL_OPTIONS_KEYS.add("localthresholdms");
         GENERAL_OPTIONS_KEYS.add("heartbeatfrequencyms");
+        GENERAL_OPTIONS_KEYS.add("servermonitoringmode");
         GENERAL_OPTIONS_KEYS.add("retrywrites");
         GENERAL_OPTIONS_KEYS.add("retryreads");
 
@@ -665,6 +677,9 @@ private void translateOptions(final Map<String, List<String>> optionsMap) {
                 case "heartbeatfrequencyms":
                     heartbeatFrequency = parseInteger(value, "heartbeatfrequencyms");
                     break;
+                case "servermonitoringmode":
+                    serverMonitoringMode = ServerMonitoringModeUtil.fromString(value);
+                    break;
                 case "appname":
                     applicationName = value;
                     break;
@@ -1623,6 +1638,20 @@ public Integer getHeartbeatFrequency() {
         return heartbeatFrequency;
     }
 
+    /**
+     * The server monitoring mode, which defines the monitoring protocol to use.
+     * <p>
+     * Default is {@link ServerMonitoringMode#AUTO}.</p>
+     *
+     * @return The {@link ServerMonitoringMode}, or {@code null} if unset and the default is to be used.
+     * @see ServerSettings#getServerMonitoringMode()
+     * @since 5.1
+     */
+    @Nullable
+    public ServerMonitoringMode getServerMonitoringMode() {
+        return serverMonitoringMode;
+    }
+
     /**
      * Gets the logical name of the application.  The application name may be used by the client to identify the application to the server,
      * for use in server logs, slow query logs, and profile collection.
@@ -1704,6 +1733,7 @@ public boolean equals(final Object o) {
                 && Objects.equals(serverSelectionTimeout, that.serverSelectionTimeout)
                 && Objects.equals(localThreshold, that.localThreshold)
                 && Objects.equals(heartbeatFrequency, that.heartbeatFrequency)
+                && Objects.equals(serverMonitoringMode, that.serverMonitoringMode)
                 && Objects.equals(applicationName, that.applicationName)
                 && Objects.equals(compressorList, that.compressorList)
                 && Objects.equals(uuidRepresentation, that.uuidRepresentation)
@@ -1717,7 +1747,7 @@ public int hashCode() {
                 writeConcern, retryWrites, retryReads, readConcern, minConnectionPoolSize, maxConnectionPoolSize, maxWaitTime,
                 maxConnectionIdleTime, maxConnectionLifeTime, maxConnecting, connectTimeout, socketTimeout, sslEnabled,
                 sslInvalidHostnameAllowed, requiredReplicaSetName, serverSelectionTimeout, localThreshold, heartbeatFrequency,
-                applicationName, compressorList, uuidRepresentation, srvServiceName, srvMaxHosts, proxyHost, proxyPort,
-                proxyUsername, proxyPassword);
+                serverMonitoringMode, applicationName, compressorList, uuidRepresentation, srvServiceName, srvMaxHosts, proxyHost,
+                proxyPort, proxyUsername, proxyPassword);
     }
 }
diff --git a/driver-core/src/main/com/mongodb/connection/ServerMonitoringMode.java b/driver-core/src/main/com/mongodb/connection/ServerMonitoringMode.java
new file mode 100644
index 0000000000..cf54afef4b
--- /dev/null
+++ b/driver-core/src/main/com/mongodb/connection/ServerMonitoringMode.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.connection;
+
+import com.mongodb.event.ClusterListener;
+import com.mongodb.event.ServerHeartbeatFailedEvent;
+import com.mongodb.event.ServerHeartbeatStartedEvent;
+import com.mongodb.event.ServerHeartbeatSucceededEvent;
+import com.mongodb.event.ServerListener;
+
+/**
+ * The server monitoring mode, which defines the monitoring protocol to use.
+ *
+ * @see <a href="https://www.mongodb.com/docs/drivers/java/sync/current/fundamentals/monitoring/#server-discovery-and-monitoring-events">
+ * server discovery and monitoring (SDAM)</a>
+ * @since 5.1
+ */
+public enum ServerMonitoringMode {
+    /**
+     * Use the streaming protocol when the server supports it or fall back to the polling protocol otherwise.
+     * When the streaming protocol comes into play,
+     * {@link ServerHeartbeatStartedEvent#isAwaited()}, {@link ServerHeartbeatSucceededEvent#isAwaited()},
+     * {@link ServerHeartbeatFailedEvent#isAwaited()} return {@code true} for new events.
+     * <p>
+     * The streaming protocol uses long polling for server monitoring, and is intended to reduce the delay between a server change
+     * that warrants a new event for {@link ServerListener}/{@link ClusterListener},
+     * and that event being emitted, as well as the related housekeeping work being done.</p>
+     */
+    STREAM(),
+    /**
+     * Use the polling protocol.
+     */
+    POLL(),
+    /**
+     * Behave the same as {@link #POLL} if running in a FaaS environment, otherwise behave as {@link #STREAM}.
+     * This is the default.
+     */
+    AUTO()
+}
diff --git a/driver-core/src/main/com/mongodb/connection/ServerSettings.java b/driver-core/src/main/com/mongodb/connection/ServerSettings.java
index d4ef398a04..b4394d0dc7 100644
--- a/driver-core/src/main/com/mongodb/connection/ServerSettings.java
+++ b/driver-core/src/main/com/mongodb/connection/ServerSettings.java
@@ -24,6 +24,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.assertions.Assertions.notNull;
@@ -38,6 +39,7 @@
 public class ServerSettings {
     private final long heartbeatFrequencyMS;
     private final long minHeartbeatFrequencyMS;
+    private final ServerMonitoringMode serverMonitoringMode;
     private final List<ServerListener> serverListeners;
     private final List<ServerMonitorListener> serverMonitorListeners;
 
@@ -68,6 +70,7 @@ public static Builder builder(final ServerSettings serverSettings) {
     public static final class Builder {
         private long heartbeatFrequencyMS = 10000;
         private long minHeartbeatFrequencyMS = 500;
+        private ServerMonitoringMode serverMonitoringMode = ServerMonitoringMode.AUTO;
         private List<ServerListener> serverListeners = new ArrayList<>();
         private List<ServerMonitorListener> serverMonitorListeners = new ArrayList<>();
 
@@ -87,6 +90,7 @@ public Builder applySettings(final ServerSettings serverSettings) {
             notNull("serverSettings", serverSettings);
             heartbeatFrequencyMS = serverSettings.heartbeatFrequencyMS;
             minHeartbeatFrequencyMS = serverSettings.minHeartbeatFrequencyMS;
+            serverMonitoringMode = serverSettings.serverMonitoringMode;
             serverListeners = new ArrayList<>(serverSettings.serverListeners);
             serverMonitorListeners = new ArrayList<>(serverSettings.serverMonitorListeners);
             return this;
@@ -117,6 +121,20 @@ public Builder minHeartbeatFrequency(final long minHeartbeatFrequency, final Tim
             return this;
         }
 
+        /**
+         * Sets the server monitoring mode, which defines the monitoring protocol to use.
+         * The default value is {@link ServerMonitoringMode#AUTO}.
+         *
+         * @param serverMonitoringMode The {@link ServerMonitoringMode}.
+         * @return {@code this}.
+         * @see #getServerMonitoringMode()
+         * @since 5.1
+         */
+        public Builder serverMonitoringMode(final ServerMonitoringMode serverMonitoringMode) {
+            this.serverMonitoringMode = notNull("serverMonitoringMode", serverMonitoringMode);
+            return this;
+        }
+
         /**
          * Add a server listener.
          *
@@ -181,6 +199,10 @@ public Builder applyConnectionString(final ConnectionString connectionString) {
             if (heartbeatFrequency != null) {
                 heartbeatFrequencyMS = heartbeatFrequency;
             }
+            ServerMonitoringMode serverMonitoringMode = connectionString.getServerMonitoringMode();
+            if (serverMonitoringMode != null) {
+                this.serverMonitoringMode = serverMonitoringMode;
+            }
             return this;
         }
 
@@ -215,6 +237,19 @@ public long getMinHeartbeatFrequency(final TimeUnit timeUnit) {
         return timeUnit.convert(minHeartbeatFrequencyMS, TimeUnit.MILLISECONDS);
     }
 
+    /**
+     * Gets the server monitoring mode, which defines the monitoring protocol to use.
+     * The default value is {@link ServerMonitoringMode#AUTO}.
+     *
+     * @return The {@link ServerMonitoringMode}.
+     * @see Builder#serverMonitoringMode(ServerMonitoringMode)
+     * @see ConnectionString#getServerMonitoringMode()
+     * @since 5.1
+     */
+    public ServerMonitoringMode getServerMonitoringMode() {
+        return serverMonitoringMode;
+    }
+
     /**
      * Gets the server listeners.  The default value is an empty list.
      *
@@ -243,33 +278,22 @@ public boolean equals(final Object o) {
         if (o == null || getClass() != o.getClass()) {
             return false;
         }
-
-        ServerSettings that = (ServerSettings) o;
-
-        if (heartbeatFrequencyMS != that.heartbeatFrequencyMS) {
-            return false;
-        }
-        if (minHeartbeatFrequencyMS != that.minHeartbeatFrequencyMS) {
-            return false;
-        }
-
-        if (!serverListeners.equals(that.serverListeners)) {
-            return false;
-        }
-        if (!serverMonitorListeners.equals(that.serverMonitorListeners)) {
-            return false;
-        }
-
-        return true;
+        final ServerSettings that = (ServerSettings) o;
+        return heartbeatFrequencyMS == that.heartbeatFrequencyMS
+                && minHeartbeatFrequencyMS == that.minHeartbeatFrequencyMS
+                && serverMonitoringMode == that.serverMonitoringMode
+                && Objects.equals(serverListeners, that.serverListeners)
+                && Objects.equals(serverMonitorListeners, that.serverMonitorListeners);
     }
 
     @Override
     public int hashCode() {
-        int result = (int) (heartbeatFrequencyMS ^ (heartbeatFrequencyMS >>> 32));
-        result = 31 * result + (int) (minHeartbeatFrequencyMS ^ (minHeartbeatFrequencyMS >>> 32));
-        result = 31 * result + serverListeners.hashCode();
-        result = 31 * result + serverMonitorListeners.hashCode();
-        return result;
+        return Objects.hash(
+                heartbeatFrequencyMS,
+                minHeartbeatFrequencyMS,
+                serverMonitoringMode,
+                serverListeners,
+                serverMonitorListeners);
     }
 
     @Override
@@ -277,6 +301,7 @@ public String toString() {
         return "ServerSettings{"
                + "heartbeatFrequencyMS=" + heartbeatFrequencyMS
                + ", minHeartbeatFrequencyMS=" + minHeartbeatFrequencyMS
+               + ", serverMonitoringMode=" + serverMonitoringMode
                + ", serverListeners='" + serverListeners + '\''
                + ", serverMonitorListeners='" + serverMonitorListeners + '\''
                + '}';
@@ -285,6 +310,7 @@ public String toString() {
     ServerSettings(final Builder builder) {
         heartbeatFrequencyMS = builder.heartbeatFrequencyMS;
         minHeartbeatFrequencyMS = builder.minHeartbeatFrequencyMS;
+        serverMonitoringMode = builder.serverMonitoringMode;
         serverListeners = unmodifiableList(builder.serverListeners);
         serverMonitorListeners = unmodifiableList(builder.serverMonitorListeners);
     }
diff --git a/driver-core/src/main/com/mongodb/event/ServerHeartbeatFailedEvent.java b/driver-core/src/main/com/mongodb/event/ServerHeartbeatFailedEvent.java
index b324ddb84c..a8468effe4 100644
--- a/driver-core/src/main/com/mongodb/event/ServerHeartbeatFailedEvent.java
+++ b/driver-core/src/main/com/mongodb/event/ServerHeartbeatFailedEvent.java
@@ -17,6 +17,7 @@
 package com.mongodb.event;
 
 import com.mongodb.connection.ConnectionId;
+import com.mongodb.connection.ServerMonitoringMode;
 
 import java.util.concurrent.TimeUnit;
 
@@ -77,6 +78,7 @@ public long getElapsedTime(final TimeUnit timeUnit) {
      * to the server and the time that the server waited before sending a response.
      *
      * @return whether the response was awaited
+     * @see ServerMonitoringMode#STREAM
      * @since 4.1
      * @mongodb.server.release 4.4
      */
diff --git a/driver-core/src/main/com/mongodb/event/ServerHeartbeatStartedEvent.java b/driver-core/src/main/com/mongodb/event/ServerHeartbeatStartedEvent.java
index 4397c510d9..f83dc5ef00 100644
--- a/driver-core/src/main/com/mongodb/event/ServerHeartbeatStartedEvent.java
+++ b/driver-core/src/main/com/mongodb/event/ServerHeartbeatStartedEvent.java
@@ -17,6 +17,7 @@
 package com.mongodb.event;
 
 import com.mongodb.connection.ConnectionId;
+import com.mongodb.connection.ServerMonitoringMode;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -27,14 +28,30 @@
  */
 public final class ServerHeartbeatStartedEvent {
     private final ConnectionId connectionId;
+    private final boolean awaited;
 
     /**
      * Construct an instance.
      *
      * @param connectionId the non-null connnectionId
+     * @param awaited {@code true} if and only if the heartbeat is for an awaitable `hello` / legacy hello.
+     * @since 5.1
      */
-    public ServerHeartbeatStartedEvent(final ConnectionId connectionId) {
+    public ServerHeartbeatStartedEvent(final ConnectionId connectionId, final boolean awaited) {
         this.connectionId = notNull("connectionId", connectionId);
+        this.awaited = awaited;
+    }
+
+    /**
+     * Construct an instance.
+     *
+     * @param connectionId the non-null connnectionId
+     * @deprecated Prefer {@link #ServerHeartbeatStartedEvent(ConnectionId, boolean)}.
+     * If this constructor is used then {@link #isAwaited()} is {@code false}.
+     */
+    @Deprecated
+    public ServerHeartbeatStartedEvent(final ConnectionId connectionId) {
+        this(connectionId, false);
     }
 
     /**
@@ -46,12 +63,24 @@ public ConnectionId getConnectionId() {
         return connectionId;
     }
 
+    /**
+     * Gets whether the heartbeat is for an awaitable `hello` / legacy hello.
+     *
+     * @return {@code true} if and only if the heartbeat is for an awaitable `hello` / legacy hello.
+     * @see ServerMonitoringMode#STREAM
+     * @since 5.1
+     */
+    public boolean isAwaited() {
+        return awaited;
+    }
+
     @Override
     public String toString() {
         return "ServerHeartbeatStartedEvent{"
                 + "connectionId=" + connectionId
                 + ", server=" + connectionId.getServerId().getAddress()
                 + ", clusterId=" + connectionId.getServerId().getClusterId()
+                + ", awaited=" + awaited
                 + "} " + super.toString();
     }
 }
diff --git a/driver-core/src/main/com/mongodb/event/ServerHeartbeatSucceededEvent.java b/driver-core/src/main/com/mongodb/event/ServerHeartbeatSucceededEvent.java
index e6deb0bb7a..20e9741275 100644
--- a/driver-core/src/main/com/mongodb/event/ServerHeartbeatSucceededEvent.java
+++ b/driver-core/src/main/com/mongodb/event/ServerHeartbeatSucceededEvent.java
@@ -17,6 +17,7 @@
 package com.mongodb.event;
 
 import com.mongodb.connection.ConnectionId;
+import com.mongodb.connection.ServerMonitoringMode;
 import org.bson.BsonDocument;
 
 import java.util.concurrent.TimeUnit;
@@ -87,6 +88,7 @@ public long getElapsedTime(final TimeUnit timeUnit) {
      * to the server and the time that the server waited before sending a response.
      *
      * @return whether the response was awaited
+     * @see ServerMonitoringMode#STREAM
      * @since 4.1
      * @mongodb.server.release 4.4
      */
diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java
index 38fa28d3b3..0375373c23 100644
--- 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
index 5f5b1e97b1..7d0f5b62e5 100644
--- 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 isFunctionAsAServiceEnvironment;
 
     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 isFunctionAsAServiceEnvironment) {
         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.isFunctionAsAServiceEnvironment = isFunctionAsAServiceEnvironment;
     }
 
     @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, isFunctionAsAServiceEnvironment, 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
index e4618fc31f..55030a6db3 100644
--- 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,12 +77,15 @@ class DefaultServerMonitor implements ServerMonitor {
     private final ClusterConnectionMode clusterConnectionMode;
     @Nullable
     private final ServerApi serverApi;
+    private final boolean isFunctionAsAServiceEnvironment;
     private final ServerSettings serverSettings;
-    private final ServerMonitorRunnable monitor;
-    private final Thread monitorThread;
-    private final RoundTripTimeRunnable roundTripTimeMonitor;
+    private final ServerMonitor monitor;
+    /**
+     * Must be guarded by {@link #lock}.
+     */
+    @Nullable
+    private RoundTripTimeMonitor roundTripTimeMonitor;
     private final ExponentiallyWeightedMovingAverage averageRoundTripTime = new ExponentiallyWeightedMovingAverage(0.2);
-    private final Thread roundTripTimeMonitorThread;
     private final Lock lock = new ReentrantLock();
     private final Condition condition = lock.newCondition();
     private volatile boolean isClosed;
@@ -90,6 +94,7 @@ class DefaultServerMonitor implements ServerMonitor {
             final InternalConnectionFactory internalConnectionFactory,
                          final ClusterConnectionMode clusterConnectionMode,
                          @Nullable final ServerApi serverApi,
+                         final boolean isFunctionAsAServiceEnvironment,
                          final Provider<SdamServerDescriptionManager> sdamProvider) {
         this.serverSettings = notNull("serverSettings", serverSettings);
         this.serverId = notNull("serverId", serverId);
@@ -97,21 +102,25 @@ class DefaultServerMonitor implements ServerMonitor {
         this.internalConnectionFactory = notNull("internalConnectionFactory", internalConnectionFactory);
         this.clusterConnectionMode = notNull("clusterConnectionMode", clusterConnectionMode);
         this.serverApi = serverApi;
+        this.isFunctionAsAServiceEnvironment = isFunctionAsAServiceEnvironment;
         this.sdamProvider = sdamProvider;
-        monitor = new ServerMonitorRunnable();
-        monitorThread = new Thread(monitor, "cluster-" + this.serverId.getClusterId() + "-" + this.serverId.getAddress());
-        monitorThread.setDaemon(true);
-        roundTripTimeMonitor = new RoundTripTimeRunnable();
-        roundTripTimeMonitorThread = new Thread(roundTripTimeMonitor,
-                "cluster-rtt-" + this.serverId.getClusterId() + "-" + this.serverId.getAddress());
-        roundTripTimeMonitorThread.setDaemon(true);
+        monitor = new ServerMonitor();
+        roundTripTimeMonitor = null;
         isClosed = false;
     }
 
     @Override
     public void start() {
-        monitorThread.start();
-        roundTripTimeMonitorThread.start();
+        monitor.start();
+    }
+
+    private void ensureRoundTripTimeMonitorStarted() {
+        withLock(lock, () -> {
+            if (!isClosed && roundTripTimeMonitor == null) {
+                roundTripTimeMonitor = new RoundTripTimeMonitor();
+                roundTripTimeMonitor.start();
+            }
+        });
     }
 
     @Override
@@ -120,12 +129,16 @@ public void connect() {
     }
 
     @Override
+    @SuppressWarnings("try")
     public void close() {
-        isClosed = true;
-        monitor.close();
-        monitorThread.interrupt();
-        roundTripTimeMonitor.close();
-        roundTripTimeMonitorThread.interrupt();
+        withLock(lock, () -> {
+            isClosed = true;
+            //noinspection EmptyTryBlock
+            try (ServerMonitor ignoredAutoClosed = monitor;
+                RoundTripTimeMonitor ignoredAutoClose2 = roundTripTimeMonitor) {
+                // we are automatically closing resources here
+            }
+        });
     }
 
     @Override
@@ -133,11 +146,18 @@ public void cancelCurrentCheck() {
         monitor.cancelCurrentCheck();
     }
 
-    class ServerMonitorRunnable implements Runnable {
+    class ServerMonitor extends Thread implements AutoCloseable {
         private volatile InternalConnection connection = null;
         private volatile boolean currentCheckCancelled;
 
-        void close() {
+        ServerMonitor() {
+            super("cluster-" + serverId.getClusterId() + "-" + serverId.getAddress());
+            setDaemon(true);
+        }
+
+        @Override
+        public void close() {
+            interrupt();
             InternalConnection connection = this.connection;
             if (connection != null) {
                 connection.close();
@@ -151,6 +171,10 @@ public void run() {
                 while (!isClosed) {
                     ServerDescription previousServerDescription = currentServerDescription;
                     currentServerDescription = lookupServerDescription(currentServerDescription);
+                    boolean shouldStreamResponses = shouldStreamResponses(currentServerDescription);
+                    if (shouldStreamResponses) {
+                        ensureRoundTripTimeMonitorStarted();
+                    }
 
                     if (isClosed) {
                         continue;
@@ -165,8 +189,7 @@ public void run() {
                     logStateChange(previousServerDescription, currentServerDescription);
                     sdamProvider.get().update(currentServerDescription);
 
-                    if (((connection == null || shouldStreamResponses(currentServerDescription))
-                            && currentServerDescription.getTopologyVersion() != null && currentServerDescription.getType() != UNKNOWN)
+                    if ((shouldStreamResponses && currentServerDescription.getType() != UNKNOWN)
                             || (connection != null && connection.hasMoreToCome())
                             || (currentServerDescription.getException() instanceof MongoSocketException
                             && previousServerDescription.getType() != UNKNOWN)) {
@@ -199,7 +222,9 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug(format("Checking status of %s", serverId.getAddress()));
                 }
-                serverMonitorListener.serverHearbeatStarted(new ServerHeartbeatStartedEvent(connection.getDescription().getConnectionId()));
+                boolean shouldStreamResponses = shouldStreamResponses(currentServerDescription);
+                serverMonitorListener.serverHearbeatStarted(new ServerHeartbeatStartedEvent(
+                        connection.getDescription().getConnectionId(), shouldStreamResponses));
 
                 long start = System.nanoTime();
                 try {
@@ -207,7 +232,7 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                     if (!connection.hasMoreToCome()) {
                         BsonDocument helloDocument = new BsonDocument(getHandshakeCommandName(currentServerDescription), new BsonInt32(1))
                                 .append("helloOk", BsonBoolean.TRUE);
-                        if (shouldStreamResponses(currentServerDescription)) {
+                        if (shouldStreamResponses) {
                             helloDocument.append("topologyVersion", assertNotNull(currentServerDescription.getTopologyVersion()).asDocument());
                             helloDocument.append("maxAwaitTimeMS", new BsonInt64(serverSettings.getHeartbeatFrequency(MILLISECONDS)));
                         }
@@ -217,7 +242,7 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                     }
 
                     BsonDocument helloResult;
-                    if (shouldStreamResponses(currentServerDescription)) {
+                    if (shouldStreamResponses) {
                         helloResult = connection.receive(new BsonDocumentCodec(), sessionContext,
                                 Math.toIntExact(serverSettings.getHeartbeatFrequency(MILLISECONDS)));
                     } else {
@@ -225,15 +250,18 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                     }
 
                     long elapsedTimeNanos = System.nanoTime() - start;
+                    if (!shouldStreamResponses) {
+                        averageRoundTripTime.addSample(elapsedTimeNanos);
+                    }
                     serverMonitorListener.serverHeartbeatSucceeded(
                             new ServerHeartbeatSucceededEvent(connection.getDescription().getConnectionId(), helloResult,
-                                    elapsedTimeNanos, currentServerDescription.getTopologyVersion() != null));
+                                    elapsedTimeNanos, shouldStreamResponses));
 
                     return createServerDescription(serverId.getAddress(), helloResult, averageRoundTripTime.getAverage());
                 } catch (Exception e) {
                     serverMonitorListener.serverHeartbeatFailed(
                             new ServerHeartbeatFailedEvent(connection.getDescription().getConnectionId(), System.nanoTime() - start,
-                                    currentServerDescription.getTopologyVersion() != null, e));
+                                    shouldStreamResponses, e));
                     throw e;
                 }
             } catch (Throwable t) {
@@ -251,7 +279,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 !isFunctionAsAServiceEnvironment && serverSupportsStreaming;
+                }
+                default: {
+                    throw fail();
+                }
+            }
         }
 
         private CommandMessage createCommandMessage(final BsonDocument command, final InternalConnection connection,
@@ -381,10 +423,17 @@ static boolean shouldLogStageChange(final ServerDescription previous, final Serv
     }
 
 
-    private class RoundTripTimeRunnable implements Runnable {
+    private class RoundTripTimeMonitor extends Thread implements AutoCloseable {
         private volatile InternalConnection connection = null;
 
-        void close() {
+        RoundTripTimeMonitor() {
+            super("cluster-rtt-" + serverId.getClusterId() + "-" + serverId.getAddress());
+            setDaemon(true);
+        }
+
+        @Override
+        public void close() {
+            interrupt();
             InternalConnection connection = this.connection;
             if (connection != null) {
                 connection.close();
diff --git a/driver-core/src/main/com/mongodb/internal/connection/ServerMonitoringModeUtil.java b/driver-core/src/main/com/mongodb/internal/connection/ServerMonitoringModeUtil.java
new file mode 100644
index 0000000000..17629f38a5
--- /dev/null
+++ b/driver-core/src/main/com/mongodb/internal/connection/ServerMonitoringModeUtil.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.internal.connection;
+
+import com.mongodb.connection.ServerMonitoringMode;
+
+import static java.lang.String.format;
+
+/**
+ * <p>This class is not part of the public API and may be removed or changed at any time</p>
+ */
+public final class ServerMonitoringModeUtil {
+    /**
+     * Returns the string value of the provided {@code serverMonitoringMode}.
+     *
+     * @return The string value.
+     * @see #fromString(String)
+     */
+    public static String getValue(final ServerMonitoringMode serverMonitoringMode) {
+        return serverMonitoringMode.name().toLowerCase();
+    }
+
+    /**
+     * Parses a string into {@link ServerMonitoringMode}.
+     *
+     * @param serverMonitoringMode A server monitoring mode string.
+     * @return The corresponding {@link ServerMonitoringMode} value.
+     * @see #getValue(ServerMonitoringMode)
+     */
+    public static ServerMonitoringMode fromString(final String serverMonitoringMode) {
+        for (ServerMonitoringMode mode : ServerMonitoringMode.values()) {
+            if (serverMonitoringMode.equalsIgnoreCase(mode.name())) {
+                return mode;
+            }
+        }
+        throw new IllegalArgumentException(format("'%s' is not a valid %s",
+                serverMonitoringMode, ServerMonitoringMode.class.getSimpleName()));
+    }
+
+    private ServerMonitoringModeUtil() {
+    }
+}
diff --git a/driver-core/src/test/functional/com/mongodb/internal/connection/ServerMonitorSpecification.groovy b/driver-core/src/test/functional/com/mongodb/internal/connection/ServerMonitorSpecification.groovy
index 8e69c609c8..0f2ba70d4c 100644
--- a/driver-core/src/test/functional/com/mongodb/internal/connection/ServerMonitorSpecification.groovy
+++ b/driver-core/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
index 55ba6875a1..e715bfb5cd 100644
--- 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
diff --git a/driver-core/src/test/resources/unified-test-format/server-discovery-and-monitoring/serverMonitoringMode-java-specific.json b/driver-core/src/test/resources/unified-test-format/server-discovery-and-monitoring/serverMonitoringMode-java-specific.json
new file mode 100644
index 0000000000..2d50d211c8
--- /dev/null
+++ b/driver-core/src/test/resources/unified-test-format/server-discovery-and-monitoring/serverMonitoringMode-java-specific.json
@@ -0,0 +1,190 @@
+{
+  "description": "serverMonitoringMode-Java-specific",
+  "schemaVersion": "1.17",
+  "runOnRequirements": [
+    {
+      "topologies": [
+        "single",
+        "sharded",
+        "sharded-replicaset"
+      ],
+      "serverless": "forbid"
+    }
+  ],
+  "tests": [
+    {
+      "description": "connect with serverMonitoringMode=auto >=4.4 Java-specific",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "4.4.0"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "auto"
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": true
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": true
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": true
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "connect with serverMonitoringMode=stream >=4.4 Java-specific",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "4.4.0"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "stream"
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": true
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": true
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": true
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
diff --git a/driver-core/src/test/resources/unified-test-format/server-discovery-and-monitoring/serverMonitoringMode.json b/driver-core/src/test/resources/unified-test-format/server-discovery-and-monitoring/serverMonitoringMode.json
new file mode 100644
index 0000000000..7d681b4f9e
--- /dev/null
+++ b/driver-core/src/test/resources/unified-test-format/server-discovery-and-monitoring/serverMonitoringMode.json
@@ -0,0 +1,449 @@
+{
+  "description": "serverMonitoringMode",
+  "schemaVersion": "1.17",
+  "runOnRequirements": [
+    {
+      "topologies": [
+        "single",
+        "sharded",
+        "sharded-replicaset"
+      ],
+      "serverless": "forbid"
+    }
+  ],
+  "tests": [
+    {
+      "description": "connect with serverMonitoringMode=auto >=4.4",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "4.4.0"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "auto"
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": true
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "connect with serverMonitoringMode=auto <4.4",
+      "runOnRequirements": [
+        {
+          "maxServerVersion": "4.2.99"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "auto",
+                    "heartbeatFrequencyMS": 500
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "connect with serverMonitoringMode=stream >=4.4",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "4.4.0"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "stream"
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": true
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "connect with serverMonitoringMode=stream <4.4",
+      "runOnRequirements": [
+        {
+          "maxServerVersion": "4.2.99"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "stream",
+                    "heartbeatFrequencyMS": 500
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "connect with serverMonitoringMode=poll",
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "uriOptions": {
+                    "serverMonitoringMode": "poll",
+                    "heartbeatFrequencyMS": 500
+                  },
+                  "useMultipleMongoses": false,
+                  "observeEvents": [
+                    "serverHeartbeatStartedEvent",
+                    "serverHeartbeatSucceededEvent",
+                    "serverHeartbeatFailedEvent"
+                  ]
+                }
+              },
+              {
+                "database": {
+                  "id": "db",
+                  "client": "client",
+                  "databaseName": "sdam-tests"
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "runCommand",
+          "object": "db",
+          "arguments": {
+            "commandName": "ping",
+            "command": {
+              "ping": 1
+            }
+          },
+          "expectResult": {
+            "ok": 1
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "serverHeartbeatStartedEvent": {}
+            },
+            "count": 2
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "ignoreExtraEvents": true,
+          "events": [
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatSucceededEvent": {
+                "awaited": false
+              }
+            },
+            {
+              "serverHeartbeatStartedEvent": {
+                "awaited": false
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
diff --git a/driver-core/src/test/resources/uri-options/sdam-options.json b/driver-core/src/test/resources/uri-options/sdam-options.json
new file mode 100644
index 0000000000..673f5607ee
--- /dev/null
+++ b/driver-core/src/test/resources/uri-options/sdam-options.json
@@ -0,0 +1,46 @@
+{
+  "tests": [
+    {
+      "description": "serverMonitoringMode=auto",
+      "uri": "mongodb://example.com/?serverMonitoringMode=auto",
+      "valid": true,
+      "warning": false,
+      "hosts": null,
+      "auth": null,
+      "options": {
+        "serverMonitoringMode": "auto"
+      }
+    },
+    {
+      "description": "serverMonitoringMode=stream",
+      "uri": "mongodb://example.com/?serverMonitoringMode=stream",
+      "valid": true,
+      "warning": false,
+      "hosts": null,
+      "auth": null,
+      "options": {
+        "serverMonitoringMode": "stream"
+      }
+    },
+    {
+      "description": "serverMonitoringMode=poll",
+      "uri": "mongodb://example.com/?serverMonitoringMode=poll",
+      "valid": true,
+      "warning": false,
+      "hosts": null,
+      "auth": null,
+      "options": {
+        "serverMonitoringMode": "poll"
+      }
+    },
+    {
+      "description": "invalid serverMonitoringMode",
+      "uri": "mongodb://example.com/?serverMonitoringMode=invalid",
+      "valid": true,
+      "warning": true,
+      "hosts": null,
+      "auth": null,
+      "options": {}
+    }
+  ]
+}
diff --git a/driver-core/src/test/unit/com/mongodb/AbstractConnectionStringTest.java b/driver-core/src/test/unit/com/mongodb/AbstractConnectionStringTest.java
index c5b90aa95f..bb26a3edb5 100644
--- a/driver-core/src/test/unit/com/mongodb/AbstractConnectionStringTest.java
+++ b/driver-core/src/test/unit/com/mongodb/AbstractConnectionStringTest.java
@@ -16,6 +16,7 @@
 
 package com.mongodb;
 
+import com.mongodb.internal.connection.ServerMonitoringModeUtil;
 import com.mongodb.lang.Nullable;
 import junit.framework.TestCase;
 import org.bson.BsonArray;
@@ -139,6 +140,9 @@ protected void testValidOptions() {
             } else if (option.getKey().equalsIgnoreCase("heartbeatfrequencyms")) {
                 int expected = option.getValue().asInt32().getValue();
                 assertEquals(expected, connectionString.getHeartbeatFrequency().intValue());
+            } else if (option.getKey().equalsIgnoreCase("servermonitoringmode")) {
+                String expected = option.getValue().asString().getValue();
+                assertEquals(expected, ServerMonitoringModeUtil.getValue(connectionString.getServerMonitoringMode()));
             } else if (option.getKey().equalsIgnoreCase("localthresholdms")) {
                 int expected = option.getValue().asInt32().getValue();
                 assertEquals(expected, connectionString.getLocalThreshold().intValue());
diff --git a/driver-core/src/test/unit/com/mongodb/ConnectionStringSpecification.groovy b/driver-core/src/test/unit/com/mongodb/ConnectionStringSpecification.groovy
index e0245e8009..e8731439a8 100644
--- a/driver-core/src/test/unit/com/mongodb/ConnectionStringSpecification.groovy
+++ b/driver-core/src/test/unit/com/mongodb/ConnectionStringSpecification.groovy
@@ -35,6 +35,9 @@ import static com.mongodb.ReadPreference.secondaryPreferred
 import static java.util.Arrays.asList
 import static java.util.concurrent.TimeUnit.MILLISECONDS
 
+/**
+ * Update {@link ConnectionStringUnitTest} instead.
+ */
 class ConnectionStringSpecification extends Specification {
     static final LONG_STRING = new String((1..256).collect { (byte) 1 } as byte[])
 
diff --git a/driver-core/src/test/unit/com/mongodb/ConnectionStringUnitTest.java b/driver-core/src/test/unit/com/mongodb/ConnectionStringUnitTest.java
new file mode 100644
index 0000000000..d2e41ebeaf
--- /dev/null
+++ b/driver-core/src/test/unit/com/mongodb/ConnectionStringUnitTest.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb;
+
+import com.mongodb.connection.ServerMonitoringMode;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+final class ConnectionStringUnitTest {
+    private static final String DEFAULT_OPTIONS = "mongodb://localhost/?";
+    @Test
+    void defaults() {
+        ConnectionString connectionStringDefault = new ConnectionString(DEFAULT_OPTIONS);
+        assertAll(() -> assertNull(connectionStringDefault.getServerMonitoringMode()));
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {DEFAULT_OPTIONS + "serverMonitoringMode=stream"})
+    void equalAndHashCode(final String connectionString) {
+        ConnectionString default1 = new ConnectionString(DEFAULT_OPTIONS);
+        ConnectionString default2 = new ConnectionString(DEFAULT_OPTIONS);
+        ConnectionString actual1 = new ConnectionString(connectionString);
+        ConnectionString actual2 = new ConnectionString(connectionString);
+        assertAll(
+                () -> assertEquals(default1, default2),
+                () -> assertEquals(default1.hashCode(), default2.hashCode()),
+                () -> assertEquals(actual1, actual2),
+                () -> assertEquals(actual1.hashCode(), actual2.hashCode()),
+                () -> assertNotEquals(default1, actual1)
+        );
+    }
+
+    @Test
+    void serverMonitoringMode() {
+        assertAll(
+                () -> assertEquals(ServerMonitoringMode.POLL,
+                        new ConnectionString(DEFAULT_OPTIONS + "serverMonitoringMode=poll").getServerMonitoringMode()),
+                () -> assertThrows(IllegalArgumentException.class,
+                        () -> new ConnectionString(DEFAULT_OPTIONS + "serverMonitoringMode=invalid"))
+        );
+    }
+}
diff --git a/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsSpecification.groovy b/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsSpecification.groovy
index b92d8630f1..b11ed3a65a 100644
--- a/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsSpecification.groovy
+++ b/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsSpecification.groovy
@@ -24,6 +24,9 @@ import spock.lang.Specification
 import static java.util.concurrent.TimeUnit.MILLISECONDS
 import static java.util.concurrent.TimeUnit.SECONDS
 
+/**
+ * Update {@link ServerSettingsTest} instead.
+ */
 class ServerSettingsSpecification extends Specification {
     def 'should have correct defaults'() {
         when:
diff --git a/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsTest.java b/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsTest.java
new file mode 100644
index 0000000000..e8868813b0
--- /dev/null
+++ b/driver-core/src/test/unit/com/mongodb/connection/ServerSettingsTest.java
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.connection;
+
+import com.mongodb.ConnectionString;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+final class ServerSettingsTest {
+    private static final String DEFAULT_OPTIONS = "mongodb://localhost/?";
+
+    @Test
+    void defaults() {
+        ServerSettings defaultServerSettings = ServerSettings.builder().build();
+        assertAll(() -> assertEquals(ServerMonitoringMode.AUTO, defaultServerSettings.getServerMonitoringMode()));
+    }
+
+    @ParameterizedTest
+    @MethodSource("equalAndHashCodeArgs")
+    void equalAndHashCode(final ServerSettings.Builder serverSettingsBuilder) {
+        ServerSettings default1 = ServerSettings.builder().build();
+        ServerSettings default2 = ServerSettings.builder().build();
+        ServerSettings actual1 = serverSettingsBuilder.build();
+        ServerSettings actual2 = serverSettingsBuilder.build();
+        assertAll(
+                () -> assertEquals(default1, default2),
+                () -> assertEquals(default1.hashCode(), default2.hashCode()),
+                () -> assertEquals(actual1, actual2),
+                () -> assertEquals(actual1.hashCode(), actual2.hashCode()),
+                () -> assertNotEquals(default1, actual1)
+        );
+    }
+
+    private static Stream<Arguments> equalAndHashCodeArgs() {
+        return Stream.of(
+                Arguments.of(ServerSettings.builder().serverMonitoringMode(ServerMonitoringMode.POLL))
+        );
+    }
+
+    @Test
+    void serverMonitoringMode() {
+        assertAll(
+                () -> assertEquals(
+                        ServerMonitoringMode.POLL,
+                        ServerSettings.builder()
+                                .serverMonitoringMode(ServerMonitoringMode.POLL)
+                                .build()
+                                .getServerMonitoringMode(),
+                        "should set"),
+                () -> assertEquals(
+                        ServerMonitoringMode.STREAM,
+                        ServerSettings.builder()
+                                .applySettings(ServerSettings.builder()
+                                        .serverMonitoringMode(ServerMonitoringMode.STREAM)
+                                        .build())
+                                .build()
+                                .getServerMonitoringMode(),
+                        "should apply from settings"),
+                () -> assertEquals(
+                        ServerMonitoringMode.AUTO,
+                        ServerSettings.builder()
+                                .serverMonitoringMode(ServerMonitoringMode.STREAM)
+                                .applySettings(ServerSettings.builder()
+                                        .build())
+                                .build()
+                                .getServerMonitoringMode(),
+                        "should apply unset from settings"),
+                () -> assertEquals(
+                        ServerMonitoringMode.POLL,
+                        ServerSettings.builder()
+                                .applyConnectionString(new ConnectionString(DEFAULT_OPTIONS + "serverMonitoringMode=POLL"))
+                                .build()
+                                .getServerMonitoringMode(),
+                        "should apply from connection string"),
+                () -> assertEquals(
+                        ServerMonitoringMode.STREAM,
+                        ServerSettings.builder()
+                                .serverMonitoringMode(ServerMonitoringMode.STREAM)
+                                .applyConnectionString(new ConnectionString(DEFAULT_OPTIONS))
+                                .build()
+                                .getServerMonitoringMode(),
+                        "should not apply unset from connection string")
+        );
+    }
+}
diff --git a/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java b/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java
new file mode 100644
index 0000000000..b009b5094f
--- /dev/null
+++ b/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.event;
+
+import com.mongodb.annotations.ThreadSafe;
+import com.mongodb.lang.Nullable;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static com.mongodb.assertions.Assertions.assertNotNull;
+import static com.mongodb.assertions.Assertions.assertTrue;
+import static java.util.Collections.unmodifiableSet;
+import static java.util.stream.StreamSupport.stream;
+
+@ThreadSafe
+public final class TestServerMonitorListener implements ServerMonitorListener {
+    private final Set<Class<?>> listenableEventTypes;
+    private final Lock lock;
+    private final Condition condition;
+    private final List<Object> events;
+
+    public TestServerMonitorListener(final Iterable<String> listenableEventTypes) {
+        this.listenableEventTypes = unmodifiableSet(stream(listenableEventTypes.spliterator(), false)
+                .map(TestServerMonitorListener::nullableEventType)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toSet()));
+        lock = new ReentrantLock();
+        condition = lock.newCondition();
+        events = new ArrayList<>();
+    }
+
+    public void serverHearbeatStarted(final ServerHeartbeatStartedEvent event) {
+        register(event);
+    }
+
+    public void serverHeartbeatSucceeded(final ServerHeartbeatSucceededEvent event) {
+        register(event);
+    }
+
+    public void serverHeartbeatFailed(final ServerHeartbeatFailedEvent event) {
+        register(event);
+    }
+
+    public <T> void waitForEvents(final Class<T> type, final Predicate<? super T> matcher, final int count, final Duration duration)
+            throws InterruptedException, TimeoutException {
+        assertTrue(listenable(type));
+        long remainingNanos = duration.toNanos();
+        lock.lock();
+        try {
+            long observedCount = countEvents(type, matcher);
+            while (observedCount < count) {
+                if (remainingNanos <= 0) {
+                    throw new TimeoutException(String.format("Timed out waiting for %d %s events. The observed count is %d.",
+                            count, type.getSimpleName(), observedCount));
+                }
+                remainingNanos = condition.awaitNanos(remainingNanos);
+                observedCount = countEvents(type, matcher);
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public <T> long countEvents(final Class<T> type, final Predicate<? super T> matcher) {
+        assertTrue(listenable(type));
+        lock.lock();
+        try {
+            return events.stream()
+                    .filter(type::isInstance)
+                    .map(type::cast)
+                    .filter(matcher)
+                    .count();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public List<Object> getEvents() {
+        lock.lock();
+        try {
+            return new ArrayList<>(events);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public static Class<?> eventType(final String eventType) {
+        return assertNotNull(nullableEventType(eventType));
+    }
+
+    @Nullable
+    private static Class<?> nullableEventType(final String eventType) {
+        switch (eventType) {
+            case "serverHeartbeatStartedEvent": {
+                return ServerHeartbeatStartedEvent.class;
+            }
+            case "serverHeartbeatSucceededEvent": {
+                return ServerHeartbeatSucceededEvent.class;
+            }
+            case "serverHeartbeatFailedEvent": {
+                return ServerHeartbeatFailedEvent.class;
+            }
+            default: {
+                return null;
+            }
+        }
+    }
+
+    private boolean listenable(final Class<?> eventType) {
+        return listenableEventTypes.contains(eventType);
+    }
+
+    private void register(final Object event) {
+        if (!listenable(event.getClass())) {
+            return;
+        }
+        lock.lock();
+        try {
+            events.add(event);
+            condition.signalAll();
+        } finally {
+            lock.unlock();
+        }
+    }
+}
diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/AsynchronousClusterEventListenerTest.java b/driver-core/src/test/unit/com/mongodb/internal/connection/AsynchronousClusterEventListenerTest.java
index 2403f3f324..09d2fa864a 100644
--- a/driver-core/src/test/unit/com/mongodb/internal/connection/AsynchronousClusterEventListenerTest.java
+++ b/driver-core/src/test/unit/com/mongodb/internal/connection/AsynchronousClusterEventListenerTest.java
@@ -72,7 +72,7 @@ public void testEventsPublished() throws InterruptedException {
         listener.clusterDescriptionChanged(clusterDescriptionChangedEvent);
         assertEquals(clusterDescriptionChangedEvent, targetListener.take());
 
-        ServerHeartbeatStartedEvent serverHeartbeatStartedEvent = new ServerHeartbeatStartedEvent(connectionId);
+        ServerHeartbeatStartedEvent serverHeartbeatStartedEvent = new ServerHeartbeatStartedEvent(connectionId, false);
         listener.serverHearbeatStarted(serverHeartbeatStartedEvent);
         assertEquals(serverHeartbeatStartedEvent, targetListener.take());
 
diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/DefaultServerMonitorSpecification.groovy b/driver-core/src/test/unit/com/mongodb/internal/connection/DefaultServerMonitorSpecification.groovy
index 1e77995c21..42626a46d9 100644
--- a/driver-core/src/test/unit/com/mongodb/internal/connection/DefaultServerMonitorSpecification.groovy
+++ b/driver-core/src/test/unit/com/mongodb/internal/connection/DefaultServerMonitorSpecification.groovy
@@ -84,12 +84,12 @@ class DefaultServerMonitorSpecification extends Specification {
             }
         }
         monitor = new DefaultServerMonitor(new ServerId(new ClusterId(), new ServerAddress()), ServerSettings.builder().build(),
-                internalConnectionFactory, ClusterConnectionMode.SINGLE, null, SameObjectProvider.initialized(sdam))
+                internalConnectionFactory, ClusterConnectionMode.SINGLE, null, false, SameObjectProvider.initialized(sdam))
         monitor.start()
 
         when:
         monitor.close()
-        monitor.monitorThread.join()
+        monitor.monitor.join()
 
         then:
         !stateChanged
@@ -167,7 +167,7 @@ class DefaultServerMonitorSpecification extends Specification {
         }
         monitor = new DefaultServerMonitor(new ServerId(new ClusterId(), new ServerAddress()),
                 ServerSettings.builder().heartbeatFrequency(1, TimeUnit.SECONDS).addServerMonitorListener(serverMonitorListener).build(),
-                internalConnectionFactory, ClusterConnectionMode.SINGLE, null, mockSdamProvider())
+                internalConnectionFactory, ClusterConnectionMode.SINGLE, null, false, mockSdamProvider())
 
         when:
         monitor.start()
@@ -246,7 +246,7 @@ class DefaultServerMonitorSpecification extends Specification {
         }
         monitor = new DefaultServerMonitor(new ServerId(new ClusterId(), new ServerAddress()),
                 ServerSettings.builder().heartbeatFrequency(1, TimeUnit.SECONDS).addServerMonitorListener(serverMonitorListener).build(),
-                internalConnectionFactory, ClusterConnectionMode.SINGLE, null, mockSdamProvider())
+                internalConnectionFactory, ClusterConnectionMode.SINGLE, null, false, mockSdamProvider())
 
         when:
         monitor.start()
diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/ServerMonitoringModeUtilTest.java b/driver-core/src/test/unit/com/mongodb/internal/connection/ServerMonitoringModeUtilTest.java
new file mode 100644
index 0000000000..f549207b74
--- /dev/null
+++ b/driver-core/src/test/unit/com/mongodb/internal/connection/ServerMonitoringModeUtilTest.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.mongodb.internal.connection;
+
+import com.mongodb.connection.ServerMonitoringMode;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+final class ServerMonitoringModeUtilTest {
+    @Test
+    public void fromString() {
+        assertAll(
+                () -> assertEquals(ServerMonitoringMode.STREAM, ServerMonitoringModeUtil.fromString("stream")),
+                () -> assertEquals(ServerMonitoringMode.POLL, ServerMonitoringModeUtil.fromString("poll")),
+                () -> assertEquals(ServerMonitoringMode.AUTO, ServerMonitoringModeUtil.fromString("auto")),
+                () -> assertThrows(IllegalArgumentException.class, () -> ServerMonitoringModeUtil.fromString("invalid"))
+        );
+    }
+
+    @Test
+    public void getValue() {
+        assertAll(
+                () -> assertEquals("stream", ServerMonitoringModeUtil.getValue(ServerMonitoringMode.STREAM)),
+                () -> assertEquals("poll", ServerMonitoringModeUtil.getValue(ServerMonitoringMode.POLL)),
+                () -> assertEquals("auto", ServerMonitoringModeUtil.getValue(ServerMonitoringMode.AUTO))
+        );
+    }
+}
diff --git a/driver-lambda/build.gradle b/driver-lambda/build.gradle
index 76e293c9c1..d7b9928e8f 100644
--- a/driver-lambda/build.gradle
+++ b/driver-lambda/build.gradle
@@ -55,6 +55,8 @@ dependencies {
 
     implementation('com.amazonaws:aws-lambda-java-core:1.2.2')
     implementation('com.amazonaws:aws-lambda-java-events:3.11.1')
+    implementation(platform("org.junit:junit-bom:$junitBomVersion"))
+    implementation('org.junit.jupiter:junit-jupiter-api')
 }
 
 
diff --git a/driver-lambda/src/main/com/mongodb/lambdatest/LambdaTestApp.java b/driver-lambda/src/main/com/mongodb/lambdatest/LambdaTestApp.java
index c9a0d2ca99..c264337516 100644
--- a/driver-lambda/src/main/com/mongodb/lambdatest/LambdaTestApp.java
+++ b/driver-lambda/src/main/com/mongodb/lambdatest/LambdaTestApp.java
@@ -32,6 +32,7 @@
 import com.mongodb.event.ConnectionCreatedEvent;
 import com.mongodb.event.ConnectionPoolListener;
 import com.mongodb.event.ServerHeartbeatFailedEvent;
+import com.mongodb.event.ServerHeartbeatStartedEvent;
 import com.mongodb.event.ServerHeartbeatSucceededEvent;
 import com.mongodb.event.ServerMonitorListener;
 import com.mongodb.lang.NonNull;
@@ -45,8 +46,11 @@
 import java.io.StringWriter;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Test App for AWS lambda functions
@@ -58,6 +62,7 @@ public class LambdaTestApp implements RequestHandler<APIGatewayProxyRequestEvent
     private long totalHeartbeatDurationMs = 0;
     private long totalCommandCount = 0;
     private long totalCommandDurationMs = 0;
+    private final CopyOnWriteArrayList<Throwable> failedAssertions = new CopyOnWriteArrayList<>();
 
     public LambdaTestApp() {
         String connectionString = System.getenv("MONGODB_URI");
@@ -77,13 +82,20 @@ public void commandFailed(@NonNull final CommandFailedEvent event) {
                     }
                 })
                 .applyToServerSettings(builder -> builder.addServerMonitorListener(new ServerMonitorListener() {
+                    @Override
+                    public void serverHearbeatStarted(@NonNull final ServerHeartbeatStartedEvent event) {
+                        checkAssertion(() -> assertFalse(event.isAwaited(), event::toString));
+                    }
+
                     @Override
                     public void serverHeartbeatSucceeded(@NonNull final ServerHeartbeatSucceededEvent event) {
+                        checkAssertion(() -> assertFalse(event.isAwaited(), event::toString));
                         totalHeartbeatCount++;
                         totalHeartbeatDurationMs += event.getElapsedTime(MILLISECONDS);
                     }
                     @Override
                     public void serverHeartbeatFailed(@NonNull final ServerHeartbeatFailedEvent event) {
+                        checkAssertion(() -> assertFalse(event.isAwaited(), event::toString));
                         totalHeartbeatCount++;
                         totalHeartbeatDurationMs += event.getElapsedTime(MILLISECONDS);
                     }
@@ -110,6 +122,7 @@ public APIGatewayProxyResponseEvent handleRequest(final APIGatewayProxyRequestEv
             BsonValue id = collection.insertOne(new Document("n", 1)).getInsertedId();
             collection.deleteOne(new Document("_id", id));
 
+            assertTrue(failedAssertions.isEmpty(), failedAssertions.toString());
             BsonDocument responseBody = getBsonDocument();
 
             return templateResponse()
@@ -151,4 +164,12 @@ private APIGatewayProxyResponseEvent templateResponse() {
         return new APIGatewayProxyResponseEvent()
                 .withHeaders(headers);
     }
+
+    private void checkAssertion(final Runnable assertion) {
+        try {
+            assertion.run();
+        } catch (Throwable t) {
+            failedAssertions.add(t);
+        }
+    }
 }
diff --git a/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java b/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java
index 4ea7b43bb0..b32137abd4 100644
--- a/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java
+++ b/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java
@@ -19,6 +19,7 @@
 import com.mongodb.lang.Nullable;
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
+import org.junit.Before;
 import org.junit.runners.Parameterized;
 
 import java.io.IOException;
@@ -38,4 +39,9 @@ public UnifiedServerDiscoveryAndMonitoringTest(@SuppressWarnings("unused") final
     public static Collection<Object[]> data() throws URISyntaxException, IOException {
         return getTestData("unified-test-format/server-discovery-and-monitoring");
     }
+
+    @Before
+    public void before() {
+        com.mongodb.client.unified.UnifiedServerDiscoveryAndMonitoringTest.skipTests(getDefinition());
+    }
 }
diff --git a/driver-sync/src/test/functional/com/mongodb/client/unified/ContextElement.java b/driver-sync/src/test/functional/com/mongodb/client/unified/ContextElement.java
index 110cf321f4..8ffa35388e 100644
--- a/driver-sync/src/test/functional/com/mongodb/client/unified/ContextElement.java
+++ b/driver-sync/src/test/functional/com/mongodb/client/unified/ContextElement.java
@@ -23,6 +23,7 @@
 import com.mongodb.event.CommandSucceededEvent;
 import com.mongodb.internal.logging.LogMessage;
 import org.bson.BsonArray;
+import org.bson.BsonBoolean;
 import org.bson.BsonDocument;
 import org.bson.BsonString;
 import org.bson.BsonValue;
@@ -127,6 +128,48 @@ public static ContextElement ofClusterDescriptionChangedEventCount(final String
         return new EventCountContext("Cluster Description Changed Event Count", client, event, count);
     }
 
+    public static ContextElement ofWaitForServerMonitorEvents(final String client, final BsonDocument event, final int count) {
+        return new EventCountContext("Wait For Server Monitor Events", client, event, count);
+    }
+
+    public static ContextElement ofServerMonitorEventCount(final String client, final BsonDocument event, final int count) {
+        return new EventCountContext("Server Monitor Event Count", client, event, count);
+    }
+
+    public static ContextElement ofServerMonitorEvents(final String client, final BsonArray expectedEvents, final List<?> actualEvents) {
+        return new ContextElement() {
+            @Override
+            public String toString() {
+                return "Events MatchingContext: \n"
+                        + "   client: '" + client + "'\n"
+                        + "   Expected events:\n"
+                        + new BsonDocument("events", expectedEvents).toJson(JsonWriterSettings.builder().indent(true).build()) + "\n"
+                        + "   Actual events:\n"
+                        + new BsonDocument("events",
+                                new BsonArray(actualEvents.stream().map(ContextElement::serverMonitorEventToDocument).collect(Collectors.toList())))
+                                        .toJson(JsonWriterSettings.builder().indent(true).build())
+                        + "\n";
+            }
+
+            private BsonDocument toDocument(final Object event) {
+                return new BsonDocument(EventMatcher.getEventType(event.getClass()),
+                        new BsonDocument("awaited", BsonBoolean.valueOf(EventMatcher.getAwaitedFromServerMonitorEvent(event))));
+            }
+        };
+    }
+
+    public static ContextElement ofServerMonitorEvent(final BsonDocument expected, final Object actual, final int eventPosition) {
+        return new ContextElement() {
+            @Override
+            public String toString() {
+                return "Event Matching Context\n"
+                        + "   event position: " + eventPosition + "\n"
+                        + "   expected event: " + expected + "\n"
+                        + "   actual event:   " + serverMonitorEventToDocument(actual) + "\n";
+            }
+        };
+    }
+
     private static class EventCountContext extends ContextElement {
 
         private final String name;
@@ -417,4 +460,9 @@ public String toString() {
     private static BsonDocument connectionPoolEventToDocument(final Object event) {
         return new BsonDocument(event.getClass().getSimpleName(), new BsonDocument());
     }
+
+    private static BsonDocument serverMonitorEventToDocument(final Object event) {
+        return new BsonDocument(EventMatcher.getEventType(event.getClass()),
+                new BsonDocument("awaited", BsonBoolean.valueOf(EventMatcher.getAwaitedFromServerMonitorEvent(event))));
+    }
 }
diff --git a/driver-sync/src/test/functional/com/mongodb/client/unified/Entities.java b/driver-sync/src/test/functional/com/mongodb/client/unified/Entities.java
index 4399b6e386..4845ac460a 100644
--- a/driver-sync/src/test/functional/com/mongodb/client/unified/Entities.java
+++ b/driver-sync/src/test/functional/com/mongodb/client/unified/Entities.java
@@ -24,6 +24,10 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.ServerApi;
 import com.mongodb.ServerApiVersion;
+import com.mongodb.event.TestServerMonitorListener;
+import com.mongodb.internal.connection.ServerMonitoringModeUtil;
+import com.mongodb.internal.connection.TestClusterListener;
+import com.mongodb.logging.TestLoggingInterceptor;
 import com.mongodb.TransactionOptions;
 import com.mongodb.WriteConcern;
 import com.mongodb.assertions.Assertions;
@@ -55,13 +59,11 @@
 import com.mongodb.event.ConnectionPoolListener;
 import com.mongodb.event.ConnectionPoolReadyEvent;
 import com.mongodb.event.ConnectionReadyEvent;
-import com.mongodb.internal.connection.TestClusterListener;
 import com.mongodb.internal.connection.TestCommandListener;
 import com.mongodb.internal.connection.TestConnectionPoolListener;
 import com.mongodb.internal.connection.TestServerListener;
 import com.mongodb.internal.logging.LogMessage;
 import com.mongodb.lang.NonNull;
-import com.mongodb.logging.TestLoggingInterceptor;
 import org.bson.BsonArray;
 import org.bson.BsonBoolean;
 import org.bson.BsonDocument;
@@ -121,6 +123,7 @@ public final class Entities {
     private final Map<String, TestConnectionPoolListener> clientConnectionPoolListeners = new HashMap<>();
     private final Map<String, TestServerListener> clientServerListeners = new HashMap<>();
     private final Map<String, TestClusterListener> clientClusterListeners = new HashMap<>();
+    private final Map<String, TestServerMonitorListener> serverMonitorListeners = new HashMap<>();
     private final Map<String, MongoCursor<BsonDocument>> cursors = new HashMap<>();
     private final Map<String, ClusterDescription> topologyDescriptions = new HashMap<>();
     private final Map<String, Long> successCounts = new HashMap<>();
@@ -285,6 +288,10 @@ public TestClusterListener getClusterListener(final String id) {
         return getEntity(id + "-cluster-listener", clientClusterListeners, "cluster listener");
     }
 
+    public TestServerMonitorListener getServerMonitorListener(final String id) {
+        return getEntity(id + "-server-monitor-listener", serverMonitorListeners, "server monitor listener");
+    }
+
     private <T> T getEntity(final String id, final Map<String, T> entities, final String type) {
         T entity = entities.get(id);
         if (entity == null) {
@@ -378,23 +385,25 @@ private void initClient(final BsonDocument entity, final String id,
         putEntity(id + "-cluster-listener", testClusterListener, clientClusterListeners);
 
         if (entity.containsKey("observeEvents")) {
+            List<String> observeEvents = entity.getArray("observeEvents").stream()
+                    .map(type -> type.asString().getValue()).collect(Collectors.toList());
             List<String> ignoreCommandMonitoringEvents = entity
                     .getArray("ignoreCommandMonitoringEvents", new BsonArray()).stream()
                     .map(type -> type.asString().getValue()).collect(Collectors.toList());
             ignoreCommandMonitoringEvents.add("configureFailPoint");
-            TestCommandListener testCommandListener = new TestCommandListener(
-                    entity.getArray("observeEvents").stream()
-                            .map(type -> type.asString().getValue()).collect(Collectors.toList()),
+            TestCommandListener testCommandListener = new TestCommandListener(observeEvents,
                     ignoreCommandMonitoringEvents, entity.getBoolean("observeSensitiveCommands", BsonBoolean.FALSE).getValue());
             clientSettingsBuilder.addCommandListener(testCommandListener);
             putEntity(id + "-command-listener", testCommandListener, clientCommandListeners);
 
-            TestConnectionPoolListener testConnectionPoolListener = new TestConnectionPoolListener(
-                    entity.getArray("observeEvents").stream()
-                            .map(type -> type.asString().getValue()).collect(Collectors.toList()));
+            TestConnectionPoolListener testConnectionPoolListener = new TestConnectionPoolListener(observeEvents);
             clientSettingsBuilder.applyToConnectionPoolSettings(builder ->
                     builder.addConnectionPoolListener(testConnectionPoolListener));
             putEntity(id + "-connection-pool-listener", testConnectionPoolListener, clientConnectionPoolListeners);
+
+            TestServerMonitorListener testServerMonitorListener = new TestServerMonitorListener(observeEvents);
+            clientSettingsBuilder.applyToServerSettings(builder -> builder.addServerMonitorListener(testServerMonitorListener));
+            putEntity(id + "-server-monitor-listener", testServerMonitorListener, serverMonitorListeners);
         } else {
             // Regardless of whether events are observed, we still need to track some info about the pool in order to implement
             // the assertNumberConnectionsCheckedOut operation
@@ -503,6 +512,10 @@ private void initClient(final BsonDocument entity, final String id,
                     case "appName":
                         clientSettingsBuilder.applicationName(value.asString().getValue());
                         break;
+                    case "serverMonitoringMode":
+                        clientSettingsBuilder.applyToServerSettings(builder -> builder.serverMonitoringMode(
+                                ServerMonitoringModeUtil.fromString(value.asString().getValue())));
+                        break;
                     default:
                         throw new UnsupportedOperationException("Unsupported uri option: " + key);
                 }
diff --git a/driver-sync/src/test/functional/com/mongodb/client/unified/EventMatcher.java b/driver-sync/src/test/functional/com/mongodb/client/unified/EventMatcher.java
index e831082597..41ada275a6 100644
--- a/driver-sync/src/test/functional/com/mongodb/client/unified/EventMatcher.java
+++ b/driver-sync/src/test/functional/com/mongodb/client/unified/EventMatcher.java
@@ -16,6 +16,7 @@
 
 package com.mongodb.client.unified;
 
+import com.mongodb.assertions.Assertions;
 import com.mongodb.connection.ServerType;
 import com.mongodb.event.ClusterDescriptionChangedEvent;
 import com.mongodb.event.CommandEvent;
@@ -29,10 +30,15 @@
 import com.mongodb.event.ConnectionPoolReadyEvent;
 import com.mongodb.event.ConnectionReadyEvent;
 import com.mongodb.event.ServerDescriptionChangedEvent;
+import com.mongodb.event.ServerHeartbeatFailedEvent;
+import com.mongodb.event.ServerHeartbeatStartedEvent;
+import com.mongodb.event.ServerHeartbeatSucceededEvent;
+import com.mongodb.event.TestServerMonitorListener;
 import com.mongodb.internal.connection.TestClusterListener;
 import com.mongodb.internal.connection.TestConnectionPoolListener;
 import com.mongodb.internal.connection.TestServerListener;
 import com.mongodb.lang.NonNull;
+import com.mongodb.lang.Nullable;
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
 import org.bson.types.ObjectId;
@@ -284,9 +290,61 @@ public void assertClusterDescriptionChangeEventCount(final String client, final
         context.pop();
     }
 
+    public <T> void waitForServerMonitorEvents(final String client, final Class<T> expectedEventType, final BsonDocument expectedEvent,
+            final int count, final TestServerMonitorListener serverMonitorListener) {
+        context.push(ContextElement.ofWaitForServerMonitorEvents(client, expectedEvent, count));
+        BsonDocument expectedEventContents = getEventContents(expectedEvent);
+        try {
+            serverMonitorListener.waitForEvents(expectedEventType,
+                    event -> serverMonitorEventMatches(expectedEventContents, event, null), count, Duration.ofSeconds(10));
+            context.pop();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } catch (TimeoutException e) {
+            fail(context.getMessage("Timed out waiting for server monitor events"));
+        }
+    }
+
+    public <T> void assertServerMonitorEventCount(final String client, final Class<T> expectedEventType, final BsonDocument expectedEvent,
+            final int count, final TestServerMonitorListener serverMonitorListener) {
+        BsonDocument expectedEventContents = getEventContents(expectedEvent);
+        context.push(ContextElement.ofServerMonitorEventCount(client, expectedEvent, count));
+        long matchCount = serverMonitorListener.countEvents(expectedEventType, event ->
+                serverMonitorEventMatches(expectedEventContents, event, null));
+        assertEquals(context.getMessage("Expected server monitor event counts to match"), count, matchCount);
+        context.pop();
+    }
+
+    public void assertServerMonitorEventsEquality(
+            final String client,
+            final boolean ignoreExtraEvents,
+            final BsonArray expectedEventDocuments,
+            final List<?> events) {
+        context.push(ContextElement.ofServerMonitorEvents(client, expectedEventDocuments, events));
+        if (ignoreExtraEvents) {
+            assertTrue(context.getMessage("Number of events must be greater than or equal to the expected number of events"),
+                    events.size() >= expectedEventDocuments.size());
+        } else {
+            assertEquals(context.getMessage("Number of events must be the same"), expectedEventDocuments.size(), events.size());
+        }
+        for (int i = 0; i < expectedEventDocuments.size(); i++) {
+            Object actualEvent = events.get(i);
+            BsonDocument expectedEventDocument = expectedEventDocuments.get(i).asDocument();
+            String expectedEventType = expectedEventDocument.getFirstKey();
+            context.push(ContextElement.ofServerMonitorEvent(expectedEventDocument, actualEvent, i));
+            assertEquals(context.getMessage("Expected event type to match"), expectedEventType, getEventType(actualEvent.getClass()));
+            BsonDocument expectedEventContents = expectedEventDocument.getDocument(expectedEventType);
+            serverMonitorEventMatches(expectedEventContents, actualEvent, context);
+            context.pop();
+        }
+        context.pop();
+    }
+
     @NonNull
     private BsonDocument getEventContents(final BsonDocument expectedEvent) {
-        HashSet<String> supportedEventTypes = new HashSet<>(asList("serverDescriptionChangedEvent", "topologyDescriptionChangedEvent"));
+        HashSet<String> supportedEventTypes = new HashSet<>(asList(
+                "serverDescriptionChangedEvent", "topologyDescriptionChangedEvent",
+                "serverHeartbeatStartedEvent", "serverHeartbeatSucceededEvent", "serverHeartbeatFailedEvent"));
         String expectedEventType = expectedEvent.getFirstKey();
         if (!supportedEventTypes.contains(expectedEventType)) {
             throw new UnsupportedOperationException("Unsupported event type " + expectedEventType);
@@ -333,12 +391,52 @@ private static boolean clusterDescriptionChangedEventMatches(final BsonDocument
         return true;
     }
 
-    private static String getEventType(final Class<?> eventClass) {
+    /**
+     * @param context Not {@code null} iff mismatch must result in an error, that is, this method works as an assertion.
+     */
+    private static <T> boolean serverMonitorEventMatches(
+            final BsonDocument expectedEventContents,
+            final T event,
+            @Nullable final AssertionContext context) {
+        if (expectedEventContents.size() > 1) {
+            throw new UnsupportedOperationException("Matching for the following event is not implemented " + expectedEventContents.toJson());
+        }
+        if (expectedEventContents.containsKey("awaited")) {
+            boolean expectedAwaited = expectedEventContents.getBoolean("awaited").getValue();
+            boolean actualAwaited = getAwaitedFromServerMonitorEvent(event);
+            boolean awaitedMatches = expectedAwaited == actualAwaited;
+            if (context != null) {
+                assertTrue(context.getMessage("Expected `awaited` to match"), awaitedMatches);
+            }
+            return awaitedMatches;
+        }
+        return true;
+    }
+
+    static boolean getAwaitedFromServerMonitorEvent(final Object event) {
+        if (event instanceof ServerHeartbeatStartedEvent) {
+            return ((ServerHeartbeatStartedEvent) event).isAwaited();
+        } else if (event instanceof ServerHeartbeatSucceededEvent) {
+            return ((ServerHeartbeatSucceededEvent) event).isAwaited();
+        } else if (event instanceof ServerHeartbeatFailedEvent) {
+            return ((ServerHeartbeatFailedEvent) event).isAwaited();
+        } else {
+            throw Assertions.fail(event.toString());
+        }
+    }
+
+    static String getEventType(final Class<?> eventClass) {
         String eventClassName = eventClass.getSimpleName();
         if (eventClassName.startsWith("ConnectionPool")) {
             return eventClassName.replace("ConnectionPool", "pool");
-        } else {
+        } else if (eventClassName.startsWith("Connection")) {
             return eventClassName.replace("Connection", "connection");
+        } else if (eventClassName.startsWith("ServerHeartbeat")) {
+            StringBuilder eventTypeBuilder = new StringBuilder(eventClassName);
+            eventTypeBuilder.setCharAt(0, Character.toLowerCase(eventTypeBuilder.charAt(0)));
+            return eventTypeBuilder.toString();
+        } else {
+            throw new UnsupportedOperationException(eventClassName);
         }
     }
 
diff --git a/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java b/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java
index c01bdca845..7f2b4bce60 100644
--- a/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java
+++ b/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java
@@ -19,12 +19,16 @@
 import com.mongodb.lang.Nullable;
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
+import org.bson.BsonString;
+import org.junit.Before;
 import org.junit.runners.Parameterized;
 
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.util.Collection;
 
+import static org.junit.Assume.assumeFalse;
+
 public class UnifiedServerDiscoveryAndMonitoringTest extends UnifiedSyncTest {
 
     public UnifiedServerDiscoveryAndMonitoringTest(@SuppressWarnings("unused") final String fileDescription,
@@ -38,4 +42,17 @@ public UnifiedServerDiscoveryAndMonitoringTest(@SuppressWarnings("unused") final
     public static Collection<Object[]> data() throws URISyntaxException, IOException {
         return getTestData("unified-test-format/server-discovery-and-monitoring");
     }
+
+    @Before
+    public void before() {
+        skipTests(getDefinition());
+    }
+
+    public static void skipTests(final BsonDocument definition) {
+        String description = definition.getString("description", new BsonString("")).getValue();
+        assumeFalse("Skipping because our server monitoring events behave differently for now",
+                description.equals("connect with serverMonitoringMode=auto >=4.4"));
+        assumeFalse("Skipping because our server monitoring events behave differently for now",
+                description.equals("connect with serverMonitoringMode=stream >=4.4"));
+    }
 }
diff --git a/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java b/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java
index 8c3d41e00a..a7cb9793c4 100644
--- a/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java
+++ b/driver-sync/src/test/functional/com/mongodb/client/unified/UnifiedTest.java
@@ -21,6 +21,7 @@
 import com.mongodb.MongoNamespace;
 import com.mongodb.ReadPreference;
 import com.mongodb.UnixServerAddress;
+import com.mongodb.event.TestServerMonitorListener;
 import com.mongodb.internal.logging.LogMessage;
 import com.mongodb.logging.TestLoggingInterceptor;
 import com.mongodb.WriteConcern;
@@ -178,6 +179,10 @@ private static Object[] createTestData(final BsonDocument fileDocument, final Bs
                 testDocument};
     }
 
+    protected BsonDocument getDefinition() {
+        return definition;
+    }
+
     protected abstract MongoClient createMongoClient(MongoClientSettings settings);
 
     protected abstract GridFSBucket createGridFSBucket(MongoDatabase database);
@@ -202,7 +207,9 @@ public void setUp() {
                         || schemaVersion.equals("1.12")
                         || schemaVersion.equals("1.13")
                         || schemaVersion.equals("1.14")
-                        || schemaVersion.equals("1.15"));
+                        || schemaVersion.equals("1.15")
+                        || schemaVersion.equals("1.16")
+                        || schemaVersion.equals("1.17"));
         if (runOnRequirements != null) {
             assumeTrue("Run-on requirements not met",
                     runOnRequirementsMet(runOnRequirements, getMongoClientSettings(), getServerVersion()));
@@ -269,14 +276,18 @@ private void compareEvents(final UnifiedTestContext context, final BsonDocument
             String client = curClientEvents.getString("client").getValue();
             boolean ignoreExtraEvents = curClientEvents.getBoolean("ignoreExtraEvents", BsonBoolean.FALSE).getValue();
             String eventType = curClientEvents.getString("eventType", new BsonString("command")).getValue();
+            BsonArray expectedEvents = curClientEvents.getArray("events");
             if (eventType.equals("command")) {
                 TestCommandListener listener = entities.getClientCommandListener(client);
-                context.getEventMatcher().assertCommandEventsEquality(client, ignoreExtraEvents, curClientEvents.getArray("events"),
+                context.getEventMatcher().assertCommandEventsEquality(client, ignoreExtraEvents, expectedEvents,
                         listener.getEvents());
             } else if (eventType.equals("cmap")) {
                 TestConnectionPoolListener listener = entities.getConnectionPoolListener(client);
-                context.getEventMatcher().assertConnectionPoolEventsEquality(client, ignoreExtraEvents, curClientEvents.getArray("events"),
+                context.getEventMatcher().assertConnectionPoolEventsEquality(client, ignoreExtraEvents, expectedEvents,
                         listener.getEvents());
+            } else if (eventType.equals("sdam")) {
+                TestServerMonitorListener listener = entities.getServerMonitorListener(client);
+                context.getEventMatcher().assertServerMonitorEventsEquality(client, ignoreExtraEvents, expectedEvents, listener.getEvents());
             } else {
                 throw new UnsupportedOperationException("Unexpected event type: " + eventType);
             }
@@ -621,6 +632,12 @@ private OperationResult executeWaitForEvent(final UnifiedTestContext context, fi
             case "connectionReadyEvent":
                 context.getEventMatcher().waitForConnectionPoolEvents(clientId, event, count, entities.getConnectionPoolListener(clientId));
                 break;
+            case "serverHeartbeatStartedEvent":
+            case "serverHeartbeatSucceededEvent":
+            case "serverHeartbeatFailedEvent":
+                context.getEventMatcher().waitForServerMonitorEvents(clientId, TestServerMonitorListener.eventType(eventName), event, count,
+                        entities.getServerMonitorListener(clientId));
+                break;
             default:
                 throw new UnsupportedOperationException("Unsupported event: " + eventName);
         }
@@ -649,6 +666,12 @@ private OperationResult executeAssertEventCount(final UnifiedTestContext context
                 context.getEventMatcher().assertConnectionPoolEventCount(clientId, event, count,
                         entities.getConnectionPoolListener(clientId).getEvents());
                 break;
+            case "serverHeartbeatStartedEvent":
+            case "serverHeartbeatSucceededEvent":
+            case "serverHeartbeatFailedEvent":
+                context.getEventMatcher().assertServerMonitorEventCount(clientId, TestServerMonitorListener.eventType(eventName), event, count,
+                        entities.getServerMonitorListener(clientId));
+                break;
             default:
                 throw new UnsupportedOperationException("Unsupported event: " + eventName);
         }