Upgrade to Tokio 1.0

Cargo.toml

@@ -17,15 +17,17 @@ default = ["tokio", "unstable"]
 unstable = []
 
 [dependencies]
-futures = "0.3"
-tokio = { version = "0.3", features=["rt", "time"], optional = true }
+futures-core = "0.3"
+futures-channel = { version = "0.3", features=["sink"] }
+futures-util = { version = "0.3", features=["sink"] }
+tokio = { version = "1", features=["rt", "rt-multi-thread", "time"], optional = true }
 async-std = { version = "1.0", features=["unstable"], optional = true }
 async-trait = "0.1"
 futures-timer = "3.0.2"
 log = "0.4"
 
 [dev-dependencies]
-tokio = { version = "0.3", features = ["full"] }
+tokio = { version = "1", features = ["full"] }
 tokio-postgres = "=0.6.0"
 env_logger = "0.7"
 tide = "0.5"

src/lib.rs

@@ -81,13 +81,13 @@ mod time;
 pub use async_trait::async_trait;
 pub use config::Builder;
 use config::{Config, InternalConfig, ShareConfig};
-use futures::channel::mpsc::{self, Receiver, Sender};
-use futures::channel::oneshot::{self, Sender as ReqSender};
-use futures::lock::{Mutex, MutexGuard};
-use futures::select;
-use futures::FutureExt;
-use futures::SinkExt;
-use futures::StreamExt;
+use futures_channel::mpsc::{self, Receiver, Sender};
+use futures_channel::oneshot::{self, Sender as ReqSender};
+use futures_util::lock::{Mutex, MutexGuard};
+use futures_util::select;
+use futures_util::FutureExt;
+use futures_util::SinkExt;
+use futures_util::StreamExt;
 pub use spawn::spawn;
 use std::collections::HashMap;
 use std::error;
@@ -818,4 +818,3 @@ impl<M: Manager> DerefMut for Connection<M> {
         self.conn.as_mut().unwrap().raw.as_mut().unwrap()
     }
 }
-

src/runtime.rs

@@ -3,7 +3,7 @@
 
 pub use runtime::{DefaultExecutor, Runtime, TaskExecutor};
 
-use futures::Future;
+use std::future::Future;
 use std::pin::Pin;
 
 // A new type exports the default executor of Tokio..
@@ -20,24 +20,20 @@ pub trait Executor: Send + Sync + 'static + Clone {
     fn spawn(&mut self, future: Pin<Box<dyn Future<Output = ()> + Send>>);
 }
 
-#[cfg(all(
-    feature = "tokio",
-    not(feature = "async-std")
-))]
+#[cfg(all(feature = "tokio", not(feature = "async-std")))]
 mod runtime {
     use super::*;
 
     /// Wrapper of the Tokio Runtime
-    pub struct Runtime{
+    pub struct Runtime {
         rt: tokio::runtime::Runtime,
         spawner: TaskExecutor,
     }
-
 
     impl Runtime {
         /// Creates a new Runtime
         pub fn new() -> Option<Self> {
-            Some(Runtime{
+            Some(Runtime {
                 rt: tokio::runtime::Runtime::new().unwrap(),
                 spawner: TaskExecutor,
             })
@@ -67,7 +63,6 @@ mod runtime {
         }
     }
 
-
     /// Simple handler for spawning task
     #[derive(Clone)]
     pub struct TaskExecutor;
@@ -101,7 +96,6 @@ mod runtime {
     }
 }
 
-
 #[cfg(all(feature = "async-std"))]
 mod runtime {
     use super::*;
@@ -147,7 +141,6 @@ mod runtime {
         }
     }
 
-
     #[derive(Clone)]
     pub struct DefaultExecutor;
 

src/time.rs

@@ -1,5 +1,5 @@
 use crate::Error;
-use futures::FutureExt;
+use futures_util::{select, FutureExt};
 use std::future::Future;
 use std::time::Duration;
 pub use time::{delay_for, interval};
@@ -8,7 +8,7 @@ pub(crate) async fn timeout<F, T, E>(duration: Duration, f: F) -> Result<T, Erro
 where
     F: Future<Output = Result<T, Error<E>>>,
 {
-    futures::select! {
+    select! {
         () = delay_for(duration).fuse() => Err(Error::Timeout),
         rsp = f.fuse() => rsp,
     }

tests/mobc.rs

@@ -967,12 +967,12 @@ fn test_timeout_when_db_has_gone() {
         type Error = TestError;
 
         async fn connect(&self) -> Result<Self::Connection, Self::Error> {
-            futures::future::pending::<()>().await;
+            futures_util::future::pending::<()>().await;
             Ok(Connection)
         }
 
         async fn check(&self, conn: Self::Connection) -> Result<Self::Connection, Self::Error> {
-            futures::future::pending::<()>().await;
+            futures_util::future::pending::<()>().await;
             Ok(conn)
         }
     }
@@ -1007,7 +1007,7 @@ fn test_timeout_when_db_has_gone2() {
 
         async fn connect(&self) -> Result<Self::Connection, Self::Error> {
             if self.0.load(Ordering::Relaxed) > 0 {
-                futures::future::pending::<()>().await;
+                futures_util::future::pending::<()>().await;
             } else {
                 self.0.fetch_add(1, Ordering::Relaxed);
             }
@@ -1016,7 +1016,7 @@ fn test_timeout_when_db_has_gone2() {
 
         async fn check(&self, conn: Self::Connection) -> Result<Self::Connection, Self::Error> {
             if self.0.load(Ordering::Relaxed) > 0 {
-                futures::future::pending::<()>().await;
+                futures_util::future::pending::<()>().await;
             }
             Ok(conn)
         }