RUST-952 Emit SDAM events for load balanced topologies (#470)

Author: abr-egn

src/sdam/description/server.rs

@@ -177,20 +177,6 @@ impl ServerDescription {
         description
     }
 
-    pub(crate) fn new_load_balancer(mut address: ServerAddress) -> Self {
-        address = ServerAddress::Tcp {
-            host: address.host().to_lowercase(),
-            port: address.port(),
-        };
-        Self {
-            address,
-            server_type: ServerType::LoadBalancer,
-            last_update_time: None,
-            reply: Ok(None),
-            average_round_trip_time: None,
-        }
-    }
-
     /// Whether this server is "available" as per the definition in the server selection spec.
     pub(crate) fn is_available(&self) -> bool {
         self.server_type.is_available()

src/sdam/description/topology/mod.rs

@@ -140,15 +140,7 @@ impl TopologyDescription {
         let servers: HashMap<_, _> = options
             .hosts
             .into_iter()
-            .map(|address| {
-                let description = if topology_type == TopologyType::LoadBalanced {
-                    ServerDescription::new_load_balancer(address.clone())
-                } else {
-                    ServerDescription::new(address.clone(), None)
-                };
-
-                (address, description)
-            })
+            .map(|address| (address.clone(), ServerDescription::new(address, None)))
             .collect();
 
         let session_support_status = if topology_type == TopologyType::LoadBalanced {

src/sdam/state/mod.rs

@@ -204,11 +204,6 @@ impl Topology {
             topology_state.add_new_server(address, options.clone(), &topology.downgrade());
         }
 
-        // When the client is in load balanced mode, it doesn't poll for changes in the SRV record.
-        if !is_load_balanced {
-            SrvPollingMonitor::start(topology.downgrade());
-        }
-
         if let Some(ref handler) = options.sdam_event_handler {
             let event = TopologyDescriptionChangedEvent {
                 topology_id: id,
@@ -226,7 +221,23 @@ impl Topology {
             }
         }
 
-        SrvPollingMonitor::start(topology.downgrade());
+        if is_load_balanced {
+            for server_address in &options.hosts {
+                // Load-balanced clients don't have a heartbeat monitor, so we synthesize
+                // updating the server to `ServerType::LoadBalancer`.
+                let new_desc = ServerDescription {
+                    server_type: ServerType::LoadBalancer,
+                    ..ServerDescription::new(server_address.clone(), None)
+                };
+                topology_state
+                    .update(new_desc, &options, topology.downgrade())
+                    .map_err(Error::internal)?;
+            }
+        } else {
+            // When the client is in load balanced mode, it doesn't poll for changes in the SRV
+            // record.
+            SrvPollingMonitor::start(topology.downgrade());
+        }
 
         drop(topology_state);
         Ok(topology)
@@ -235,6 +246,15 @@ impl Topology {
     pub(crate) fn close(&self) {
         self.common.is_alive.store(false, Ordering::SeqCst);
         if let Some(ref handler) = self.common.options.sdam_event_handler {
+            if self.common.options.load_balanced.unwrap_or(false) {
+                for host in &self.common.options.hosts {
+                    let event = ServerClosedEvent {
+                        address: host.clone(),
+                        topology_id: self.common.id,
+                    };
+                    handler.handle_server_closed_event(event);
+                }
+            }
             let event = TopologyClosedEvent {
                 topology_id: self.common.id,
             };