RUST-1888 Remove incidental mutability from EventBuffer (#1111)

Author: abr-egn

src/client/session/test.rs

@@ -383,17 +383,14 @@ async fn session_usage() {
         return;
     }
 
-    #[allow(deprecated)]
     async fn session_usage_test<F, G>(command_name: &str, operation: F)
     where
         F: Fn(EventClient) -> G,
         G: Future<Output = ()>,
     {
         let client = Client::test_builder().monitor_events().build().await;
         operation(client.clone()).await;
-        let mut events = client.events.clone();
-        #[allow(deprecated)]
-        let (command_started, _) = events.get_successful_command_execution(command_name);
+        let (command_started, _) = client.events.get_successful_command_execution(command_name);
         assert!(
             command_started.command.get("lsid").is_some(),
             "implicit session not passed to {}",
@@ -427,11 +424,7 @@ async fn implicit_session_returned_after_immediate_exhaust() {
     let mut cursor = coll.find(doc! {}).await.expect("find should succeed");
     assert!(matches!(cursor.next().await, Some(Ok(_))));
 
-    #[allow(deprecated)]
-    let (find_started, _) = {
-        let mut events = client.events.clone();
-        events.get_successful_command_execution("find")
-    };
+    let (find_started, _) = client.events.get_successful_command_execution("find");
     let session_id = find_started
         .command
         .get("lsid")
@@ -480,11 +473,7 @@ async fn implicit_session_returned_after_exhaust_by_get_more() {
         assert!(matches!(cursor.next().await, Some(Ok(_))));
     }
 
-    #[allow(deprecated)]
-    let (find_started, _) = {
-        let mut events = client.events.clone();
-        events.get_successful_command_execution("find")
-    };
+    let (find_started, _) = client.events.get_successful_command_execution("find");
 
     let session_id = find_started
         .command
@@ -545,6 +534,8 @@ async fn find_and_getmore_share_session() {
         coll: &Collection<Document>,
         read_preference: ReadPreference,
     ) {
+        client.events.clone().clear_cached_events();
+
         let options = FindOptions::builder()
             .batch_size(2)
             .selection_criteria(SelectionCriteria::ReadPreference(read_preference.clone()))
@@ -582,17 +573,14 @@ async fn find_and_getmore_share_session() {
                 });
         }
 
-        let mut events = client.events.clone();
-        #[allow(deprecated)]
-        let (find_started, _) = events.get_successful_command_execution("find");
+        let (find_started, _) = client.events.get_successful_command_execution("find");
         let session_id = find_started
             .command
             .get("lsid")
             .expect("find should use implicit session");
         assert!(session_id != &Bson::Null);
 
-        #[allow(deprecated)]
-        let (command_started, _) = events.get_successful_command_execution("getMore");
+        let (command_started, _) = client.events.get_successful_command_execution("getMore");
         let getmore_session_id = command_started
             .command
             .get("lsid")

src/client/session/test/causal_consistency.rs

@@ -144,6 +144,8 @@ async fn first_read_no_after_cluser_time() {
     }
 
     for op in all_session_ops().filter(|o| o.is_read) {
+        client.events.clone().clear_cached_events();
+
         let mut session = client
             .start_session()
             .causal_consistency(true)
@@ -160,10 +162,7 @@ async fn first_read_no_after_cluser_time() {
         .await
         .unwrap_or_else(|e| panic!("{} failed: {}", name, e));
         #[allow(deprecated)]
-        let (started, _) = {
-            let mut events = client.events.clone();
-            events.get_successful_command_execution(name)
-        };
+        let (started, _) = client.events.get_successful_command_execution(name);
 
         // assert that no read concern was set.
         started.command.get_document("readConcern").unwrap_err();
@@ -181,6 +180,8 @@ async fn first_op_update_op_time() {
     }
 
     for op in all_session_ops() {
+        client.events.clone().clear_cached_events();
+
         let mut session = client
             .start_session()
             .causal_consistency(true)
@@ -197,15 +198,12 @@ async fn first_op_update_op_time() {
         .await
         .unwrap();
 
-        #[allow(deprecated)]
-        let event = {
-            let mut events = client.events.clone();
-            events
-                .get_command_events(&[name])
-                .into_iter()
-                .find(|e| matches!(e, CommandEvent::Succeeded(_) | CommandEvent::Failed(_)))
-                .unwrap_or_else(|| panic!("no event found for {}", name))
-        };
+        let event = client
+            .events
+            .get_command_events(&[name])
+            .into_iter()
+            .find(|e| matches!(e, CommandEvent::Succeeded(_) | CommandEvent::Failed(_)))
+            .unwrap_or_else(|| panic!("no event found for {}", name));
 
         match event {
             CommandEvent::Succeeded(s) => {
@@ -490,11 +488,7 @@ async fn omit_cluster_time_standalone() {
 
     coll.find_one(doc! {}).await.unwrap();
 
-    #[allow(deprecated)]
-    let (started, _) = {
-        let mut events = client.events.clone();
-        events.get_successful_command_execution("find")
-    };
+    let (started, _) = client.events.get_successful_command_execution("find");
     started.command.get_document("$clusterTime").unwrap_err();
 }
 
@@ -513,9 +507,6 @@ async fn cluster_time_sent_in_commands() {
 
     coll.find_one(doc! {}).await.unwrap();
 
-    #[allow(deprecated)]
-    let mut events = client.events.clone();
-    #[allow(deprecated)]
-    let (started, _) = events.get_successful_command_execution("find");
+    let (started, _) = client.events.get_successful_command_execution("find");
     started.command.get_document("$clusterTime").unwrap();
 }

src/test/change_stream.rs

@@ -82,18 +82,15 @@ async fn tracks_resume_token() -> Result<()> {
         tokens.push(stream.resume_token().unwrap().parsed()?);
     }
 
-    #[allow(deprecated)]
-    let events: Vec<_> = {
-        let mut events = client.events.clone();
-        events
-            .get_command_events(&["aggregate", "getMore"])
-            .into_iter()
-            .filter_map(|ev| match ev {
-                CommandEvent::Succeeded(s) => Some(s),
-                _ => None,
-            })
-            .collect()
-    };
+    let events: Vec<_> = client
+        .events
+        .get_command_events(&["aggregate", "getMore"])
+        .into_iter()
+        .filter_map(|ev| match ev {
+            CommandEvent::Succeeded(s) => Some(s),
+            _ => None,
+        })
+        .collect();
     let mut expected = vec![];
     // Token from `aggregate`
     if let Some(initial) = events[0]
@@ -229,11 +226,7 @@ async fn empty_batch_not_closed() -> Result<()> {
     coll.insert_one(doc! {}).await?;
     stream.next().await.transpose()?;
 
-    #[allow(deprecated)]
-    let events = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["aggregate", "getMore"])
-    };
+    let events = client.events.get_command_events(&["aggregate", "getMore"]);
     let cursor_id = match &events[1] {
         CommandEvent::Succeeded(CommandSucceededEvent { reply, .. }) => {
             reply.get_document("cursor")?.get_i64("id")?
@@ -315,11 +308,7 @@ async fn resume_start_at_operation_time() -> Result<()> {
     coll.insert_one(doc! { "_id": 2 }).await?;
     stream.next().await.transpose()?;
 
-    #[allow(deprecated)]
-    let events = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["aggregate"])
-    };
+    let events = client.events.get_command_events(&["aggregate"]);
     assert_eq!(events.len(), 4);
 
     fn has_saot(command: &Document) -> Result<bool> {
@@ -362,11 +351,7 @@ async fn batch_end_resume_token() -> Result<()> {
 
     assert_eq!(stream.next_if_any().await?, None);
     let token = stream.resume_token().unwrap().parsed()?;
-    #[allow(deprecated)]
-    let commands = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["aggregate", "getMore"])
-    };
+    let commands = client.events.get_command_events(&["aggregate", "getMore"]);
     assert!(matches!(commands.last(), Some(
         CommandEvent::Succeeded(CommandSucceededEvent {
             reply,

src/test/coll.rs

@@ -696,10 +696,7 @@ async fn no_read_preference_to_standalone() {
         .await
         .unwrap();
 
-    #[allow(deprecated)]
-    let mut events = client.events.clone();
-    #[allow(deprecated)]
-    let command_started = events.get_successful_command_execution("find").0;
+    let command_started = client.events.get_successful_command_execution("find").0;
 
     assert!(!command_started.command.contains_key("$readPreference"));
 }

src/test/cursor.rs

@@ -138,15 +138,12 @@ async fn batch_exhaustion() {
     assert_eq!(4, v.len());
 
     // Assert that the last `getMore` response always has id 0, i.e. is exhausted.
-    #[allow(deprecated)]
-    let replies: Vec<_> = {
-        let mut events = client.events.clone();
-        events
-            .get_command_events(&["getMore"])
-            .into_iter()
-            .filter_map(|e| e.as_command_succeeded().map(|e| e.reply.clone()))
-            .collect()
-    };
+    let replies: Vec<_> = client
+        .events
+        .get_command_events(&["getMore"])
+        .into_iter()
+        .filter_map(|e| e.as_command_succeeded().map(|e| e.reply.clone()))
+        .collect();
     let last = replies.last().unwrap();
     let cursor = last.get_document("cursor").unwrap();
     let id = cursor.get_i64("id").unwrap();

src/test/spec/retryable_reads.rs

@@ -200,11 +200,7 @@ async fn retry_read_different_mongos() {
         .find(doc! {})
         .await;
     assert!(result.is_err());
-    #[allow(deprecated)]
-    let events = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["find"])
-    };
+    let events = client.events.get_command_events(&["find"]);
     assert!(
         matches!(
             &events[..],
@@ -261,11 +257,7 @@ async fn retry_read_same_mongos() {
         .find(doc! {})
         .await;
     assert!(result.is_ok(), "{:?}", result);
-    #[allow(deprecated)]
-    let events = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["find"])
-    };
+    let events = client.events.get_command_events(&["find"]);
     assert!(
         matches!(
             &events[..],

src/test/spec/retryable_writes.rs

@@ -45,12 +45,13 @@ async fn transaction_ids_excluded() {
 
     let coll = client.init_db_and_coll(function_name!(), "coll").await;
 
-    #[allow(deprecated)]
-    let mut events = client.events.clone();
-    let mut excludes_txn_number = move |command_name: &str| -> bool {
-        #[allow(deprecated)]
-        let (started, _) = events.get_successful_command_execution(command_name);
-        !started.command.contains_key("txnNumber")
+    let excludes_txn_number = {
+        let events = client.events.clone();
+        move |command_name: &str| -> bool {
+            let (started, _) = events.get_successful_command_execution(command_name);
+            events.clone().clear_cached_events();
+            !started.command.contains_key("txnNumber")
+        }
     };
 
     coll.update_many(doc! {}, doc! { "$set": doc! { "x": 1 } })
@@ -93,12 +94,13 @@ async fn transaction_ids_included() {
 
     let coll = client.init_db_and_coll(function_name!(), "coll").await;
 
-    #[allow(deprecated)]
-    let mut events = client.events.clone();
-    let mut includes_txn_number = move |command_name: &str| -> bool {
-        #[allow(deprecated)]
-        let (started, _) = events.get_successful_command_execution(command_name);
-        started.command.contains_key("txnNumber")
+    let includes_txn_number = {
+        let events = client.events.clone();
+        move |command_name: &str| -> bool {
+            let (started, _) = events.get_successful_command_execution(command_name);
+            events.clone().clear_cached_events();
+            started.command.contains_key("txnNumber")
+        }
     };
 
     coll.insert_one(doc! { "x": 1 }).await.unwrap();
@@ -454,11 +456,7 @@ async fn retry_write_different_mongos() {
         .insert_one(doc! {})
         .await;
     assert!(result.is_err());
-    #[allow(deprecated)]
-    let events = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["insert"])
-    };
+    let events = client.events.get_command_events(&["insert"]);
     assert!(
         matches!(
             &events[..],
@@ -516,11 +514,7 @@ async fn retry_write_same_mongos() {
         .insert_one(doc! {})
         .await;
     assert!(result.is_ok(), "{:?}", result);
-    #[allow(deprecated)]
-    let events = {
-        let mut events = client.events.clone();
-        events.get_command_events(&["insert"])
-    };
+    let events = client.events.get_command_events(&["insert"]);
     assert!(
         matches!(
             &events[..],

src/test/spec/write_error.rs

@@ -30,11 +30,7 @@ async fn details() {
         ErrorKind::Write(WriteFailure::WriteError(e)) => e,
         _ => panic!("expected WriteError, got {:?}", err.kind),
     };
-    #[allow(deprecated)]
-    let (_, event) = {
-        let mut events = client.events.clone();
-        events.get_successful_command_execution("insert")
-    };
+    let (_, event) = client.events.get_successful_command_execution("insert");
     assert_eq!(write_err.code, 121 /* DocumentValidationFailure */);
     assert_eq!(
         &write_err.details.unwrap(),

src/test/util/event.rs

@@ -6,13 +6,16 @@ use crate::{
     bson::doc,
     event::{
         cmap::CmapEvent,
-        command::{CommandEvent, CommandStartedEvent, CommandSucceededEvent},
+        command::{CommandEvent, CommandSucceededEvent},
         sdam::SdamEvent,
     },
     test::get_client_options,
     Client,
 };
 
+#[cfg(feature = "in-use-encryption-unstable")]
+use crate::event::command::CommandStartedEvent;
+
 #[derive(Clone, Debug, From, Serialize)]
 #[serde(untagged)]
 #[allow(clippy::large_enum_variant)]
@@ -89,21 +92,6 @@ impl CommandEvent {
         }
     }
 
-    pub(crate) fn request_id(&self) -> i32 {
-        match self {
-            CommandEvent::Started(event) => event.request_id,
-            CommandEvent::Failed(event) => event.request_id,
-            CommandEvent::Succeeded(event) => event.request_id,
-        }
-    }
-
-    pub(crate) fn as_command_started(&self) -> Option<&CommandStartedEvent> {
-        match self {
-            CommandEvent::Started(e) => Some(e),
-            _ => None,
-        }
-    }
-
     pub(crate) fn as_command_succeeded(&self) -> Option<&CommandSucceededEvent> {
         match self {
             CommandEvent::Succeeded(e) => Some(e),