plumbing

Author: abr-egn

src/client.rs

@@ -446,7 +446,7 @@ impl Client {
         &self,
         criteria: Option<&SelectionCriteria>,
     ) -> Result<ServerAddress> {
-        let server = self
+        let (server, _) = self
             .select_server(criteria, "Test select server", None)
             .await?;
         Ok(server.address.clone())
@@ -460,7 +460,7 @@ impl Client {
         #[allow(unused_variables)] // we only use the operation_name for tracing.
         operation_name: &str,
         deprioritized: Option<&ServerAddress>,
-    ) -> Result<SelectedServer> {
+    ) -> Result<(SelectedServer, SelectionCriteria)> {
         let criteria =
             criteria.unwrap_or(&SelectionCriteria::ReadPreference(ReadPreference::Primary));
 
@@ -488,9 +488,12 @@ impl Client {
         let mut watcher = self.inner.topology.watch();
         loop {
             let state = watcher.observe_latest();
-
+            for server in state.description.servers.values() {
+                eprintln!("at selection: {:?}", server.hello_response());
+            }
+            let effective_criteria = criteria; // TODO
             let result = server_selection::attempt_to_select_server(
-                criteria,
+                effective_criteria,
                 &state.description,
                 &state.servers(),
                 deprioritized,
@@ -507,7 +510,7 @@ impl Client {
                         #[cfg(feature = "tracing-unstable")]
                         event_emitter.emit_succeeded_event(&state.description, &server);
 
-                        return Ok(server);
+                        return Ok((server, effective_criteria.clone()));
                     } else {
                         #[cfg(feature = "tracing-unstable")]
                         if !emitted_waiting_message {

src/client/executor.rs

@@ -29,6 +29,7 @@ use crate::{
             wire::{next_request_id, Message},
             PinnedConnectionHandle,
         },
+        Command,
         ConnectionPool,
         RawCommandResponse,
         StreamDescription,
@@ -59,7 +60,7 @@ use crate::{
         Retryability,
     },
     options::{ChangeStreamOptions, SelectionCriteria},
-    sdam::{HandshakePhase, SelectedServer, ServerType, TopologyType, TransactionSupportStatus},
+    sdam::{HandshakePhase, ServerType, TopologyType, TransactionSupportStatus},
     selection_criteria::ReadPreference,
     tracking_arc::TrackingArc,
     ClusterTime,
@@ -309,7 +310,7 @@ impl Client {
         let mut retry: Option<ExecutionRetry> = None;
         let mut implicit_session: Option<ClientSession> = None;
         loop {
-            op.update_for_topology(&self.inner.topology.description());
+            //op.update_for_topology(&self.inner.topology.description());
 
             if retry.is_some() {
                 op.update_for_retry();
@@ -320,15 +321,15 @@ impl Client {
                 .and_then(|s| s.transaction.pinned_mongos())
                 .or_else(|| op.selection_criteria());
 
-            let server = match self
+            let (server, effective_criteria) = match self
                 .select_server(
                     selection_criteria,
                     op.name(),
                     retry.as_ref().map(|r| &r.first_server),
                 )
                 .await
             {
-                Ok(server) => server,
+                Ok(out) => out,
                 Err(mut err) => {
                     retry.first_error()?;
 
@@ -394,13 +395,7 @@ impl Client {
                 };
 
             let details = match self
-                .execute_operation_on_connection(
-                    op,
-                    &mut conn,
-                    &mut session,
-                    txn_number,
-                    retryability,
-                )
+                .execute_command_on_connection(cmd, op, &mut conn, &mut session, retryability)
                 .await
             {
                 Ok(output) => ExecutionDetails {
@@ -866,7 +861,7 @@ impl Client {
             (matches!(topology_type, TopologyType::Single) && server_type.is_available())
                 || server_type.is_data_bearing()
         }));
-        let _: SelectedServer = self
+        let _ = self
             .select_server(Some(&criteria), operation_name, None)
             .await?;
         Ok(())

src/operation.rs

@@ -52,7 +52,6 @@ use crate::{
         WriteFailure,
     },
     options::WriteConcern,
-    sdam::TopologyDescription,
     selection_criteria::SelectionCriteria,
     BoxFuture,
     ClientSession,
@@ -77,6 +76,7 @@ pub(crate) use update::{Update, UpdateOrReplace};
 
 const SERVER_4_2_0_WIRE_VERSION: i32 = 8;
 const SERVER_4_4_0_WIRE_VERSION: i32 = 9;
+const _SERVER_5_0_0_WIRE_VERSION: i32 = 13;
 const SERVER_8_0_0_WIRE_VERSION: i32 = 25;
 // The maximum number of bytes that may be included in a write payload when auto-encryption is
 // enabled.
@@ -149,9 +149,6 @@ pub(crate) trait Operation {
     /// Updates this operation as needed for a retry.
     fn update_for_retry(&mut self);
 
-    /// Updates this operation based on server topology.
-    fn update_for_topology(&mut self, _topology: &TopologyDescription);
-
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle>;
 
     fn name(&self) -> &str;
@@ -239,9 +236,6 @@ pub(crate) trait OperationWithDefaults: Send + Sync {
     /// Updates this operation as needed for a retry.
     fn update_for_retry(&mut self) {}
 
-    /// Updates this operation based on server topology.
-    fn update_for_topology(&mut self, _topology: &TopologyDescription) {}
-
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
         None
     }
@@ -294,9 +288,6 @@ where
     fn update_for_retry(&mut self) {
         self.update_for_retry()
     }
-    fn update_for_topology(&mut self, topology: &TopologyDescription) {
-        self.update_for_topology(topology)
-    }
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
         self.pinned_connection()
     }

src/operation/aggregate.rs

@@ -7,7 +7,7 @@ use crate::{
     cursor::CursorSpecification,
     error::Result,
     operation::{append_options, remove_empty_write_concern, Retryability},
-    options::{AggregateOptions, ReadPreference, SelectionCriteria, WriteConcern},
+    options::{AggregateOptions, SelectionCriteria, WriteConcern},
     Namespace,
 };
 
@@ -109,13 +109,32 @@ impl OperationWithDefaults for Aggregate {
         ))
     }
 
-    fn update_for_topology(&mut self, _topology: &crate::sdam::TopologyDescription) {
-        match self.selection_criteria() {
-            None => return,
-            Some(SelectionCriteria::ReadPreference(ReadPreference::Primary)) => return,
-            _ => (),
+    /*
+    fn update_for_topology(&mut self, topology: &crate::sdam::TopologyDescription) {
+        eprintln!("aggregate: update_for_topology");
+        if !self.is_out_or_merge()
+            || matches!(
+                self.selection_criteria(),
+                None | Some(SelectionCriteria::ReadPreference(ReadPreference::Primary))
+            )
+            || topology.topology_type() == TopologyType::LoadBalanced
+        {
+            eprintln!("aggregate: skipping topology update");
+            return;
+        }
+        for server in topology.servers.values() {
+            let _ = dbg!(server.hello_response());
+            if let Ok(Some(v)) = server.max_wire_version() {
+                if v < SERVER_5_0_0_WIRE_VERSION {
+                    eprintln!("aggregate: updating topology");
+                    self.options.get_or_insert_default().selection_criteria =
+                        Some(SelectionCriteria::ReadPreference(ReadPreference::Primary));
+                    break;
+                }
+            }
         }
     }
+    */
 
     fn selection_criteria(&self) -> Option<&SelectionCriteria> {
         self.options

src/operation/raw_output.rs

@@ -68,10 +68,6 @@ impl<Op: Operation> Operation for RawOutput<Op> {
         self.0.update_for_retry()
     }
 
-    fn update_for_topology(&mut self, topology: &crate::sdam::TopologyDescription) {
-        self.0.update_for_topology(topology)
-    }
-
     fn pinned_connection(&self) -> Option<&crate::cmap::conn::PinnedConnectionHandle> {
         self.0.pinned_connection()
     }

src/operation/run_cursor_command.rs

@@ -79,10 +79,6 @@ impl Operation for RunCursorCommand<'_> {
         self.run_command.update_for_retry()
     }
 
-    fn update_for_topology(&mut self, topology: &crate::sdam::TopologyDescription) {
-        self.run_command.update_for_topology(topology)
-    }
-
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
         self.run_command.pinned_connection()
     }

src/sdam/description/server.rs

@@ -386,6 +386,13 @@ impl ServerDescription {
         Ok(me)
     }
 
+    pub(crate) fn hello_response(&self) -> Result<Option<&HelloCommandResponse>> {
+        self.reply
+            .as_ref()
+            .map_err(Clone::clone)
+            .map(|o| o.as_ref().map(|r| &r.command_response))
+    }
+
     pub(crate) fn last_write_date(&self) -> Result<Option<DateTime>> {
         match self.reply {
             Ok(None) => Ok(None),

src/sdam/topology.rs

@@ -229,6 +229,7 @@ impl Topology {
         self.watcher.peek_latest().servers()
     }
 
+    #[cfg(test)]
     pub(crate) fn description(&self) -> TopologyDescription {
         self.watcher.peek_latest().description.clone()
     }