RUST-1425 Clarify test event handling (#1188)

Author: abr-egn

src/client/session/test.rs

@@ -244,7 +244,7 @@ async fn cluster_time_in_commands() {
         F: Fn(Client) -> G,
         G: Future<Output = Result<R>>,
     {
-        let mut subscriber = event_buffer.subscribe();
+        let mut event_stream = event_buffer.stream();
 
         operation(client.clone())
             .await
@@ -254,8 +254,8 @@ async fn cluster_time_in_commands() {
             .await
             .expect("operation should succeed");
 
-        let (first_command_started, first_command_succeeded) = subscriber
-            .wait_for_successful_command_execution(Duration::from_secs(5), command_name)
+        let (first_command_started, first_command_succeeded) = event_stream
+            .next_successful_command_execution(Duration::from_secs(5), command_name)
             .await
             .unwrap_or_else(|| {
                 panic!(
@@ -270,8 +270,8 @@ async fn cluster_time_in_commands() {
             .get("$clusterTime")
             .expect("should get cluster time from command response");
 
-        let (second_command_started, _) = subscriber
-            .wait_for_successful_command_execution(Duration::from_secs(5), command_name)
+        let (second_command_started, _) = event_stream
+            .next_successful_command_execution(Duration::from_secs(5), command_name)
             .await
             .unwrap_or_else(|| {
                 panic!(
@@ -311,13 +311,13 @@ async fn cluster_time_in_commands() {
         }
     }
 
-    let mut subscriber = buffer.subscribe();
+    let mut event_stream = buffer.stream();
 
     let client = Client::with_options(options).unwrap();
 
     // Wait for initial monitor check to complete and discover the server.
-    subscriber
-        .wait_for_event(Duration::from_secs(5), |event| match event {
+    event_stream
+        .next_match(Duration::from_secs(5), |event| match event {
             Event::Sdam(SdamEvent::ServerDescriptionChanged(e)) => {
                 !e.previous_description.server_type().is_available()
                     && e.new_description.server_type().is_available()

src/cmap/test.rs

@@ -146,7 +146,7 @@ impl Executor {
     }
 
     async fn execute_test(self) {
-        let mut subscriber = self.state.events.subscribe();
+        let mut event_stream = self.state.events.stream();
 
         let (updater, mut receiver) = TopologyUpdater::channel();
 
@@ -202,8 +202,8 @@ impl Executor {
         let description = self.description;
         let filter = |e: &CmapEvent| !ignored_event_names.iter().any(|name| e.name() == name);
         for expected_event in self.events {
-            let actual_event = subscriber
-                .wait_for_event(EVENT_TIMEOUT, filter)
+            let actual_event = event_stream
+                .next_match(EVENT_TIMEOUT, filter)
                 .await
                 .unwrap_or_else(|| {
                     panic!(
@@ -214,7 +214,12 @@ impl Executor {
             assert_matches(&actual_event, &expected_event, Some(description.as_str()));
         }
 
-        assert_eq!(subscriber.all(filter), Vec::new(), "{}", description);
+        assert_eq!(
+            event_stream.collect_now(filter),
+            Vec::new(),
+            "{}",
+            description
+        );
     }
 }
 
@@ -262,16 +267,16 @@ impl Operation {
                 }
             }
             Operation::CheckIn { connection } => {
-                let mut subscriber = state.events.subscribe();
+                let mut event_stream = state.events.stream();
                 let conn = state.connections.write().await.remove(&connection).unwrap();
                 let id = conn.id;
                 // connections are checked in via tasks spawned in their drop implementation,
                 // they are not checked in explicitly.
                 drop(conn);
 
                 // wait for event to be emitted to ensure check in has completed.
-                subscriber
-                    .wait_for_event(EVENT_TIMEOUT, |e| {
+                event_stream
+                    .next_match(EVENT_TIMEOUT, |e| {
                         matches!(e, CmapEvent::ConnectionCheckedIn(event) if event.connection_id == id)
                     })
                     .await
@@ -300,14 +305,14 @@ impl Operation {
                 }
             }
             Operation::Close => {
-                let mut subscriber = state.events.subscribe();
+                let mut event_stream = state.events.stream();
 
                 // pools are closed via their drop implementation
                 state.pool.write().await.take();
 
                 // wait for event to be emitted to ensure drop has completed.
-                subscriber
-                    .wait_for_event(EVENT_TIMEOUT, |e| matches!(e, CmapEvent::PoolClosed(_)))
+                event_stream
+                    .next_match(EVENT_TIMEOUT, |e| matches!(e, CmapEvent::PoolClosed(_)))
                     .await
                     .expect("did not receive ConnectionPoolClosed event after closing pool");
             }

src/cmap/test/integration.rs

@@ -199,7 +199,7 @@ async fn connection_error_during_establishment() {
 
     let buffer = EventBuffer::<CmapEvent>::new();
 
-    let mut subscriber = buffer.subscribe();
+    let mut event_stream = buffer.stream();
 
     let mut options = ConnectionPoolOptions::from_client_options(&client_options);
     options.ready = Some(true);
@@ -215,8 +215,8 @@ async fn connection_error_during_establishment() {
 
     pool.check_out().await.expect_err("check out should fail");
 
-    subscriber
-        .wait_for_event(EVENT_TIMEOUT, |e| match e {
+    event_stream
+        .next_match(EVENT_TIMEOUT, |e| match e {
             CmapEvent::ConnectionClosed(event) => {
                 event.connection_id == 1 && event.reason == ConnectionClosedReason::Error
             }
@@ -249,16 +249,16 @@ async fn connection_error_during_operation() {
         FailPoint::fail_command(&["ping"], FailPointMode::Times(10)).close_connection(true);
     let _guard = client.enable_fail_point(fail_point).await.unwrap();
 
-    let mut subscriber = buffer.subscribe();
+    let mut event_stream = buffer.stream();
 
     client
         .database("test")
         .run_command(doc! { "ping": 1 })
         .await
         .expect_err("ping should fail due to fail point");
 
-    subscriber
-        .wait_for_event(EVENT_TIMEOUT, |e| match e {
+    event_stream
+        .next_match(EVENT_TIMEOUT, |e| match e {
             CmapEvent::ConnectionClosed(event) => {
                 event.connection_id == 1 && event.reason == ConnectionClosedReason::Error
             }

src/sdam/description/topology/server_selection/test/in_window.rs

@@ -221,7 +221,7 @@ async fn load_balancing_test() {
         .build()
         .await;
 
-    let mut subscriber = client.events.subscribe_all();
+    let mut subscriber = client.events.stream_all();
 
     // wait for both servers pools to be saturated.
     for address in hosts {
@@ -243,7 +243,7 @@ async fn load_balancing_test() {
     let mut conns = 0;
     while conns < max_pool_size * 2 {
         subscriber
-            .wait_for_event(Duration::from_secs(30), |event| {
+            .next_match(Duration::from_secs(30), |event| {
                 matches!(event, Event::Cmap(CmapEvent::ConnectionReady(_)))
             })
             .await

src/sdam/description/topology/test/sdam.rs

@@ -277,7 +277,7 @@ async fn run_test(test_file: TestFile) {
     options.sdam_event_handler = Some(buffer.handler());
     options.test_options_mut().disable_monitoring_threads = true;
 
-    let mut event_subscriber = buffer.subscribe();
+    let mut event_stream = buffer.stream();
     let mut topology = Topology::new(options.clone()).unwrap();
 
     for (i, phase) in test_file.phases.into_iter().enumerate() {
@@ -374,8 +374,8 @@ async fn run_test(test_file: TestFile) {
                 );
             }
             Outcome::Events(EventsOutcome { events: expected }) => {
-                let actual = event_subscriber
-                    .collect_events(Duration::from_millis(500), |e| matches!(e, Event::Sdam(_)))
+                let actual = event_stream
+                    .collect(Duration::from_millis(500), |e| matches!(e, Event::Sdam(_)))
                     .await
                     .into_iter()
                     .map(|e| e.unwrap_sdam_event());
@@ -598,7 +598,7 @@ async fn topology_closed_event_last() {
         .await;
     let events = client.events.clone();
 
-    let mut subscriber = events.subscribe_all();
+    let mut subscriber = events.stream_all();
 
     client
         .database(function_name!())
@@ -609,15 +609,15 @@ async fn topology_closed_event_last() {
     drop(client);
 
     subscriber
-        .wait_for_event(Duration::from_millis(1000), |event| {
+        .next_match(Duration::from_millis(1000), |event| {
             matches!(event, Event::Sdam(SdamEvent::TopologyClosed(_)))
         })
         .await
         .expect("should see topology closed event");
 
     // no further SDAM events should be emitted after the TopologyClosedEvent
     let event = subscriber
-        .wait_for_event(Duration::from_millis(1000), |event| {
+        .next_match(Duration::from_millis(1000), |event| {
             matches!(event, Event::Sdam(_))
         })
         .await;
@@ -643,22 +643,22 @@ async fn heartbeat_events() {
         .build()
         .await;
 
-    let mut subscriber = client.events.subscribe_all();
+    let mut subscriber = client.events.stream_all();
 
     if client.is_load_balanced() {
         log_uncaptured("skipping heartbeat_events tests due to load-balanced topology");
         return;
     }
 
     subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .next_match(Duration::from_millis(500), |event| {
             matches!(event, Event::Sdam(SdamEvent::ServerHeartbeatStarted(_)))
         })
         .await
         .expect("should see server heartbeat started event");
 
     subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .next_match(Duration::from_millis(500), |event| {
             matches!(event, Event::Sdam(SdamEvent::ServerHeartbeatSucceeded(_)))
         })
         .await
@@ -681,7 +681,7 @@ async fn heartbeat_events() {
     let _guard = fp_client.enable_fail_point(fail_point).await.unwrap();
 
     subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .next_match(Duration::from_millis(500), |event| {
             matches!(event, Event::Sdam(SdamEvent::ServerHeartbeatFailed(_)))
         })
         .await

src/sdam/test.rs

@@ -103,7 +103,7 @@ async fn sdam_pool_management() {
         .build()
         .await;
 
-    let mut subscriber = client.events.subscribe_all();
+    let mut subscriber = client.events.stream_all();
 
     if !VersionReq::parse(">= 4.2.9")
         .unwrap()
@@ -116,14 +116,14 @@ async fn sdam_pool_management() {
     }
 
     subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .next_match(Duration::from_millis(500), |event| {
             matches!(event, Event::Cmap(CmapEvent::PoolReady(_)))
         })
         .await
         .expect("should see pool ready event");
 
     subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .next_match(Duration::from_millis(500), |event| {
             matches!(event, Event::Sdam(SdamEvent::ServerHeartbeatSucceeded(_)))
         })
         .await
@@ -143,9 +143,7 @@ async fn sdam_pool_management() {
 
     // Since there is no deterministic ordering, simply collect all the events and check for their
     // presence.
-    let events = subscriber
-        .collect_events(Duration::from_secs(1), |_| true)
-        .await;
+    let events = subscriber.collect(Duration::from_secs(1), |_| true).await;
     assert!(events
         .iter()
         .any(|e| matches!(e, Event::Sdam(SdamEvent::ServerHeartbeatFailed(_)))));
@@ -186,7 +184,7 @@ async fn hello_ok_true() {
 
     let buffer = EventBuffer::new();
 
-    let mut subscriber = buffer.subscribe();
+    let mut event_stream = buffer.stream();
 
     let mut options = setup_client_options.clone();
     options.sdam_event_handler = Some(buffer.handler());
@@ -195,8 +193,8 @@ async fn hello_ok_true() {
     let _client = Client::with_options(options).expect("client creation should succeed");
 
     // first heartbeat should be legacy hello but contain helloOk
-    subscriber
-        .wait_for_event(Duration::from_millis(2000), |event| {
+    event_stream
+        .next_match(Duration::from_millis(2000), |event| {
             if let Event::Sdam(SdamEvent::ServerHeartbeatSucceeded(e)) = event {
                 assert_eq!(e.reply.get_bool("helloOk"), Ok(true));
                 assert!(e.reply.get(LEGACY_HELLO_COMMAND_NAME_LOWERCASE).is_some());
@@ -210,8 +208,8 @@ async fn hello_ok_true() {
 
     // subsequent heartbeats should just be hello
     for _ in 0..3 {
-        subscriber
-            .wait_for_event(Duration::from_millis(2000), |event| {
+        event_stream
+            .next_match(Duration::from_millis(2000), |event| {
                 if let Event::Sdam(SdamEvent::ServerHeartbeatSucceeded(e)) = event {
                     assert!(e.reply.get("isWritablePrimary").is_some());
                     assert!(e.reply.get(LEGACY_HELLO_COMMAND_NAME_LOWERCASE).is_none());
@@ -270,13 +268,13 @@ async fn removed_server_monitor_stops() -> crate::error::Result<()> {
     let hosts = options.hosts.clone();
     let set_name = options.repl_set_name.clone().unwrap();
 
-    let mut subscriber = buffer.subscribe();
+    let mut event_stream = buffer.stream();
     let topology = Topology::new(options)?;
 
     // Wait until all three monitors have started.
     let mut seen_monitors = HashSet::new();
-    subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+    event_stream
+        .next_match(Duration::from_millis(500), |event| {
             if let Event::Sdam(SdamEvent::ServerHeartbeatStarted(e)) = event {
                 seen_monitors.insert(e.server_address.clone());
             }
@@ -315,13 +313,13 @@ async fn removed_server_monitor_stops() -> crate::error::Result<()> {
         ))
         .await;
 
-    subscriber.wait_for_event(Duration::from_secs(1), |event| {
+    event_stream.next_match(Duration::from_secs(1), |event| {
         matches!(event, Event::Sdam(SdamEvent::ServerClosed(e)) if e.address == hosts[2])
     }).await.expect("should see server closed event");
 
     // Capture heartbeat events for 1 second. The monitor for the removed server should stop
     // publishing them.
-    let events = subscriber.collect_events(Duration::from_secs(1), |event| {
+    let events = event_stream.collect(Duration::from_secs(1), |event| {
         matches!(event, Event::Sdam(SdamEvent::ServerHeartbeatStarted(e)) if e.server_address == hosts[2])
     }).await;