allow tests to specify backgroundThreadIntervalMS

Author: abr-egn

src/cmap/options.rs

@@ -1,7 +1,11 @@
 use std::{sync::Arc, time::Duration};
+#[cfg(test)]
+use std::cmp::Ordering;
 
 use derivative::Derivative;
-use serde::Deserialize;
+use serde::{Deserialize};
+#[cfg(test)]
+use serde::de::{Deserializer, Error};
 use typed_builder::TypedBuilder;
 
 use crate::{
@@ -40,10 +44,10 @@ pub(crate) struct ConnectionPoolOptions {
     #[serde(skip)]
     pub(crate) event_handler: Option<Arc<dyn CmapEventHandler>>,
 
-    /// How often the background thread performs its maintenance (e.g. ensure minPoolSize).
+    /// Interval between background thread maintenance runs (e.g. ensure minPoolSize).
     #[cfg(test)]
-    #[serde(skip)]
-    pub(crate) maintenance_frequency: Option<Duration>,
+    #[serde(default, rename = "backgroundThreadIntervalMS", deserialize_with = "BackgroundThreadInterval::deserialize_from_i64_millis")]
+    pub(crate) background_thread_interval: Option<BackgroundThreadInterval>,
 
     /// Connections that have been ready for usage in the pool for longer than `max_idle_time` will
     /// not be used.
@@ -101,7 +105,7 @@ impl ConnectionPoolOptions {
             credential: options.credential.clone(),
             event_handler: options.cmap_event_handler.clone(),
             #[cfg(test)]
-            maintenance_frequency: None,
+            background_thread_interval: None,
             #[cfg(test)]
             ready: None,
         }
@@ -116,6 +120,31 @@ impl ConnectionPoolOptions {
     }
 }
 
+#[cfg(test)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub(crate) enum BackgroundThreadInterval {
+    Never,
+    Every(Duration),
+}
+
+#[cfg(test)]
+impl BackgroundThreadInterval {
+    pub(crate) fn deserialize_from_i64_millis<'de, D>(
+        deserializer: D,
+    ) -> std::result::Result<Option<Self>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let millis = Option::<i64>::deserialize(deserializer)?;
+        let millis = if let Some(m) = millis { m } else { return Ok(None) };
+        Ok(Some(match millis.cmp(&0) {
+            Ordering::Less => BackgroundThreadInterval::Never,
+            Ordering::Equal => return Err(D::Error::custom("zero is not allowed")),
+            Ordering::Greater => BackgroundThreadInterval::Every(Duration::from_millis(millis as u64)),
+        }))
+    }
+}
+
 /// Options used for constructing a `Connection`.
 #[derive(Derivative)]
 #[derivative(Debug)]

src/cmap/test/mod.rs

@@ -150,17 +150,13 @@ impl Executor {
     async fn execute_test(self) {
         let mut subscriber = self.state.handler.subscribe();
 
-        // CMAP spec requires setting this to 50ms.
-        let mut options = self.pool_options;
-        options.maintenance_frequency = Some(Duration::from_millis(50));
-
         let (update_sender, mut update_receiver) = ServerUpdateSender::channel();
 
         let pool = ConnectionPool::new(
             CLIENT_OPTIONS.hosts[0].clone(),
             Default::default(),
             update_sender,
-            Some(options),
+            Some(self.pool_options),
         );
 
         // Mock a monitoring task responding to errors reported by the pool.

src/cmap/worker.rs

@@ -18,6 +18,8 @@ use super::{
     Connection,
     DEFAULT_MAX_POOL_SIZE,
 };
+#[cfg(test)]
+use super::options::BackgroundThreadInterval;
 use crate::{
     error::{Error, ErrorKind, Result},
     event::cmap::{
@@ -177,7 +179,11 @@ impl ConnectionPoolWorker {
         #[cfg(test)]
         let maintenance_frequency = options
             .as_ref()
-            .and_then(|opts| opts.maintenance_frequency)
+            .and_then(|opts| opts.background_thread_interval)
+            .map(|i| match i {
+                BackgroundThreadInterval::Never => Duration::MAX,
+                BackgroundThreadInterval::Every(d) => d,
+            })
             .unwrap_or(MAINTENACE_FREQUENCY);
 
         #[cfg(not(test))]