RUST-633 update driver to use tokio 1.0 (#283)

Author: saghm

.evergreen/aws-ecs-test/Cargo.toml

@@ -5,7 +5,7 @@ authors = ["Saghm Rossi <[email protected]>"]
 edition = "2018"
 
 [dependencies]
-tokio = "0.2.21"
+tokio = "1.0.2"
 
 [dependencies.mongodb]
 path = "../.."

Cargo.toml

@@ -21,12 +21,12 @@ exclude = [
 
 [features]
 default = ["tokio-runtime"]
-tokio-runtime = ["tokio/dns", "tokio/macros", "tokio/rt-core", "tokio/tcp", "tokio/rt-threaded", "tokio/time", "reqwest", "serde_bytes"]
-async-std-runtime = ["async-std", "async-std/attributes"]
+tokio-runtime = ["tokio/macros", "tokio/net", "tokio/rt", "tokio/time", "reqwest", "serde_bytes"]
+async-std-runtime = ["async-std", "async-std/attributes", "async-std-resolver", "tokio-util/compat"]
 sync = ["async-std-runtime"]
 
 [dependencies]
-async-trait = "0.1.24"
+async-trait = "0.1.42"
 base64 = "0.11.0"
 bitflags = "1.1.0"
 bson = "1.1.0"
@@ -49,8 +49,8 @@ stringprep = "0.1.2"
 strsim = "0.10.0"
 take_mut = "0.2.2"
 time = "0.1.42"
-trust-dns-proto = "0.19.4"
-trust-dns-resolver = "0.19.5"
+trust-dns-proto = "0.20.0"
+trust-dns-resolver = "0.20.0"
 typed-builder = "0.4.0"
 version_check = "0.9.1"
 webpki = "0.21.0"
@@ -60,18 +60,22 @@ webpki-roots = "0.18.0"
 version = "1.6.2"
 optional = true
 
+[dependencies.async-std-resolver]
+version = "0.20.0"
+optional = true
+
 [dependencies.pbkdf2]
 version = "0.3.0"
 default-features = false
 
 [dependencies.reqwest]
-version = "0.10.6"
+version = "0.11.0"
 optional = true
 default-features = false
 features = ["json", "rustls-tls"]
 
 [dependencies.rustls]
-version = "0.17.0"
+version = "0.19.0"
 features = ["dangerous_configuration"]
 
 [dependencies.serde]
@@ -83,13 +87,17 @@ version = "0.11.5"
 optional = true
 
 [dependencies.tokio]
-version = "~0.2.18"
+version = "1.0.1"
 features = ["io-util", "sync", "macros"]
 
 [dependencies.tokio-rustls]
-version = "0.13.0"
+version = "0.22.0"
 features = ["dangerous_configuration"]
 
+[dependencies.tokio-util]
+version = "0.6.1"
+features = ["io"]
+
 [dependencies.uuid]
 version = "0.8.1"
 features = ["v4"]

src/client/mod.rs

@@ -270,7 +270,7 @@ impl Client {
             }
 
             let mut topology_change_subscriber =
-                self.inner.topology.subscribe_to_topology_changes().await;
+                self.inner.topology.subscribe_to_topology_changes();
             self.inner.topology.request_topology_check();
 
             let time_passed = start_time.to(PreciseTime::now());

src/cmap/status.rs

@@ -1,5 +1,3 @@
-use crate::RUNTIME;
-
 /// Struct used to track the latest status of the pool.
 #[derive(Clone, Debug)]
 struct PoolStatus {
@@ -15,11 +13,7 @@ impl Default for PoolStatus {
 
 /// Create a channel for publishing and receiving updates to the pool's generation.
 pub(super) fn channel() -> (PoolGenerationPublisher, PoolGenerationSubscriber) {
-    let (sender, mut receiver) = tokio::sync::watch::channel(Default::default());
-    // The first call to recv on a watch channel returns immediately with the initial value.
-    // We use RUNTIME.block_in_place because this is not a truly blocking task, so
-    // the runtimes don't need to shift things around to ensure scheduling continues normally.
-    RUNTIME.block_in_place(receiver.recv());
+    let (sender, receiver) = tokio::sync::watch::channel(Default::default());
     (
         PoolGenerationPublisher { sender },
         PoolGenerationSubscriber { receiver },
@@ -40,7 +34,7 @@ impl PoolGenerationPublisher {
         };
 
         // if nobody is listening, this will return an error, which we don't mind.
-        let _: std::result::Result<_, _> = self.sender.broadcast(new_status);
+        let _: std::result::Result<_, _> = self.sender.send(new_status);
     }
 }
 
@@ -62,10 +56,11 @@ impl PoolGenerationSubscriber {
         timeout: std::time::Duration,
     ) -> Option<u32> {
         crate::RUNTIME
-            .timeout(timeout, self.receiver.recv())
+            .timeout(timeout, self.receiver.changed())
             .await
-            .ok()
-            .flatten()
-            .map(|status| status.generation)
+            .ok()?
+            .ok()?;
+
+        Some(self.receiver.borrow().generation)
     }
 }

src/cmap/test/event.rs

@@ -6,7 +6,7 @@ use std::{
 use serde::{de::Unexpected, Deserialize, Deserializer};
 
 use crate::{event::cmap::*, options::StreamAddress, RUNTIME};
-use tokio::sync::broadcast::{RecvError, SendError};
+use tokio::sync::broadcast::error::{RecvError, SendError};
 
 #[derive(Clone, Debug)]
 pub struct EventHandler {

src/cmap/test/integration.rs

@@ -158,7 +158,7 @@ async fn concurrent_connections() {
         .expect("disabling fail point should succeed");
 }
 
-#[cfg_attr(feature = "tokio-runtime", tokio::test(threaded_scheduler))]
+#[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[function_name::named]
 async fn connection_error_during_establishment() {
@@ -209,7 +209,7 @@ async fn connection_error_during_establishment() {
         .expect("closed event with error reason should have been seen");
 }
 
-#[cfg_attr(feature = "tokio-runtime", tokio::test(threaded_scheduler))]
+#[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[function_name::named]
 async fn connection_error_during_operation() {

src/error.rs

@@ -35,16 +35,12 @@ pub struct Error {
 }
 
 impl Error {
-    pub(crate) fn new(e: Arc<ErrorKind>) -> Error {
-        Error {
-            kind: e,
-            labels: Vec::new(),
-        }
-    }
-
     pub(crate) fn pool_cleared_error(address: &StreamAddress) -> Self {
         ErrorKind::ConnectionPoolClearedError {
-            message: format!("Conneciton pool for {} cleared during operation execution", address)
+            message: format!(
+                "Connection pool for {} cleared during operation execution",
+                address
+            ),
         }
         .into()
     }
@@ -68,20 +64,6 @@ impl Error {
         Error::authentication_error(mechanism_name, "invalid server response")
     }
 
-    /// Attempts to get the `std::io::Error` from this `Error`.
-    /// If there are other references to the underlying `Arc`, or if the `ErrorKind` is not `Io`,
-    /// then the original error is returned as a custom `std::io::Error`.
-    pub(crate) fn into_io_error(self) -> std::io::Error {
-        match Arc::try_unwrap(self.kind) {
-            Ok(ErrorKind::Io(io_error)) => io_error,
-            Ok(other_error_kind) => {
-                let error: Error = other_error_kind.into();
-                std::io::Error::new(std::io::ErrorKind::Other, Box::new(error))
-            }
-            Err(e) => std::io::Error::new(std::io::ErrorKind::Other, Box::new(Error::new(e))),
-        }
-    }
-
     /// Whether this error is an "ns not found" error or not.
     pub(crate) fn is_ns_not_found(&self) -> bool {
         matches!(self.kind.as_ref(), ErrorKind::CommandError(err) if err.code == 26)
@@ -129,10 +111,13 @@ impl Error {
 
     /// Whether an error originated from the server.
     pub(crate) fn is_server_error(&self) -> bool {
-        matches!(self.kind.as_ref(), ErrorKind::AuthenticationError { .. }
-        | ErrorKind::BulkWriteError(_)
-        | ErrorKind::CommandError(_)
-        | ErrorKind::WriteError(_))
+        matches!(
+            self.kind.as_ref(),
+            ErrorKind::AuthenticationError { .. }
+                | ErrorKind::BulkWriteError(_)
+                | ErrorKind::CommandError(_)
+                | ErrorKind::WriteError(_)
+        )
     }
 
     /// Returns the labels for this error.
@@ -153,7 +138,9 @@ impl Error {
 
     /// Whether this error contains the specified label.
     pub fn contains_label<T: AsRef<str>>(&self, label: T) -> bool {
-        self.labels().iter().any(|actual_label| actual_label.as_str() == label.as_ref())
+        self.labels()
+            .iter()
+            .any(|actual_label| actual_label.as_str() == label.as_ref())
     }
 
     /// Returns a copy of this Error with the specified label added.
@@ -334,7 +321,7 @@ pub enum ErrorKind {
     /// A timeout occurred before a Tokio task could be completed.
     #[cfg(feature = "tokio-runtime")]
     #[error(display = "{}", _0)]
-    TokioTimeoutElapsed(#[error(source)] tokio::time::Elapsed),
+    TokioTimeoutElapsed(#[error(source)] tokio::time::error::Elapsed),
 
     #[error(display = "{}", _0)]
     RustlsConfig(#[error(source)] rustls::TLSError),
@@ -372,7 +359,10 @@ impl ErrorKind {
     }
 
     pub(crate) fn is_network_error(&self) -> bool {
-        matches!(self, ErrorKind::Io(..) | ErrorKind::ConnectionPoolClearedError { .. })
+        matches!(
+            self,
+            ErrorKind::Io(..) | ErrorKind::ConnectionPoolClearedError { .. }
+        )
     }
 
     /// Gets the code/message tuple from this error, if applicable. In the case of write errors, the
@@ -396,13 +386,14 @@ impl ErrorKind {
     pub(crate) fn code_name(&self) -> Option<&str> {
         match self {
             ErrorKind::CommandError(ref cmd_err) => Some(cmd_err.code_name.as_str()),
-            ErrorKind::WriteError(ref failure) => {
-                match failure {
-                    WriteFailure::WriteConcernError(ref wce) => Some(wce.code_name.as_str()),
-                    WriteFailure::WriteError(ref we) => we.code_name.as_deref(),
-                }
-            }
-            ErrorKind::BulkWriteError(ref bwe) => bwe.write_concern_error.as_ref().map(|wce| wce.code_name.as_str()),
+            ErrorKind::WriteError(ref failure) => match failure {
+                WriteFailure::WriteConcernError(ref wce) => Some(wce.code_name.as_str()),
+                WriteFailure::WriteError(ref we) => we.code_name.as_deref(),
+            },
+            ErrorKind::BulkWriteError(ref bwe) => bwe
+                .write_concern_error
+                .as_ref()
+                .map(|wce| wce.code_name.as_str()),
             _ => None,
         }
     }

src/runtime/mod.rs

@@ -118,7 +118,7 @@ impl AsyncRuntime {
     pub(crate) async fn delay_for(self, delay: Duration) {
         #[cfg(feature = "tokio-runtime")]
         {
-            tokio::time::delay_for(delay).await
+            tokio::time::sleep(delay).await
         }
 
         #[cfg(feature = "async-std-runtime")]