minor: fix clippy (#502)

Author: isabelatkinson

src/cmap/conn/stream_description.rs

@@ -15,6 +15,7 @@ pub(crate) struct StreamDescription {
     pub(crate) max_wire_version: Option<i32>,
 
     /// The minimum wire version that the server understands.
+    #[allow(dead_code)]
     pub(crate) min_wire_version: Option<i32>,
 
     /// The supported authentication mechanisms that the server understands.

src/cmap/connection_requester.rs

@@ -3,21 +3,18 @@ use tokio::sync::{mpsc, oneshot};
 use super::{worker::PoolWorkerHandle, Connection};
 use crate::{
     error::{Error, Result},
-    options::ServerAddress,
     runtime::AsyncJoinHandle,
 };
 
 /// Returns a new requester/receiver pair.
 pub(super) fn channel(
-    address: ServerAddress,
     handle: PoolWorkerHandle,
 ) -> (ConnectionRequester, ConnectionRequestReceiver) {
     let (sender, receiver) = mpsc::unbounded_channel();
     (
         ConnectionRequester {
-            address,
             sender,
-            handle,
+            _handle: handle,
         },
         ConnectionRequestReceiver { receiver },
     )
@@ -28,9 +25,8 @@ pub(super) fn channel(
 /// the pool will stop servicing requests, drop its available connections, and close.
 #[derive(Clone, Debug)]
 pub(super) struct ConnectionRequester {
-    address: ServerAddress,
     sender: mpsc::UnboundedSender<oneshot::Sender<ConnectionRequestResult>>,
-    handle: PoolWorkerHandle,
+    _handle: PoolWorkerHandle,
 }
 
 impl ConnectionRequester {
@@ -83,7 +79,7 @@ impl ConnectionRequest {
 #[derive(Debug)]
 pub(super) enum ConnectionRequestResult {
     /// A connection that was already established and was simply checked out of the pool.
-    Pooled(Connection),
+    Pooled(Box<Connection>),
 
     /// A new connection in the process of being established.
     /// The handle can be awaited upon to receive the established connection.
@@ -97,7 +93,7 @@ pub(super) enum ConnectionRequestResult {
 impl ConnectionRequestResult {
     pub(super) fn unwrap_pooled_connection(self) -> Connection {
         match self {
-            ConnectionRequestResult::Pooled(c) => c,
+            ConnectionRequestResult::Pooled(c) => *c,
             _ => panic!("attempted to unwrap pooled connection when was establishing"),
         }
     }

src/cmap/establish/handshake/mod.rs

@@ -146,6 +146,8 @@ pub(crate) struct Handshaker {
     credential: Option<Credential>,
     #[cfg(test)]
     mock_service_id: bool,
+    // This field is not read without a compression feature flag turned on.
+    #[allow(dead_code)]
     compressors: Option<Vec<Compressor>>,
 }
 
@@ -286,8 +288,13 @@ impl Handshaker {
                 .iter()
                 .find(|c| server_compressors.iter().any(|x| c.name() == x))
             {
+                // Without a feature flag turned on, the Compressor enum is empty which causes an
+                // unreachable code warning.
+                #[allow(unreachable_code)]
                 // zlib compression level is already set
-                conn.compressor = Some(compressor.clone());
+                {
+                    conn.compressor = Some(compressor.clone());
+                }
             }
         }
 

src/cmap/manager.rs

@@ -27,7 +27,7 @@ impl PoolManager {
         if self
             .sender
             .send(PoolManagementRequest::Clear {
-                completion_handler: message,
+                _completion_handler: message,
                 cause,
                 service_id,
             })
@@ -43,7 +43,7 @@ impl PoolManager {
         if self
             .sender
             .send(PoolManagementRequest::MarkAsReady {
-                completion_handler: message,
+                _completion_handler: message,
             })
             .is_ok()
         {
@@ -54,7 +54,10 @@ impl PoolManager {
     /// Check in the given connection to the pool.
     /// This returns an error containing the connection if the pool has been dropped already.
     pub(crate) fn check_in(&self, connection: Connection) -> std::result::Result<(), Connection> {
-        if let Err(request) = self.sender.send(PoolManagementRequest::CheckIn(connection)) {
+        if let Err(request) = self
+            .sender
+            .send(PoolManagementRequest::CheckIn(Box::new(connection)))
+        {
             let conn = request.0.unwrap_check_in();
             return Err(conn);
         }
@@ -99,18 +102,18 @@ impl ManagementRequestReceiver {
 pub(super) enum PoolManagementRequest {
     /// Clear the pool, transitioning it to Paused.
     Clear {
-        completion_handler: AcknowledgedMessage<()>,
+        _completion_handler: AcknowledgedMessage<()>,
         cause: Error,
         service_id: Option<ObjectId>,
     },
 
     /// Mark the pool as Ready, allowing connections to be created and checked out.
     MarkAsReady {
-        completion_handler: AcknowledgedMessage<()>,
+        _completion_handler: AcknowledgedMessage<()>,
     },
 
     /// Check in the given connection.
-    CheckIn(Connection),
+    CheckIn(Box<Connection>),
 
     /// Update the pool based on the given establishment error.
     HandleConnectionFailed,
@@ -127,15 +130,15 @@ pub(super) enum PoolManagementRequest {
 impl PoolManagementRequest {
     fn unwrap_check_in(self) -> Connection {
         match self {
-            PoolManagementRequest::CheckIn(conn) => conn,
+            PoolManagementRequest::CheckIn(conn) => *conn,
             _ => panic!("tried to unwrap checkin but got {:?}", self),
         }
     }
 }
 
 #[derive(Debug)]
 pub(super) enum ConnectionSucceeded {
-    ForPool(Connection),
+    ForPool(Box<Connection>),
     Used { service_id: Option<ObjectId> },
 }
 

src/cmap/mod.rs

@@ -98,7 +98,7 @@ impl ConnectionPool {
     pub(crate) fn new_mocked(address: ServerAddress) -> Self {
         let (manager, _) = manager::channel();
         let handle = PoolWorkerHandle::new_mocked();
-        let (connection_requester, _) = connection_requester::channel(Default::default(), handle);
+        let (connection_requester, _) = connection_requester::channel(handle);
         let (_, generation_subscriber) = status::channel(PoolGeneration::normal());
 
         Self {
@@ -134,7 +134,7 @@ impl ConnectionPool {
         let response = self.connection_requester.request().await;
 
         let conn = match response {
-            ConnectionRequestResult::Pooled(c) => Ok(c),
+            ConnectionRequestResult::Pooled(c) => Ok(*c),
             ConnectionRequestResult::Establishing(task) => task.await,
             ConnectionRequestResult::PoolCleared(e) => {
                 Err(Error::pool_cleared_error(&self.address, &e))

src/cmap/test/file.rs

@@ -7,6 +7,9 @@ use crate::{bson_util, cmap::options::ConnectionPoolOptions, error::Result, test
 use bson::Document;
 
 #[derive(Debug, Deserialize)]
+// TODO RUST-1079: remove the #[allow(dead_code)] tag and add #[serde(deny_unknown_fields)] to
+// ensure these tests are being fully run
+#[allow(dead_code)]
 #[serde(rename_all = "camelCase")]
 pub struct TestFile {
     version: u8,
@@ -83,6 +86,9 @@ pub enum Operation {
 }
 
 #[derive(Debug, Deserialize)]
+// TODO RUST-1077: remove the #[allow(dead_code)] tag and add #[serde(deny_unknown_fields)] to
+// ensure these tests are being fully run
+#[allow(dead_code)]
 pub struct Error {
     #[serde(rename = "type")]
     pub type_: String,

src/cmap/test/integration.rs

@@ -140,9 +140,10 @@ async fn concurrent_connections() {
                     consecutive_creations += 1;
                 }
                 Event::ConnectionReady(_) => {
-                    if consecutive_creations < 2 {
-                        panic!("connections not created concurrently");
-                    }
+                    assert!(
+                        consecutive_creations >= 2,
+                        "connections not created concurrently"
+                    );
                 }
                 _ => (),
             }

src/cmap/test/mod.rs

@@ -127,7 +127,7 @@ impl Executor {
         let handler = Arc::new(EventHandler::new());
         let error = test_file.error;
 
-        let mut pool_options = test_file.pool_options.unwrap_or_else(Default::default);
+        let mut pool_options = test_file.pool_options.unwrap_or_default();
         pool_options.tls_options = CLIENT_OPTIONS.tls_options();
         pool_options.cmap_event_handler = Some(handler.clone());
         pool_options.server_api = SERVER_API.clone();

src/cmap/worker.rs

@@ -171,8 +171,7 @@ impl ConnectionPoolWorker {
             .map(|pool_options| ConnectionOptions::from(pool_options.clone()));
 
         let (handle, handle_listener) = handle_channel();
-        let (connection_requester, request_receiver) =
-            connection_requester::channel(address.clone(), handle);
+        let (connection_requester, request_receiver) = connection_requester::channel(handle);
         let (manager, management_receiver) = manager::channel();
 
         let is_load_balanced = options
@@ -292,31 +291,29 @@ impl ConnectionPoolWorker {
                         ));
                     }
                 },
-                PoolTask::HandleManagementRequest(PoolManagementRequest::CheckIn(connection)) => {
-                    self.check_in(connection);
-                }
-                PoolTask::HandleManagementRequest(PoolManagementRequest::Clear {
-                    completion_handler: _,
-                    cause,
-                    service_id,
-                }) => {
-                    self.clear(cause, service_id);
-                }
-                PoolTask::HandleManagementRequest(PoolManagementRequest::MarkAsReady {
-                    completion_handler: _handler,
-                }) => {
-                    self.mark_as_ready();
-                }
-                PoolTask::HandleManagementRequest(
-                    PoolManagementRequest::HandleConnectionSucceeded(conn),
-                ) => self.handle_connection_succeeded(conn),
-                PoolTask::HandleManagementRequest(
-                    PoolManagementRequest::HandleConnectionFailed,
-                ) => self.handle_connection_failed(),
-                #[cfg(test)]
-                PoolTask::HandleManagementRequest(PoolManagementRequest::Sync(tx)) => {
-                    let _ = tx.send(());
-                }
+                PoolTask::HandleManagementRequest(request) => match *request {
+                    PoolManagementRequest::CheckIn(connection) => {
+                        self.check_in(*connection);
+                    }
+                    PoolManagementRequest::Clear {
+                        cause, service_id, ..
+                    } => {
+                        self.clear(cause, service_id);
+                    }
+                    PoolManagementRequest::MarkAsReady { .. } => {
+                        self.mark_as_ready();
+                    }
+                    PoolManagementRequest::HandleConnectionSucceeded(conn) => {
+                        self.handle_connection_succeeded(conn);
+                    }
+                    PoolManagementRequest::HandleConnectionFailed => {
+                        self.handle_connection_failed();
+                    }
+                    #[cfg(test)]
+                    PoolManagementRequest::Sync(tx) => {
+                        let _ = tx.send(());
+                    }
+                },
                 PoolTask::Maintenance => {
                     self.perform_maintenance();
                 }
@@ -372,7 +369,7 @@ impl ConnectionPoolWorker {
             }
 
             conn.mark_as_in_use(self.manager.clone());
-            if let Err(request) = request.fulfill(ConnectionRequestResult::Pooled(conn)) {
+            if let Err(request) = request.fulfill(ConnectionRequestResult::Pooled(Box::new(conn))) {
                 // checking out thread stopped listening, indicating it hit the WaitQueue
                 // timeout, so we put connection back into pool.
                 let mut connection = request.unwrap_pooled_connection();
@@ -466,7 +463,8 @@ impl ConnectionPoolWorker {
             let count = self.service_connection_count.entry(sid).or_insert(0);
             *count += 1;
         }
-        if let ConnectionSucceeded::ForPool(mut connection) = connection {
+        if let ConnectionSucceeded::ForPool(connection) = connection {
+            let mut connection = *connection;
             connection.mark_as_available();
             self.available_connections.push_back(connection);
         }
@@ -623,8 +621,9 @@ impl ConnectionPoolWorker {
                     .await;
 
                     if let Ok(connection) = connection {
-                        manager
-                            .handle_connection_succeeded(ConnectionSucceeded::ForPool(connection))
+                        manager.handle_connection_succeeded(ConnectionSucceeded::ForPool(Box::new(
+                            connection,
+                        )))
                     }
                 });
             }
@@ -690,7 +689,7 @@ enum PoolState {
 #[derive(Debug)]
 enum PoolTask {
     /// Handle a management request from a `PoolManager`.
-    HandleManagementRequest(PoolManagementRequest),
+    HandleManagementRequest(Box<PoolManagementRequest>),
 
     /// Fulfill the given connection request.
     CheckOut(ConnectionRequest),
@@ -701,22 +700,25 @@ enum PoolTask {
 
 impl From<PoolManagementRequest> for PoolTask {
     fn from(request: PoolManagementRequest) -> Self {
-        PoolTask::HandleManagementRequest(request)
+        PoolTask::HandleManagementRequest(Box::new(request))
     }
 }
 
 /// Constructs a new channel for for monitoring whether this pool still has references
 /// to it.
 fn handle_channel() -> (PoolWorkerHandle, HandleListener) {
     let (sender, receiver) = mpsc::channel(1);
-    (PoolWorkerHandle { sender }, HandleListener { receiver })
+    (
+        PoolWorkerHandle { _sender: sender },
+        HandleListener { receiver },
+    )
 }
 
 /// Handle to the worker. Once all handles have been dropped, the worker
 /// will stop waiting for new requests and drop the pool itself.
 #[derive(Debug, Clone)]
 pub(super) struct PoolWorkerHandle {
-    sender: mpsc::Sender<()>,
+    _sender: mpsc::Sender<()>,
 }
 
 impl PoolWorkerHandle {

src/cursor/mod.rs

@@ -124,9 +124,10 @@ where
     /// on.
     #[cfg(test)]
     pub(crate) fn set_kill_watcher(&mut self, tx: oneshot::Sender<()>) {
-        if self.kill_watcher.is_some() {
-            panic!("cursor already has a kill_watcher");
-        }
+        assert!(
+            self.kill_watcher.is_none(),
+            "cursor already has a kill_watcher"
+        );
         self.kill_watcher = Some(tx);
     }
 }