RUST-926 Allow direct connections to hidden nodes (#423)

Author: patrickfreed

src/client/executor.rs

@@ -31,6 +31,7 @@ use crate::{
         SelectedServer,
         ServerType,
         SessionSupportStatus,
+        TopologyType,
         TransactionSupportStatus,
     },
     selection_criteria::ReadPreference,
@@ -573,6 +574,17 @@ impl Client {
         }
     }
 
+    async fn select_data_bearing_server(&self) -> Result<()> {
+        let topology_type = self.inner.topology.topology_type().await;
+        let criteria = SelectionCriteria::Predicate(Arc::new(move |server_info| {
+            let server_type = server_info.server_type();
+            (matches!(topology_type, TopologyType::Single) && server_type.is_available())
+                || server_type.is_data_bearing()
+        }));
+        let _: SelectedServer = self.select_server(Some(&criteria)).await?;
+        Ok(())
+    }
+
     /// Gets whether the topology supports sessions, and if so, returns the topology's logical
     /// session timeout. If it has yet to be determined if the topology supports sessions, this
     /// method will perform a server selection that will force that determination to be made.
@@ -583,10 +595,7 @@ impl Client {
         // sessions are supported or not.
         match initial_status {
             SessionSupportStatus::Undetermined => {
-                let criteria = SelectionCriteria::Predicate(Arc::new(move |server_info| {
-                    server_info.server_type().is_data_bearing()
-                }));
-                let _: SelectedServer = self.select_server(Some(&criteria)).await?;
+                self.select_data_bearing_server().await?;
                 Ok(self.inner.topology.session_support_status().await)
             }
             _ => Ok(initial_status),
@@ -603,10 +612,7 @@ impl Client {
         // sessions are supported or not.
         match initial_status {
             TransactionSupportStatus::Undetermined => {
-                let criteria = SelectionCriteria::Predicate(Arc::new(move |server_info| {
-                    server_info.server_type().is_data_bearing()
-                }));
-                let _: SelectedServer = self.select_server(Some(&criteria)).await?;
+                self.select_data_bearing_server().await?;
                 Ok(self.inner.topology.transaction_support_status().await)
             }
             _ => Ok(initial_status),

src/client/options/mod.rs

@@ -370,7 +370,7 @@ pub struct ClientOptions {
     ///
     /// Note that by default, the driver will autodiscover other nodes in the cluster. To connect
     /// directly to a single server (rather than autodiscovering the rest of the cluster), set the
-    /// `direct` field to `true`.
+    /// `direct_connection` field to `true`.
     #[builder(default_code = "vec![ServerAddress::Tcp {
         host: \"localhost\".to_string(),
         port: Some(27017),

src/sdam/description/server.rs

@@ -47,6 +47,10 @@ impl ServerType {
                 | ServerType::LoadBalancer
         )
     }
+
+    pub(crate) fn is_available(self) -> bool {
+        !matches!(self, ServerType::Unknown)
+    }
 }
 
 impl Default for ServerType {
@@ -185,7 +189,7 @@ impl ServerDescription {
 
     /// Whether this server is "available" as per the definition in the server selection spec.
     pub(crate) fn is_available(&self) -> bool {
-        !matches!(self.server_type, ServerType::Unknown)
+        self.server_type.is_available()
     }
 
     pub(crate) fn compatibility_error_message(&self) -> Option<String> {

src/sdam/description/topology/mod.rs

@@ -272,6 +272,11 @@ impl TopologyDescription {
     /// Updates the topology's logical session timeout value based on the server's value for it.
     fn update_session_support_status(&mut self, server_description: &ServerDescription) {
         if !server_description.server_type.is_data_bearing() {
+            if let TopologyType::Single = self.topology_type {
+                self.session_support_status = SessionSupportStatus::Unsupported {
+                    logical_session_timeout: None,
+                };
+            }
             return;
         }
 

src/sdam/mod.rs

@@ -17,6 +17,7 @@ pub(crate) use self::{
         server_selection::SelectedServer,
         SessionSupportStatus,
         TopologyDescription,
+        TopologyType,
         TransactionSupportStatus,
     },
     message_manager::TopologyMessageManager,

src/sdam/state/mod.rs

@@ -440,7 +440,7 @@ impl Topology {
             .transaction_support_status()
     }
 
-    pub(super) async fn topology_type(&self) -> TopologyType {
+    pub(crate) async fn topology_type(&self) -> TopologyType {
         self.state.read().await.description.topology_type()
     }