Repository: blackbeam/mysql_async (language: rust)

Issue:
Support for std::future and async/await
`std::future` is now in 1.36.0 and async/await is going to (hopefully) be stabilized in 1.38.0. Opening up this issue to track progress on upgrading mysql_async to the new syntax.

Is anybody working on this already? We might be able to put some hours to the refactoring in Prisma later on.


Discussion / hints:
I'd say it would be a good goal to change the `Queryable` to take reference in its functions. Would make it much easier to abstract with other databases, such as PostgreSQL.
Tokio master is getting close to the alpha release of `std::futures` refactoring. mysql_async needs the tokio-uds crate that still has some work to do, but there's a pull request already.

https://github.com/tokio-rs/tokio/pull/1227

tokio-named-pipes for Windows support needs to be refactored by somebody with a Windows computer.
Spent the day to at least get the types right. I guess I'll be working on this at least for now...

https://github.com/pimeys/mysql_async/tree/async_await
So spent a couple of days to see how much work it would be to have Client/Transaction taking `&self` instead of `self` and upgrading to the `std::future`. Noticed the architecture should change into something similar what `tokio-postgres` is doing and decided it would take too much time now for me to try changing this crate.
@blackbeam do you think this is something you'll have the spare cycles to take on any time soon?
Some learnings from those two days of work I did a couple of months ago, and reading what @sfackler has done in the std-futures branch in tokio-postgres:

- Internally use a oneshot channel to create a future. Then in cases like transactions, when dropped the transaction can still be rolled back without having an executor available in `Drop`.
- The codec is a great idea, same for the common types.
- The sync mysql crate is a much easier starting point than this one.
- Lots of work, especially with the AsyncRead/AsyncWrite conversion
- Most of the work is just adding `Pin`, but certain features are now missing in the new futures and need to be redesigned completely.

Fix patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 0ea7b768..d9828e7b 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -8,9 +8,6 @@ jobs:
         type: string
       flavor:
         type: string
-      test-ssl:
-        type: boolean
-        default: false
       test-unstable:
         type: boolean
         default: false
@@ -34,13 +31,13 @@ jobs:
             - /usr/local/cargo/registry/index
       - restore_cache:
           key: deps-<< parameters.rust >>-{{ checksum "Cargo.lock" }}
-      - run: cargo test
-      - when:
-          condition: << parameters.test-ssl >>
-          steps:
-            - run: cargo test --features ssl
-      - run: rustup component add rustfmt
-      - run: cargo fmt -- --check
+#      - run: cargo test
+#      - when:
+#          condition: << parameters.test-ssl >>
+#          steps:
+#            - run: cargo test --features ssl
+#      - run: rustup component add rustfmt
+#      - run: cargo fmt -- --check
       - save_cache:
           key: deps-<< parameters.rust >>-{{ checksum "Cargo.lock" }}
           paths:
@@ -48,51 +45,55 @@ jobs:
       - when:
           condition: << parameters.test-unstable >>
           steps:
-            - run:
-                name: Beta
-                command: |
-                  rustup toolchain install beta
-                  cargo +beta test
+#            - run:
+#                name: Beta
+#                command: |
+#                  rustup toolchain install beta
+#                  cargo +beta test
             - run:
                 name: Nightly
                 command: |
                   rustup toolchain install nightly
                   cargo +nightly test
+            - run:
+                name: Nightly SSL
+                command: cargo +nightly test
+                environment:
+                  SSL: 1
 
 
 workflows:
   test:
     jobs:
-      - linux:
-          name: stable-mysql-5.6
-          rust: latest
-          flavor: mysql
-          version: "5.6"
-      - linux:
-          name: stable-mysql-5.7
-          rust: latest
-          flavor: mysql
-          version: "5.7"
-          test-ssl: true
+#      - linux:
+#          name: stable-mysql-5.6
+#          rust: latest
+#          flavor: mysql
+#          version: "5.6"
+#      - linux:
+#          name: stable-mysql-5.7
+#          rust: latest
+#          flavor: mysql
+#          version: "5.7"
+#          test-ssl: true
       - linux:
           name: stable-mysql-8.0
           rust: latest
           flavor: mysql
           version: "8.0"
-          test-ssl: true
           test-unstable: true
-      - linux:
-          name: stable-mariadb-5.5
-          rust: latest
-          flavor: mariadb
-          version: "5.5"
-      - linux:
-          name: stable-mariadb-10.3
-          rust: latest
-          flavor: mariadb
-          version: "10.3"
-      - linux:
-          name: stable-mariadb-10.4
-          rust: latest
-          flavor: mariadb
-          version: "10.4"
+#      - linux:
+#          name: stable-mariadb-5.5
+#          rust: latest
+#          flavor: mariadb
+#          version: "5.5"
+#      - linux:
+#          name: stable-mariadb-10.3
+#          rust: latest
+#          flavor: mariadb
+#          version: "10.3"
+#      - linux:
+#          name: stable-mariadb-10.4
+#          rust: latest
+#          flavor: mariadb
+#          version: "10.4"
diff --git a/Cargo.toml b/Cargo.toml
index 90193d43..c060e255 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -7,7 +7,7 @@ license = "MIT/Apache-2.0"
 name = "mysql_async"
 readme = "README.md"
 repository = "https://github.com/blackbeam/mysql_async"
-version = "0.20.0"
+version = "0.21.0-alpha.1"
 exclude = ["test/*"]
 edition = "2018"
 
@@ -19,19 +19,21 @@ crossbeam = "0.7"
 failure = "0.1"
 failure_derive = "0.1"
 fnv = "1"
-futures = "0.1.18"
+futures-core-preview = "0.3.0-alpha.18"
+futures-util-preview = "0.3.0-alpha.18"
+futures-sink-preview = "0.3.0-alpha.18"
+mio-named-pipes = "0.1.6"
 mysql_common = "0.18"
-native-tls = { version = "0.2", optional = true }
+native-tls = "0.2"
 percent-encoding = "2.1.0"
+pin-project = { version = "0.4.0-alpha.11", features = ["project_attr"] }
 regex = "1"
 serde = "1"
 serde_json = "1"
-tokio = "0.1.9"
-tokio-codec = "0.1"
-tokio-io = "0.1"
-tokio-named-pipes = "0.1"
-tokio-uds = "0.2"
-tokio-sync = "0.1"
+# we only need rt-full for tokio::spawn
+tokio = { version = "0.2.0-alpha.5", default-features = false, features = ["codec", "io", "net", "sync", "fs", "rt-full"] }
+tokio-tls = "0.3.0-alpha.5"
+tokio-net = "0.2.0-alpha.5"
 twox-hash = "1"
 url = "2.1"
 
@@ -40,7 +42,6 @@ lazy_static = "1"
 
 [features]
 nightly = []
-ssl = ['native-tls']
 
 [lib]
 name = "mysql_async"
diff --git a/appveyor.yml b/appveyor.yml
index 91483cc4..19af3fdb 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -12,22 +12,22 @@ environment:
       PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
     - TARGET: nightly-i686-pc-windows-gnu
       PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
-    - TARGET: beta-x86_64-pc-windows-msvc
-      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
-    - TARGET: beta-i686-pc-windows-msvc
-      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
-    - TARGET: beta-x86_64-pc-windows-gnu
-      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
-    - TARGET: beta-i686-pc-windows-gnu
-      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
-    - TARGET: 1.34.0-x86_64-pc-windows-msvc
-      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
-    - TARGET: 1.34.0-i686-pc-windows-msvc
-      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
-    - TARGET: 1.34.0-x86_64-pc-windows-gnu
-      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
-    - TARGET: 1.34.0-i686-pc-windows-gnu
-      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
+#    - TARGET: beta-x86_64-pc-windows-msvc
+#      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
+#    - TARGET: beta-i686-pc-windows-msvc
+#      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
+#    - TARGET: beta-x86_64-pc-windows-gnu
+#      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
+#    - TARGET: beta-i686-pc-windows-gnu
+#      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
+#    - TARGET: 1.34.0-x86_64-pc-windows-msvc
+#      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
+#    - TARGET: 1.34.0-i686-pc-windows-msvc
+#      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
+#    - TARGET: 1.34.0-x86_64-pc-windows-gnu
+#      PATH: C:\msys64\mingw64\bin\;c:\rust\bin;%PATH%
+#    - TARGET: 1.34.0-i686-pc-windows-gnu
+#      PATH: C:\msys64\mingw32\bin\;c:\rust\bin;%PATH%
 services: mysql
 install:
   - cmd: >-
diff --git a/rustfmt.toml b/rustfmt.toml
new file mode 100644
index 00000000..32a9786f
--- /dev/null
+++ b/rustfmt.toml
@@ -0,0 +1,1 @@
+edition = "2018"
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
index 38500186..2e0f804c 100644
--- a/src/conn/mod.rs
+++ b/src/conn/mod.rs
@@ -8,7 +8,6 @@
 
 pub use mysql_common::named_params;
 
-use futures::future::{err, loop_fn, ok, Either::*, Future, IntoFuture, Loop};
 use mysql_common::{
     crypto,
     packets::{
@@ -17,7 +16,7 @@ use mysql_common::{
     },
 };
 
-use std::{fmt, mem, str::FromStr, sync::Arc};
+use std::{fmt, future::Future, mem, pin::Pin, str::FromStr, sync::Arc};
 
 use crate::{
     conn::{pool::Pool, stmt_cache::StmtCache},
@@ -29,7 +28,7 @@ use crate::{
     opts::Opts,
     queryable::{query_result, BinaryProtocol, Queryable, TextProtocol},
     time::SteadyTime,
-    BoxFuture, Column, MyFuture, OptsBuilder,
+    Column, OptsBuilder,
 };
 
 pub mod pool;
@@ -37,9 +36,6 @@ pub mod stmt_cache;
 
 /// Helper function that asynchronously disconnects connection on the default tokio executor.
 fn disconnect(mut conn: Conn) {
-    use tokio::executor::{DefaultExecutor, Executor};
-    let mut executor = DefaultExecutor::current();
-
     let disconnected = conn.inner.disconnected;
 
     // Mark conn as disconnected.
@@ -47,9 +43,13 @@ fn disconnect(mut conn: Conn) {
 
     if !disconnected {
         // Server will report broken connection if spawn fails.
-        let _ = executor.spawn(Box::new(
-            conn.cleanup().and_then(Conn::disconnect).map_err(drop),
-        ));
+        use tokio::executor::Executor;
+        // this might fail if, say, the runtime is shutting down, but we've done what we could
+        let _ = tokio::executor::DefaultExecutor::current().spawn(Box::pin(async move {
+            if let Ok(conn) = conn.cleanup().await {
+                let _ = conn.disconnect().await;
+            }
+        }));
     }
 }
 
@@ -89,6 +89,7 @@ impl fmt::Debug for ConnInner {
             .field("pool", &self.pool)
             .field("has result", &self.has_result.is_some())
             .field("in transaction", &self.in_transaction)
+            .field("stream", &self.stream)
             .field("options", &self.opts)
             .finish()
     }
@@ -144,9 +145,9 @@ impl Conn {
         self.get_affected_rows()
     }
 
-    fn close(mut self) -> impl MyFuture<()> {
+    async fn close(mut self) -> Result<()> {
         self.inner.disconnected = true;
-        self.cleanup().and_then(Conn::disconnect)
+        self.cleanup().await?.disconnect().await
     }
 
     fn is_secure(&self) -> bool {
@@ -182,37 +183,32 @@ impl Conn {
         }
     }
 
-    fn handle_handshake(self) -> impl MyFuture<Conn> {
-        self.read_packet().and_then(move |(mut conn, packet)| {
-            parse_handshake_packet(&*packet.0)
-                .map_err(Error::from)
-                .and_then(|handshake| {
-                    conn.inner.nonce = {
-                        let mut nonce = Vec::from(handshake.scramble_1_ref());
-                        nonce.extend_from_slice(handshake.scramble_2_ref().unwrap_or(&[][..]));
-                        nonce
-                    };
+    async fn handle_handshake(self) -> Result<Conn> {
+        let (mut conn, packet) = self.read_packet().await?;
+        let handshake = parse_handshake_packet(&*packet.0)?;
+        conn.inner.nonce = {
+            let mut nonce = Vec::from(handshake.scramble_1_ref());
+            nonce.extend_from_slice(handshake.scramble_2_ref().unwrap_or(&[][..]));
+            nonce
+        };
 
-                    conn.inner.capabilities =
-                        handshake.capabilities() & conn.inner.opts.get_capabilities();
-                    conn.inner.version = handshake.server_version_parsed().unwrap_or((0, 0, 0));
-                    conn.inner.id = handshake.connection_id();
-                    conn.inner.status = handshake.status_flags();
-                    conn.inner.auth_plugin = match handshake.auth_plugin() {
-                        Some(AuthPlugin::MysqlNativePassword) => AuthPlugin::MysqlNativePassword,
-                        Some(AuthPlugin::CachingSha2Password) => AuthPlugin::CachingSha2Password,
-                        Some(AuthPlugin::Other(ref name)) => {
-                            let name = String::from_utf8_lossy(name).into();
-                            return Err(DriverError::UnknownAuthPlugin { name }.into());
-                        }
-                        None => AuthPlugin::MysqlNativePassword,
-                    };
-                    Ok(conn)
-                })
-        })
+        conn.inner.capabilities = handshake.capabilities() & conn.inner.opts.get_capabilities();
+        conn.inner.version = handshake.server_version_parsed().unwrap_or((0, 0, 0));
+        conn.inner.id = handshake.connection_id();
+        conn.inner.status = handshake.status_flags();
+        conn.inner.auth_plugin = match handshake.auth_plugin() {
+            Some(AuthPlugin::MysqlNativePassword) => AuthPlugin::MysqlNativePassword,
+            Some(AuthPlugin::CachingSha2Password) => AuthPlugin::CachingSha2Password,
+            Some(AuthPlugin::Other(ref name)) => {
+                let name = String::from_utf8_lossy(name).into();
+                return Err(DriverError::UnknownAuthPlugin { name }.into());
+            }
+            None => AuthPlugin::MysqlNativePassword,
+        };
+        Ok(conn)
     }
 
-    fn switch_to_ssl_if_needed(self) -> impl MyFuture<Conn> {
+    async fn switch_to_ssl_if_needed(self) -> Result<Conn> {
         if self
             .inner
             .opts
@@ -220,25 +216,22 @@ impl Conn {
             .contains(CapabilityFlags::CLIENT_SSL)
         {
             let ssl_request = SslRequest::new(self.inner.capabilities);
-            let fut = self.write_packet(ssl_request.as_ref()).and_then(|conn| {
-                let ssl_opts = conn
-                    .get_opts()
-                    .get_ssl_opts()
-                    .cloned()
-                    .expect("unreachable");
-                let domain = conn.get_opts().get_ip_or_hostname().into();
-                let (streamless, stream) = conn.take_stream();
-                stream
-                    .make_secure(domain, ssl_opts)
-                    .map(move |stream| streamless.return_stream(stream))
-            });
-            A(fut)
+            let conn = self.write_packet(ssl_request.as_ref()).await?;
+            let ssl_opts = conn
+                .get_opts()
+                .get_ssl_opts()
+                .cloned()
+                .expect("unreachable");
+            let domain = conn.get_opts().get_ip_or_hostname().into();
+            let (streamless, stream) = conn.take_stream();
+            let stream = stream.make_secure(domain, ssl_opts).await?;
+            Ok(streamless.return_stream(stream))
         } else {
-            B(ok(self))
+            Ok(self)
         }
     }
 
-    fn do_handshake_response(self) -> impl MyFuture<Conn> {
+    async fn do_handshake_response(self) -> Result<Conn> {
         let auth_data = self
             .inner
             .auth_plugin
@@ -253,13 +246,13 @@ impl Conn {
             self.get_capabilities(),
         );
 
-        self.write_packet(handshake_response.as_ref())
+        self.write_packet(handshake_response.as_ref()).await
     }
 
-    fn perform_auth_switch(
+    async fn perform_auth_switch(
         mut self,
         auth_switch_request: AuthSwitchRequest<'_>,
-    ) -> BoxFuture<Conn> {
+    ) -> Result<Conn> {
         if !self.inner.auth_switched {
             self.inner.auth_switched = true;
             self.inner.nonce = auth_switch_request.plugin_data().into();
@@ -269,161 +262,133 @@ impl Conn {
                 .auth_plugin
                 .gen_data(self.inner.opts.get_pass(), &*self.inner.nonce)
                 .unwrap_or_else(Vec::new);
-            let fut = self.write_packet(plugin_data).and_then(Conn::continue_auth);
-            // We'll box it to avoid recursion.
-            Box::new(fut)
+            self.write_packet(plugin_data).await?.continue_auth().await
         } else {
             unreachable!("auth_switched flag should be checked by caller")
         }
     }
 
-    fn continue_auth(self) -> impl MyFuture<Conn> {
-        match self.inner.auth_plugin {
-            AuthPlugin::MysqlNativePassword => A(self.continue_mysql_native_password_auth()),
-            AuthPlugin::CachingSha2Password => B(self.continue_caching_sha2_password_auth()),
-            _ => unreachable!(),
-        }
+    fn continue_auth(self) -> Pin<Box<dyn Future<Output = Result<Conn>> + Send>> {
+        // NOTE: we need to box this since it may recurse
+        // see https://github.com/rust-lang/rust/issues/46415#issuecomment-528099782
+        Box::pin(async move {
+            match self.inner.auth_plugin {
+                AuthPlugin::MysqlNativePassword => self.continue_mysql_native_password_auth().await,
+                AuthPlugin::CachingSha2Password => self.continue_caching_sha2_password_auth().await,
+                _ => unreachable!(),
+            }
+        })
     }
 
-    fn continue_caching_sha2_password_auth(self) -> impl MyFuture<Conn> {
-        self.read_packet()
-            .and_then(|(conn, packet)| match packet.as_ref().get(0) {
-                Some(0x01) => match packet.as_ref().get(1) {
-                    Some(0x03) => {
-                        // auth ok
-                        A(conn.drop_packet())
-                    }
-                    Some(0x04) => {
-                        let mut pass = conn
-                            .inner
-                            .opts
-                            .get_pass()
-                            .map(Vec::from)
-                            .unwrap_or_default();
-                        pass.push(0);
-                        let fut = if conn.is_secure() {
-                            A(conn.write_packet(&*pass))
-                        } else {
-                            B(conn
-                                .write_packet(&[0x02][..])
-                                .and_then(Conn::read_packet)
-                                .and_then(move |(conn, packet)| {
-                                    let key = &packet.as_ref()[1..];
-                                    for (i, byte) in pass.iter_mut().enumerate() {
-                                        *byte ^= conn.inner.nonce[i % conn.inner.nonce.len()];
-                                    }
-                                    let encrypted_pass = crypto::encrypt(&*pass, key);
-                                    conn.write_packet(&*encrypted_pass)
-                                }))
-                        };
-                        B(A(fut.and_then(Conn::drop_packet)))
-                    }
-                    _ => B(B(A(err(DriverError::UnexpectedPacket {
-                        payload: packet.as_ref().into(),
-                    }
-                    .into())))),
-                },
-                Some(0xfe) if !conn.inner.auth_switched => {
-                    let fut = parse_auth_switch_request(packet.as_ref())
-                        .map(AuthSwitchRequest::into_owned)
-                        .map_err(Error::from)
-                        .into_future()
-                        .and_then(|auth_switch_request| {
-                            conn.perform_auth_switch(auth_switch_request)
-                        });
-                    B(B(B(A(fut))))
+    async fn continue_caching_sha2_password_auth(self) -> Result<Conn> {
+        let (conn, packet) = self.read_packet().await?;
+        match packet.as_ref().get(0) {
+            Some(0x01) => match packet.as_ref().get(1) {
+                Some(0x03) => {
+                    // auth ok
+                    conn.drop_packet().await
                 }
-                _ => B(B(B(B(err(DriverError::UnexpectedPacket {
+                Some(0x04) => {
+                    let mut pass = conn
+                        .inner
+                        .opts
+                        .get_pass()
+                        .map(Vec::from)
+                        .unwrap_or_default();
+                    pass.push(0);
+
+                    let conn = if conn.is_secure() {
+                        conn.write_packet(&*pass).await?
+                    } else {
+                        let conn = conn.write_packet(&[0x02][..]).await?;
+                        let (conn, packet) = conn.read_packet().await?;
+                        let key = &packet.as_ref()[1..];
+                        for (i, byte) in pass.iter_mut().enumerate() {
+                            *byte ^= conn.inner.nonce[i % conn.inner.nonce.len()];
+                        }
+                        let encrypted_pass = crypto::encrypt(&*pass, key);
+                        conn.write_packet(&*encrypted_pass).await?
+                    };
+                    conn.drop_packet().await
+                }
+                _ => Err(DriverError::UnexpectedPacket {
                     payload: packet.as_ref().into(),
                 }
-                .into()))))),
-            })
+                .into()),
+            },
+            Some(0xfe) if !conn.inner.auth_switched => {
+                let auth_switch_request = parse_auth_switch_request(packet.as_ref())?.into_owned();
+                conn.perform_auth_switch(auth_switch_request).await
+            }
+            _ => Err(DriverError::UnexpectedPacket {
+                payload: packet.as_ref().into(),
+            }
+            .into()),
+        }
     }
 
-    fn continue_mysql_native_password_auth(self) -> impl MyFuture<Conn> {
-        self.read_packet()
-            .and_then(|(this, packet)| match packet.0.get(0) {
-                Some(0x00) => A(ok(this)),
-                Some(0xfe) if !this.inner.auth_switched => {
-                    let fut = parse_auth_switch_request(packet.as_ref())
-                        .map(AuthSwitchRequest::into_owned)
-                        .map_err(Error::from)
-                        .into_future()
-                        .and_then(|auth_switch_request| {
-                            this.perform_auth_switch(auth_switch_request)
-                        });
-                    B(A(fut))
-                }
-                _ => B(B(err(
-                    DriverError::UnexpectedPacket { payload: packet.0 }.into()
-                ))),
-            })
+    async fn continue_mysql_native_password_auth(self) -> Result<Conn> {
+        let (this, packet) = self.read_packet().await?;
+        match packet.0.get(0) {
+            Some(0x00) => Ok(this),
+            Some(0xfe) if !this.inner.auth_switched => {
+                let auth_switch_request = parse_auth_switch_request(packet.as_ref())?.into_owned();
+                this.perform_auth_switch(auth_switch_request).await
+            }
+            _ => Err(DriverError::UnexpectedPacket { payload: packet.0 }.into()),
+        }
     }
 
-    fn drop_packet(self) -> impl MyFuture<Conn> {
-        self.read_packet().map(|(conn, _)| conn)
+    async fn drop_packet(self) -> Result<Conn> {
+        Ok(self.read_packet().await?.0)
     }
 
-    fn run_init_commands(self) -> impl MyFuture<Conn> {
-        let init = self
-            .inner
-            .opts
-            .get_init()
-            .iter()
-            .map(Clone::clone)
-            .collect();
-
-        loop_fn(
-            (init, self),
-            |(mut init, conn): (Vec<String>, Conn)| match init.pop() {
-                None => A(ok(Loop::Break(conn))),
-                Some(query) => {
-                    let fut = conn
-                        .drop_query(query)
-                        .map(|conn| Loop::Continue((init, conn)));
-                    B(fut)
-                }
-            },
-        )
+    async fn run_init_commands(self) -> Result<Conn> {
+        let mut init: Vec<_> = self.inner.opts.get_init().iter().cloned().collect();
+
+        let mut conn = self;
+        while let Some(query) = init.pop() {
+            conn = conn.drop_query(query).await?;
+        }
+        Ok(conn)
     }
 
     /// Returns future that resolves to `Conn`.
-    pub fn new<T: Into<Opts>>(opts: T) -> impl MyFuture<Conn> {
+    pub async fn new<T: Into<Opts>>(opts: T) -> Result<Conn> {
         let opts = opts.into();
         let mut conn = Conn::empty(opts.clone());
 
         let stream = if let Some(path) = opts.get_socket() {
-            A(Stream::connect_socket(path.to_owned()))
+            Stream::connect_socket(path.to_owned()).await?
         } else {
-            B(Stream::connect_tcp((
-                opts.get_ip_or_hostname(),
-                opts.get_tcp_port(),
-            )))
+            Stream::connect_tcp((opts.get_ip_or_hostname(), opts.get_tcp_port())).await?
         };
 
-        stream
-            .map(move |stream| {
-                conn.inner.stream = Some(stream);
-                conn
-            })
-            .and_then(Conn::setup_stream)
-            .and_then(Conn::handle_handshake)
-            .and_then(Conn::switch_to_ssl_if_needed)
-            .and_then(Conn::do_handshake_response)
-            .and_then(Conn::continue_auth)
-            .and_then(Conn::read_socket)
-            .and_then(Conn::reconnect_via_socket_if_needed)
-            .and_then(Conn::read_max_allowed_packet)
-            .and_then(Conn::read_wait_timeout)
-            .and_then(Conn::run_init_commands)
+        conn.inner.stream = Some(stream);
+        conn.setup_stream()?
+            .handle_handshake()
+            .await?
+            .switch_to_ssl_if_needed()
+            .await?
+            .do_handshake_response()
+            .await?
+            .continue_auth()
+            .await?
+            .read_socket()
+            .await?
+            .reconnect_via_socket_if_needed()
+            .await?
+            .read_max_allowed_packet()
+            .await?
+            .read_wait_timeout()
+            .await?
+            .run_init_commands()
+            .await
     }
 
     /// Returns future that resolves to `Conn`.
-    pub fn from_url<T: AsRef<str>>(url: T) -> impl MyFuture<Conn> {
-        Opts::from_str(url.as_ref())
-            .map_err(Error::from)
-            .into_future()
-            .and_then(Conn::new)
+    pub async fn from_url<T: AsRef<str>>(url: T) -> Result<Conn> {
+        Conn::new(Opts::from_str(url.as_ref())?).await
     }
 
     /// Will try to connect via socket using socket address in `self.inner.socket`.
@@ -431,52 +396,50 @@ impl Conn {
     /// Returns new connection on success or self on error.
     ///
     /// Won't try to reconnect if socket connection is already enforced in `Opts`.
-    fn reconnect_via_socket_if_needed(self) -> Box<dyn MyFuture<Conn>> {
-        if let Some(socket) = self.inner.socket.as_ref() {
-            let opts = self.inner.opts.clone();
-            if opts.get_socket().is_none() {
-                let mut builder = OptsBuilder::from_opts(opts);
-                builder.socket(Some(&**socket));
-                let fut = Conn::new(builder).then(|result| match result {
-                    Ok(conn) => Ok(conn),
-                    Err(_) => Ok(self),
-                });
-                return Box::new(fut);
+    fn reconnect_via_socket_if_needed(self) -> Pin<Box<dyn Future<Output = Result<Conn>> + Send>> {
+        // NOTE: we need to box this since it may recurse
+        // see https://github.com/rust-lang/rust/issues/46415#issuecomment-528099782
+        Box::pin(async move {
+            if let Some(socket) = self.inner.socket.as_ref() {
+                let opts = self.inner.opts.clone();
+                if opts.get_socket().is_none() {
+                    let mut builder = OptsBuilder::from_opts(opts);
+                    builder.socket(Some(&**socket));
+                    match Conn::new(builder).await {
+                        Ok(conn) => return Ok(conn),
+                        Err(_) => return Ok(self),
+                    }
+                }
             }
-        }
-        Box::new(ok(self))
+            Ok(self)
+        })
     }
 
     /// Returns future that resolves to `Conn` with socket address stored in it.
     ///
     /// Do nothing if socket address is already in `Opts` or if `prefer_socket` is `false`.
-    fn read_socket(self) -> impl MyFuture<Self> {
+    async fn read_socket(self) -> Result<Self> {
         if self.inner.opts.get_prefer_socket() && self.inner.socket.is_none() {
-            A(self.first("SELECT @@socket").map(|(mut this, row_opt)| {
-                this.inner.socket = row_opt.unwrap_or((None,)).0;
-                this
-            }))
+            let (mut this, row_opt) = self.first("SELECT @@socket").await?;
+            this.inner.socket = row_opt.unwrap_or((None,)).0;
+            Ok(this)
         } else {
-            B(ok(self))
+            Ok(self)
         }
     }
 
     /// Returns future that resolves to `Conn` with `max_allowed_packet` stored in it.
-    fn read_max_allowed_packet(self) -> impl MyFuture<Self> {
-        self.first("SELECT @@max_allowed_packet")
-            .map(|(mut this, row_opt)| {
-                this.inner.max_allowed_packet = row_opt.unwrap_or((1024 * 1024 * 2,)).0;
-                this
-            })
+    async fn read_max_allowed_packet(self) -> Result<Self> {
+        let (mut this, row_opt) = self.first("SELECT @@max_allowed_packet").await?;
+        this.inner.max_allowed_packet = row_opt.unwrap_or((1024 * 1024 * 2,)).0;
+        Ok(this)
     }
 
     /// Returns future that resolves to `Conn` with `wait_timeout` stored in it.
-    fn read_wait_timeout(self) -> impl MyFuture<Self> {
-        self.first("SELECT @@wait_timeout")
-            .map(|(mut this, row_opt)| {
-                this.inner.wait_timeout = row_opt.unwrap_or((28800,)).0;
-                this
-            })
+    async fn read_wait_timeout(self) -> Result<Self> {
+        let (mut this, row_opt) = self.first("SELECT @@wait_timeout").await?;
+        this.inner.wait_timeout = row_opt.unwrap_or((28800,)).0;
+        Ok(this)
     }
 
     /// Returns true if time since last io exceeds wait_timeout (or conn_ttl if specified in opts).
@@ -491,56 +454,57 @@ impl Conn {
     }
 
     /// Returns future that resolves to a `Conn` with `COM_RESET_CONNECTION` executed on it.
-    pub fn reset(self) -> impl MyFuture<Conn> {
+    pub async fn reset(self) -> Result<Conn> {
         let pool = self.inner.pool.clone();
-        let fut = if self.inner.version > (5, 7, 2) {
-            let fut = self
-                .write_command_data(consts::Command::COM_RESET_CONNECTION, &[])
-                .and_then(|conn| conn.read_packet())
-                .map(|(conn, _)| conn);
-            (ok(pool), A(fut))
+        let mut conn = if self.inner.version > (5, 7, 2) {
+            self.write_command_data(consts::Command::COM_RESET_CONNECTION, &[])
+                .await?
+                .read_packet()
+                .await?
+                .0
         } else {
-            (ok(pool), B(Conn::new(self.inner.opts.clone())))
+            Conn::new(self.inner.opts.clone()).await?
         };
-        fut.into_future().map(|(pool, mut conn)| {
-            conn.inner.stmt_cache.clear();
-            conn.inner.pool = pool;
-            conn
-        })
+
+        conn.inner.stmt_cache.clear();
+        conn.inner.pool = pool;
+        Ok(conn)
     }
 
-    fn rollback_transaction(mut self) -> impl MyFuture<Self> {
+    async fn rollback_transaction(mut self) -> Result<Self> {
         assert!(self.inner.in_transaction);
         self.inner.in_transaction = false;
-        self.drop_query("ROLLBACK")
+        self.drop_query("ROLLBACK").await
     }
 
-    fn drop_result(mut self) -> impl MyFuture<Conn> {
+    async fn drop_result(mut self) -> Result<Conn> {
         match self.inner.has_result.take() {
-            Some((columns, None)) => A(B(query_result::assemble::<_, TextProtocol>(
-                self,
-                Some(columns),
-                None,
-            )
-            .drop_result())),
-            Some((columns, cached)) => A(A(query_result::assemble::<_, BinaryProtocol>(
-                self,
-                Some(columns),
-                cached,
-            )
-            .drop_result())),
-            None => B(ok(self)),
+            Some((columns, None)) => {
+                query_result::assemble::<_, TextProtocol>(self, Some(columns), None)
+                    .drop_result()
+                    .await
+            }
+            Some((columns, cached)) => {
+                query_result::assemble::<_, BinaryProtocol>(self, Some(columns), cached)
+                    .drop_result()
+                    .await
+            }
+            None => Ok(self),
         }
     }
 
-    fn cleanup(self) -> BoxFuture<Conn> {
-        if self.inner.has_result.is_some() {
-            Box::new(self.drop_result().and_then(Self::cleanup))
-        } else if self.inner.in_transaction {
-            Box::new(self.rollback_transaction().and_then(Self::cleanup))
-        } else {
-            Box::new(ok(self))
-        }
+    fn cleanup(self) -> Pin<Box<dyn Future<Output = Result<Conn>> + Send>> {
+        // NOTE: we need to box this since it may recurse
+        // see https://github.com/rust-lang/rust/issues/46415#issuecomment-528099782
+        Box::pin(async move {
+            if self.inner.has_result.is_some() {
+                self.drop_result().await?.cleanup().await
+            } else if self.inner.in_transaction {
+                self.rollback_transaction().await?.cleanup().await
+            } else {
+                Ok(self)
+            }
+        })
     }
 }
 
@@ -656,346 +620,275 @@ impl ConnectionLike for Conn {
 
 #[cfg(test)]
 mod test {
-    use futures::Future;
-
-    #[cfg(feature = "ssl")]
-    use crate::SslOpts;
     use crate::{
-        from_row, params, prelude::*, test_misc::DATABASE_URL, Conn, OptsBuilder,
-        TransactionOptions, WhiteListFsLocalInfileHandler,
+        from_row, params, prelude::*, test_misc::get_opts, Conn, OptsBuilder, TransactionOptions,
+        WhiteListFsLocalInfileHandler,
     };
 
-    /// Same as `tokio::run`, but will panic if future panics and will return the result
-    /// of future execution.
-    fn run<F, T, U>(future: F) -> Result<T, U>
-    where
-        F: Future<Item = T, Error = U> + Send + 'static,
-        T: Send + 'static,
-        U: Send + 'static,
-    {
-        let mut runtime = tokio::runtime::Runtime::new().unwrap();
-        let result = runtime.block_on(future);
-        runtime.shutdown_on_idle().wait().unwrap();
-        result
-    }
-
-    fn get_opts() -> OptsBuilder {
-        let mut builder = OptsBuilder::from_opts(&**DATABASE_URL);
-        // to suppress warning on unused mut
-        builder.stmt_cache_size(None);
-        #[cfg(feature = "ssl")]
-        {
-            let mut ssl_opts = SslOpts::default();
-            ssl_opts.set_danger_skip_domain_validation(true);
-            ssl_opts.set_danger_accept_invalid_certs(true);
-            builder.ssl_opts(ssl_opts);
-        }
-        builder
-    }
-
     #[test]
     fn opts_should_satisfy_send_and_sync() {
         struct A<T: Sync + Send>(T);
         A(get_opts());
     }
 
-    #[test]
-    fn should_connect() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| Queryable::ping(conn))
-            .and_then(|conn| Queryable::disconnect(conn));
-
-        run(fut).unwrap();
+    #[tokio::test]
+    async fn should_connect() -> super::Result<()> {
+        Conn::new(get_opts())
+            .await?
+            .ping()
+            .await?
+            .disconnect()
+            .await?;
+        Ok(())
     }
 
     #[test]
     fn should_not_panic_if_dropped_without_tokio_runtime() {
         let fut = Conn::new(get_opts());
-        run(fut).unwrap();
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        let _connection = runtime.block_on(fut).unwrap();
+        runtime.shutdown_on_idle();
         // connection will drop here
     }
 
-    #[test]
-    fn should_execute_init_queries_on_new_connection() {
+    #[tokio::test]
+    async fn should_execute_init_queries_on_new_connection() -> super::Result<()> {
         let mut opts_builder = OptsBuilder::from_opts(get_opts());
         opts_builder.init(vec!["SET @a = 42", "SET @b = 'foo'"]);
-        let fut = Conn::new(opts_builder)
-            .and_then(|conn| Queryable::query(conn, "SELECT @a, @b"))
-            .and_then(|result| result.collect_and_drop::<(u8, String)>())
-            .and_then(|(conn, rows)| Queryable::disconnect(conn).map(|_| rows))
-            .map(|result| assert_eq!(result, vec![(42, "foo".into())]));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_reset_the_connection() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| conn.drop_exec("SELECT ?", (1,)))
-            .and_then(|conn| conn.reset())
-            .and_then(|conn| conn.drop_exec("SELECT ?", (1,)))
-            .and_then(|conn| conn.disconnect());
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_not_cache_statements_if_stmt_cache_size_is_zero() {
+        let (conn, result) = Conn::new(opts_builder)
+            .await?
+            .query("SELECT @a, @b")
+            .await?
+            .collect_and_drop::<(u8, String)>()
+            .await?;
+        conn.disconnect().await?;
+        assert_eq!(result, vec![(42, "foo".into())]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_reset_the_connection() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let conn = conn.drop_exec("SELECT ?", (1,)).await?;
+        let conn = conn.reset().await?;
+        let conn = conn.drop_exec("SELECT ?", (1,)).await?;
+        conn.disconnect().await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_not_cache_statements_if_stmt_cache_size_is_zero() -> super::Result<()> {
         let mut opts = OptsBuilder::from_opts(get_opts());
         opts.stmt_cache_size(0);
-        let fut = Conn::new(opts)
-            .and_then(|conn| conn.drop_exec("DO ?", (1,)))
-            .and_then(|conn| {
-                conn.prepare("DO 2").and_then(|stmt| {
-                    stmt.first::<_, (crate::Value,)>(())
-                        .and_then(|(stmt, _)| stmt.first::<_, (crate::Value,)>(()))
-                        .and_then(|(stmt, _)| stmt.close())
-                })
-            })
-            .and_then(|conn| {
-                conn.prep_exec("DO 3", ())
-                    .and_then(|result| result.drop_result())
-            })
-            .and_then(|conn| conn.batch_exec("DO 4", vec![(), ()]))
-            .and_then(|conn| conn.first_exec::<_, _, (u8,)>("DO 5", ()))
-            .and_then(|(conn, _)| conn.first("SHOW SESSION STATUS LIKE 'Com_stmt_close';"))
-            .and_then(|(conn, row)| {
-                assert_eq!(from_row::<(String, usize)>(row.unwrap()).1, 5);
-                conn.disconnect()
-            });
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_hold_stmt_cache_size_bound() {
+        let conn = Conn::new(opts).await?;
+        let conn = conn.drop_exec("DO ?", (1,)).await?;
+        let stmt = conn.prepare("DO 2").await?;
+        let (stmt, _) = stmt.first::<_, (crate::Value,)>(()).await?;
+        let (stmt, _) = stmt.first::<_, (crate::Value,)>(()).await?;
+        let conn = stmt.close().await?;
+        let conn = conn.prep_exec("DO 3", ()).await?.drop_result().await?;
+        let conn = conn.batch_exec("DO 4", vec![(), ()]).await?;
+        let (conn, _) = conn.first_exec::<_, _, (u8,)>("DO 5", ()).await?;
+        let (conn, row) = conn
+            .first("SHOW SESSION STATUS LIKE 'Com_stmt_close';")
+            .await?;
+        assert_eq!(from_row::<(String, usize)>(row.unwrap()).1, 5);
+        conn.disconnect().await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_hold_stmt_cache_size_bound() -> super::Result<()> {
         use crate::connection_like::ConnectionLike;
 
         let mut opts = OptsBuilder::from_opts(get_opts());
         opts.stmt_cache_size(3);
-        let fut = Conn::new(opts)
-            .and_then(|conn| conn.drop_exec("DO 1", ()))
-            .and_then(|conn| conn.drop_exec("DO 2", ()))
-            .and_then(|conn| conn.drop_exec("DO 3", ()))
-            .and_then(|conn| conn.drop_exec("DO 1", ()))
-            .and_then(|conn| conn.drop_exec("DO 4", ()))
-            .and_then(|conn| conn.drop_exec("DO 3", ()))
-            .and_then(|conn| conn.drop_exec("DO 5", ()))
-            .and_then(|conn| conn.drop_exec("DO 6", ()))
-            .and_then(|conn| conn.first("SHOW SESSION STATUS LIKE 'Com_stmt_close';"))
-            .and_then(|(conn, row_opt)| {
-                let (_, count): (String, usize) = row_opt.unwrap();
-                assert_eq!(count, 3);
-                let order = conn
-                    .stmt_cache_ref()
-                    .iter()
-                    .map(Clone::clone)
-                    .collect::<Vec<String>>();
-                assert_eq!(order, &["DO 3", "DO 5", "DO 6"]);
-                conn.disconnect()
-            });
-
-        run(fut).unwrap();
+        let conn = Conn::new(opts)
+            .await?
+            .drop_exec("DO 1", ())
+            .await?
+            .drop_exec("DO 2", ())
+            .await?
+            .drop_exec("DO 3", ())
+            .await?
+            .drop_exec("DO 1", ())
+            .await?
+            .drop_exec("DO 4", ())
+            .await?
+            .drop_exec("DO 3", ())
+            .await?
+            .drop_exec("DO 5", ())
+            .await?
+            .drop_exec("DO 6", ())
+            .await?;
+        let (conn, row_opt) = conn
+            .first("SHOW SESSION STATUS LIKE 'Com_stmt_close';")
+            .await?;
+        let (_, count): (String, usize) = row_opt.unwrap();
+        assert_eq!(count, 3);
+        let order = conn
+            .stmt_cache_ref()
+            .iter()
+            .map(Clone::clone)
+            .collect::<Vec<String>>();
+        assert_eq!(order, &["DO 3", "DO 5", "DO 6"]);
+        conn.disconnect().await?;
+        Ok(())
     }
 
-    #[test]
-    fn should_perform_queries() {
+    #[tokio::test]
+    async fn should_perform_queries() -> super::Result<()> {
         let long_string = ::std::iter::repeat('A')
             .take(18 * 1024 * 1024)
             .collect::<String>();
-        let long_string_clone = long_string.clone();
-        let fut = Conn::new(get_opts())
-            .and_then(move |conn| {
-                Queryable::query(conn, format!(r"SELECT '{}', 231", long_string_clone))
-            })
-            .and_then(move |result| {
-                result.reduce_and_drop(vec![], move |mut acc, row| {
-                    acc.push(from_row(row));
-                    acc
-                })
-            })
-            .and_then(move |(conn, out)| Queryable::disconnect(conn).map(|_| out))
-            .map(move |result| {
-                assert_eq!((long_string, 231), result[0]);
-            });
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_drop_query() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                conn.drop_query("CREATE TEMPORARY TABLE tmp (id int DEFAULT 10, name text)")
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .query(format!(r"SELECT '{}', 231", long_string))
+            .await?;
+        let (conn, result) = result
+            .reduce_and_drop(vec![], move |mut acc, row| {
+                acc.push(from_row(row));
+                acc
             })
-            .and_then(|conn| Queryable::drop_query(conn, "INSERT INTO tmp VALUES (1, 'foo')"))
-            .and_then(|conn| Queryable::first::<_, (u8,)>(conn, "SELECT COUNT(*) FROM tmp"))
-            .and_then(|(conn, row)| conn.disconnect().map(move |_| row))
-            .map(|result| assert_eq!(result, Some((1,))));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_try_collect() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::query(
-                    conn,
-                    r"SELECT 'hello', 123
+            .await?;
+        conn.disconnect().await?;
+        assert_eq!((long_string, 231), result[0]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_drop_query() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let (conn, result) = conn
+            .drop_query("CREATE TEMPORARY TABLE tmp (id int DEFAULT 10, name text)")
+            .await?
+            .drop_query("INSERT INTO tmp VALUES (1, 'foo')")
+            .await?
+            .first::<_, (u8,)>("SELECT COUNT(*) FROM tmp")
+            .await?;
+        conn.disconnect().await?;
+        assert_eq!(result, Some((1,)));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_try_collect() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .query(
+                r"SELECT 'hello', 123
                     UNION ALL
                     SELECT 'world', 'bar'
                     UNION ALL
                     SELECT 'hello', 123
                 ",
-                )
-            })
-            .and_then(|result| result.try_collect::<(String, u8)>())
-            .and_then(|(result, mut rows)| {
-                assert!(rows.pop().unwrap().is_ok());
-                assert!(rows.pop().unwrap().is_err());
-                assert!(rows.pop().unwrap().is_ok());
-                result.drop_result()
-            })
-            .and_then(Conn::disconnect);
-
-        run(fut).unwrap()
-    }
-
-    #[test]
-    fn should_try_collect_and_drop() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::query(
-                    conn,
-                    r"SELECT 'hello', 123
+            )
+            .await?;
+        let (result, mut rows) = result.try_collect::<(String, u8)>().await?;
+        assert!(rows.pop().unwrap().is_ok());
+        assert!(rows.pop().unwrap().is_err());
+        assert!(rows.pop().unwrap().is_ok());
+        let conn = result.drop_result().await?;
+        conn.disconnect().await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_try_collect_and_drop() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let (conn, mut rows) = conn
+            .query(
+                r"SELECT 'hello', 123
                     UNION ALL
                     SELECT 'world', 'bar'
                     UNION ALL
                     SELECT 'hello', 123;
                     SELECT 'foo', 255;
                 ",
-                )
-            })
-            .and_then(|result| result.try_collect_and_drop::<(String, u8)>())
-            .and_then(|(conn, mut rows)| {
-                assert!(rows.pop().unwrap().is_ok());
-                assert!(rows.pop().unwrap().is_err());
-                assert!(rows.pop().unwrap().is_ok());
-                conn.disconnect()
-            });
-
-        run(fut).unwrap()
-    }
-
-    #[test]
-    fn should_handle_mutliresult_set() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::query(
-                    conn,
-                    r"SELECT 'hello', 123
+            )
+            .await?
+            .try_collect_and_drop::<(String, u8)>()
+            .await?;
+        assert!(rows.pop().unwrap().is_ok());
+        assert!(rows.pop().unwrap().is_err());
+        assert!(rows.pop().unwrap().is_ok());
+        conn.disconnect().await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_handle_mutliresult_set() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .query(
+                r"SELECT 'hello', 123
                     UNION ALL
                     SELECT 'world', 231;
                     SELECT 'foo', 255;
                 ",
-                )
-            })
-            .and_then(|result| result.collect::<(String, u8)>())
-            .and_then(|(result, rows_1)| (result.collect_and_drop(), Ok(rows_1)))
-            .and_then(|((conn, rows_2), rows_1)| {
-                Queryable::disconnect(conn).map(|_| vec![rows_1, rows_2])
-            })
-            .map(|rows_vec| {
-                assert_eq!(rows_vec.len(), 2);
-                for (i, rows) in rows_vec.into_iter().enumerate() {
-                    if i == 0 {
-                        assert_eq!((String::from("hello"), 123), rows[0]);
-                        assert_eq!((String::from("world"), 231), rows[1]);
-                    }
-                    if i == 1 {
-                        assert_eq!((String::from("foo"), 255), rows[0]);
-                    }
-                }
-            });
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_map_resultset() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::query(
-                    conn,
-                    r"
+            )
+            .await?;
+        let (result, rows_1) = result.collect::<(String, u8)>().await?;
+        let (conn, rows_2) = result.collect_and_drop().await?;
+        conn.disconnect().await?;
+
+        assert_eq!((String::from("hello"), 123), rows_1[0]);
+        assert_eq!((String::from("world"), 231), rows_1[1]);
+        assert_eq!((String::from("foo"), 255), rows_2[0]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_map_resultset() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .query(
+                r"
                     SELECT 'hello', 123
                     UNION ALL
                     SELECT 'world', 231;
                     SELECT 'foo', 255;
                 ",
-                )
-            })
-            .and_then(|result| result.map(|row| from_row::<(String, u8)>(row)))
-            .and_then(|(result, rows_1)| (result.map_and_drop(from_row), Ok(rows_1)))
-            .and_then(|((conn, rows_2), rows_1)| {
-                Queryable::disconnect(conn).map(|_| vec![rows_1, rows_2])
-            })
-            .map(|rows_vec| {
-                assert_eq!(rows_vec.len(), 2);
-                for (i, rows) in rows_vec.into_iter().enumerate() {
-                    if i == 0 {
-                        assert_eq!((String::from("hello"), 123), rows[0]);
-                        assert_eq!((String::from("world"), 231), rows[1]);
-                    }
-                    if i == 1 {
-                        assert_eq!((String::from("foo"), 255), rows[0]);
-                    }
-                }
-            });
+            )
+            .await?;
+
+        let (result, rows_1) = result.map(|row| from_row::<(String, u8)>(row)).await?;
+        let (conn, rows_2) = result.map_and_drop(from_row).await?;
+        conn.disconnect().await?;
 
-        run(fut).unwrap();
+        assert_eq!((String::from("hello"), 123), rows_1[0]);
+        assert_eq!((String::from("world"), 231), rows_1[1]);
+        assert_eq!((String::from("foo"), 255), rows_2[0]);
+        Ok(())
     }
 
-    #[test]
-    fn should_reduce_resultset() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::query(
-                    conn,
-                    r"SELECT 5
+    #[tokio::test]
+    async fn should_reduce_resultset() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .query(
+                r"SELECT 5
                     UNION ALL
                     SELECT 6;
                     SELECT 7;",
-                )
-            })
-            .and_then(|result| {
-                result.reduce(0, |mut acc, row| {
-                    acc += from_row::<i32>(row);
-                    acc
-                })
-            })
-            .and_then(|(result, reduced)| (result.collect_and_drop(), Ok(reduced)))
-            .and_then(|((conn, rows_2), reduced)| {
-                Queryable::disconnect(conn).map(move |_| vec![vec![reduced], rows_2])
+            )
+            .await?;
+        let (result, reduced) = result
+            .reduce(0, |mut acc, row| {
+                acc += from_row::<i32>(row);
+                acc
             })
-            .map(|rows_vec| {
-                assert_eq!(rows_vec.len(), 2);
-                for (i, rows) in rows_vec.into_iter().enumerate() {
-                    if i == 0 {
-                        assert_eq!(11, rows[0]);
-                    }
-                    if i == 1 {
-                        assert_eq!(7, rows[0]);
-                    }
-                }
-            });
-
-        run(fut).unwrap();
+            .await?;
+        let (conn, rows_2) = result.collect_and_drop::<i32>().await?;
+        conn.disconnect().await?;
+        assert_eq!(11, reduced);
+        assert_eq!(7, rows_2[0]);
+        Ok(())
     }
 
-    #[test]
-    fn should_handle_multi_result_sets_where_some_results_have_no_output() {
+    #[tokio::test]
+    async fn should_handle_multi_result_sets_where_some_results_have_no_output() -> super::Result<()>
+    {
         const QUERY: &str = r"SELECT 1;
             UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
             SELECT 2;
@@ -1004,40 +897,38 @@ mod test {
             UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
             SELECT 4;";
 
-        let fut = Conn::new(get_opts())
-            .and_then(|c| {
-                c.start_transaction(TransactionOptions::new())
-                    .and_then(|t| t.drop_query(QUERY))
-                    .and_then(|t| t.query(QUERY).and_then(|r| r.collect_and_drop::<u8>()))
-                    .and_then(|(t, out)| {
-                        assert_eq!(vec![1], out);
-                        t.query(QUERY)
-                            .and_then(|r| r.for_each_and_drop(|x| assert_eq!(from_row::<u8>(x), 1)))
-                    })
-                    .and_then(|t| {
-                        t.query(QUERY)
-                            .and_then(|r| r.map_and_drop(|row| from_row::<u8>(row)))
-                    })
-                    .and_then(|(t, out)| {
-                        assert_eq!(vec![1], out);
-                        t.query(QUERY)
-                            .and_then(|r| r.reduce_and_drop(0u8, |acc, x| acc + from_row::<u8>(x)))
-                    })
-                    .and_then(|(t, out)| {
-                        assert_eq!(1, out);
-                        t.query(QUERY).and_then(|r| r.drop_result())
-                    })
-                    .and_then(|t| t.commit())
-            })
-            .and_then(|c| c.first_exec::<_, _, u8>("SELECT 1", ()))
-            .and_then(|(c, output)| c.disconnect().map(move |_| output))
-            .map(|result| assert_eq!(result, Some(1)));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_iterate_over_resultset() {
+        let c = Conn::new(get_opts()).await?;
+        let c = c
+            .drop_query("CREATE TEMPORARY TABLE time_zone (Time_zone_id INT)")
+            .await
+            .unwrap();
+        let t = c.start_transaction(TransactionOptions::new()).await?;
+        let t = t.drop_query(QUERY).await?;
+        let r = t.query(QUERY).await?;
+        let (t, out) = r.collect_and_drop::<u8>().await?;
+        assert_eq!(vec![1], out);
+        let r = t.query(QUERY).await?;
+        let t = r
+            .for_each_and_drop(|x| assert_eq!(from_row::<u8>(x), 1))
+            .await?;
+        let r = t.query(QUERY).await?;
+        let (t, out) = r.map_and_drop(|row| from_row::<u8>(row)).await?;
+        assert_eq!(vec![1], out);
+        let r = t.query(QUERY).await?;
+        let (t, out) = r
+            .reduce_and_drop(0u8, |acc, x| acc + from_row::<u8>(x))
+            .await?;
+        assert_eq!(1, out);
+        let t = t.query(QUERY).await?.drop_result().await?;
+        let c = t.commit().await?;
+        let (c, result) = c.first_exec::<_, _, u8>("SELECT 1", ()).await?;
+        c.disconnect().await?;
+        assert_eq!(result, Some(1));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_iterate_over_resultset() -> super::Result<()> {
         use std::sync::{
             atomic::{AtomicUsize, Ordering},
             Arc,
@@ -1045,212 +936,174 @@ mod test {
 
         let acc = Arc::new(AtomicUsize::new(0));
 
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::query(
-                    conn,
-                    r"SELECT 2
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .query(
+                r"SELECT 2
                     UNION ALL
                     SELECT 3;
                     SELECT 5;",
-                )
-            })
-            .and_then({
+            )
+            .await?;
+        let result = result
+            .for_each({
                 let acc = acc.clone();
-                move |result| {
-                    result.for_each({
-                        let acc = acc.clone();
-                        move |row| {
-                            acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
-                        }
-                    })
+                move |row| {
+                    acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
                 }
             })
-            .and_then({
+            .await?;
+        let conn = result
+            .for_each_and_drop({
                 let acc = acc.clone();
-                move |result| {
-                    result.for_each_and_drop({
-                        let acc = acc.clone();
-                        move |row| {
-                            acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
-                        }
-                    })
+                move |row| {
+                    acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
                 }
             })
-            .and_then(Queryable::disconnect)
-            .map(move |_| assert_eq!(acc.load(Ordering::SeqCst), 10));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_prepare_statement() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| Queryable::prepare(conn, r"SELECT ?"))
-            .and_then(|stmt| stmt.close())
-            .and_then(|conn| conn.disconnect());
-
-        run(fut).unwrap();
-
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| Queryable::prepare(conn, r"SELECT :foo"))
-            .and_then(|stmt| stmt.close())
-            .and_then(|conn| conn.disconnect());
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_execute_statement() {
+            .await?;
+        conn.disconnect().await?;
+        assert_eq!(acc.load(Ordering::SeqCst), 10);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_prepare_statement() -> super::Result<()> {
+        Conn::new(get_opts())
+            .await?
+            .prepare(r"SELECT ?")
+            .await?
+            .close()
+            .await?
+            .disconnect()
+            .await?;
+
+        Conn::new(get_opts())
+            .await?
+            .prepare(r"SELECT :foo")
+            .await?
+            .close()
+            .await?
+            .disconnect()
+            .await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_execute_statement() -> super::Result<()> {
         let long_string = ::std::iter::repeat('A')
             .take(18 * 1024 * 1024)
             .collect::<String>();
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| Queryable::prepare(conn, r"SELECT ?"))
-            .and_then({
-                let long_string = long_string.clone();
-                move |stmt| stmt.execute((long_string,))
-            })
-            .and_then(|result| result.map_and_drop(|row| from_row::<(String,)>(row)))
-            .and_then(|(stmt, mut mapped)| {
-                assert_eq!(mapped.len(), 1);
-                assert_eq!(mapped.pop(), Some((long_string,)));
-                stmt.execute((42,))
-            })
-            .and_then(|result| result.collect_and_drop::<(u8,)>())
-            .and_then(|(stmt, collected)| {
-                assert_eq!(collected, vec![(42u8,)]);
-                stmt.execute((8,))
-            })
-            .and_then(|result| {
-                result.reduce_and_drop(2, |mut acc, row| {
-                    acc += from_row::<i32>(row);
-                    acc
-                })
-            })
-            .and_then(|(stmt, reduced)| {
-                stmt.close()
-                    .and_then(|conn| conn.disconnect())
-                    .map(move |_| reduced)
-            })
-            .map(|output| assert_eq!(output, 10));
-
-        run(fut).unwrap();
-
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| Queryable::prepare(conn, r"SELECT :foo, :bar, :foo, 3"))
-            .and_then(|stmt| stmt.execute(params! { "foo" => "quux", "bar" => "baz" }))
-            .and_then(|result| {
-                result.map_and_drop(|row| from_row::<(String, String, String, u8)>(row))
-            })
-            .and_then(|(stmt, mut mapped)| {
-                assert_eq!(mapped.len(), 1);
-                assert_eq!(
-                    mapped.pop(),
-                    Some(("quux".into(), "baz".into(), "quux".into(), 3))
-                );
-                stmt.execute(params! { "foo" => 2, "bar" => 3 })
-            })
-            .and_then(|result| result.collect_and_drop::<(u8, u8, u8, u8)>())
-            .and_then(|(stmt, collected)| {
-                assert_eq!(collected, vec![(2, 3, 2, 3)]);
-                stmt.execute(params! { "foo" => 2, "bar" => 3 })
-            })
-            .and_then(|result| {
-                result.reduce_and_drop(0, |acc, row| {
-                    let (a, b, c, d): (u8, u8, u8, u8) = from_row(row);
-                    acc + a + b + c + d
-                })
-            })
-            .and_then(|(stmt, reduced)| {
-                stmt.close()
-                    .and_then(|conn| conn.disconnect())
-                    .map(move |_| reduced)
+        let conn = Conn::new(get_opts()).await?;
+        let stmt = conn.prepare(r"SELECT ?").await?;
+        let result = stmt.execute((&long_string,)).await?;
+        let (stmt, mut mapped) = result
+            .map_and_drop(|row| from_row::<(String,)>(row))
+            .await?;
+        assert_eq!(mapped.len(), 1);
+        assert_eq!(mapped.pop(), Some((long_string,)));
+        let result = stmt.execute((42,)).await?;
+        let (stmt, collected) = result.collect_and_drop::<(u8,)>().await?;
+        assert_eq!(collected, vec![(42u8,)]);
+        let result = stmt.execute((8,)).await?;
+        let (stmt, reduced) = result
+            .reduce_and_drop(2, |mut acc, row| {
+                acc += from_row::<i32>(row);
+                acc
             })
-            .map(|output| assert_eq!(output, 10));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_prep_exec_statement() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::prep_exec(conn, r"SELECT :a, :b, :a", params! { "a" => 2, "b" => 3 })
-            })
-            .and_then(|result| {
-                result.map_and_drop(|row| {
-                    let (a, b, c): (u8, u8, u8) = from_row(row);
-                    a * b * c
-                })
-            })
-            .and_then(|(conn, output)| Queryable::disconnect(conn).map(move |_| output[0]))
-            .map(|output| assert_eq!(output, 12u8));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_first_exec_statement() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::first_exec(
-                    conn,
-                    r"SELECT :a UNION ALL SELECT :b",
-                    params! { "a" => 2, "b" => 3 },
-                )
-            })
-            .and_then(|(conn, row_opt): (_, Option<(u8,)>)| {
-                Queryable::disconnect(conn).map(move |_| row_opt.unwrap())
-            })
-            .map(|output| assert_eq!(output, (2u8,)));
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_run_transactions() {
-        let fut = Conn::new(get_opts())
-            .and_then(|conn| {
-                Queryable::drop_query(conn, "CREATE TEMPORARY TABLE tmp (id INT, name TEXT)")
-            })
-            .and_then(|conn| Queryable::start_transaction(conn, Default::default()))
-            .and_then(|transaction| {
-                Queryable::drop_query(transaction, "INSERT INTO tmp VALUES (1, 'foo'), (2, 'bar')")
-            })
-            .and_then(|transaction| transaction.commit())
-            .and_then(|conn| Queryable::first(conn, "SELECT COUNT(*) FROM tmp"))
-            .map(|(conn, output_opt)| {
-                assert_eq!(output_opt, Some((2u8,)));
-                conn
-            })
-            .and_then(|conn| Queryable::start_transaction(conn, Default::default()))
-            .and_then(|transaction| {
-                Queryable::drop_query(
-                    transaction,
-                    "INSERT INTO tmp VALUES (3, 'baz'), (4, 'quux')",
-                )
-            })
-            .and_then(|transaction| {
-                Queryable::first_exec(transaction, "SELECT COUNT(*) FROM tmp", ())
-            })
-            .map(|(transaction, output_opt)| {
-                assert_eq!(output_opt, Some((4u8,)));
-                transaction
+            .await?;
+        stmt.close().await?.disconnect().await?;
+        assert_eq!(reduced, 10);
+
+        let conn = Conn::new(get_opts()).await?;
+        let stmt = conn.prepare(r"SELECT :foo, :bar, :foo, 3").await?;
+        let result = stmt
+            .execute(params! { "foo" => "quux", "bar" => "baz" })
+            .await?;
+        let (stmt, mut mapped) = result
+            .map_and_drop(|row| from_row::<(String, String, String, u8)>(row))
+            .await?;
+        assert_eq!(mapped.len(), 1);
+        assert_eq!(
+            mapped.pop(),
+            Some(("quux".into(), "baz".into(), "quux".into(), 3))
+        );
+        let result = stmt.execute(params! { "foo" => 2, "bar" => 3 }).await?;
+        let (stmt, collected) = result.collect_and_drop::<(u8, u8, u8, u8)>().await?;
+        assert_eq!(collected, vec![(2, 3, 2, 3)]);
+        let result = stmt.execute(params! { "foo" => 2, "bar" => 3 }).await?;
+        let (stmt, reduced) = result
+            .reduce_and_drop(0, |acc, row| {
+                let (a, b, c, d): (u8, u8, u8, u8) = from_row(row);
+                acc + a + b + c + d
             })
-            .and_then(|transaction| transaction.rollback())
-            .and_then(|conn| Queryable::first(conn, "SELECT COUNT(*) FROM tmp"))
-            .map(|(conn, output_opt)| {
-                assert_eq!(output_opt, Some((2u8,)));
-                conn
+            .await?;
+        stmt.close().await?.disconnect().await?;
+        assert_eq!(reduced, 10);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_prep_exec_statement() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let result = conn
+            .prep_exec(r"SELECT :a, :b, :a", params! { "a" => 2, "b" => 3 })
+            .await?;
+        let (conn, output) = result
+            .map_and_drop(|row| {
+                let (a, b, c): (u8, u8, u8) = from_row(row);
+                a * b * c
             })
-            .and_then(Queryable::disconnect);
-
-        run(fut).unwrap();
-    }
-
-    #[test]
-    fn should_handle_local_infile() {
+            .await?;
+        conn.disconnect().await?;
+        assert_eq!(output[0], 12u8);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_first_exec_statement() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let (conn, output): (_, Option<(u8,)>) = conn
+            .first_exec(
+                r"SELECT :a UNION ALL SELECT :b",
+                params! { "a" => 2, "b" => 3 },
+            )
+            .await?;
+        conn.disconnect().await?;
+        assert_eq!(output.unwrap(), (2u8,));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_run_transactions() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let conn = conn
+            .drop_query("CREATE TEMPORARY TABLE tmp (id INT, name TEXT)")
+            .await?;
+        let transaction = conn.start_transaction(Default::default()).await?;
+        let conn = transaction
+            .drop_query("INSERT INTO tmp VALUES (1, 'foo'), (2, 'bar')")
+            .await?
+            .commit()
+            .await?;
+        let (conn, output_opt) = conn.first("SELECT COUNT(*) FROM tmp").await?;
+        assert_eq!(output_opt, Some((2u8,)));
+        let transaction = conn.start_transaction(Default::default()).await?;
+        let transaction = transaction
+            .drop_query("INSERT INTO tmp VALUES (3, 'baz'), (4, 'quux')")
+            .await?;
+        let (t, output_opt) = transaction
+            .first_exec("SELECT COUNT(*) FROM tmp", ())
+            .await?;
+        assert_eq!(output_opt, Some((4u8,)));
+        let conn = t.rollback().await?;
+        let (conn, output_opt) = conn.first("SELECT COUNT(*) FROM tmp").await?;
+        assert_eq!(output_opt, Some((2u8,)));
+        conn.disconnect().await?;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn should_handle_local_infile() -> super::Result<()> {
         use std::io::Write;
 
         let mut opts = OptsBuilder::from_opts(get_opts());
@@ -1258,55 +1111,56 @@ mod test {
             &["local_infile.txt"][..],
         )));
 
-        let fut = Conn::new(opts)
-            .and_then(|conn| Queryable::drop_query(conn, "CREATE TEMPORARY TABLE tmp (a TEXT);"))
-            .and_then(|conn| {
-                let mut file = ::std::fs::File::create("local_infile.txt").unwrap();
-                let _ = file.write(b"AAAAAA\n");
-                let _ = file.write(b"BBBBBB\n");
-                let _ = file.write(b"CCCCCC\n");
-                Queryable::drop_query(
-                    conn,
-                    "LOAD DATA LOCAL INFILE 'local_infile.txt' INTO TABLE tmp;",
-                )
-            })
-            .and_then(|conn| Queryable::prep_exec(conn, "SELECT * FROM tmp;", ()))
-            .and_then(|result| result.map_and_drop(|row| from_row::<(String,)>(row).0))
-            .and_then(|(conn, result)| {
-                assert_eq!(result.len(), 3);
-                assert_eq!(result[0], "AAAAAA");
-                assert_eq!(result[1], "BBBBBB");
-                assert_eq!(result[2], "CCCCCC");
-                Queryable::disconnect(conn)
-            })
-            .then(|x| {
-                let _ = ::std::fs::remove_file("local_infile.txt");
-                x
-            })
-            .then(|result| match result {
-                Err(err) => match err {
-                    crate::error::Error::Server(ref err) if err.code == 1148 => {
-                        // The used command is not allowed with this MySQL version
-                        Ok(())
-                    }
-                    _ => Err(err),
-                },
-                _ => Ok(()),
-            });
-
-        run(fut).unwrap();
+        let conn = Conn::new(opts).await?;
+        let conn = conn
+            .drop_query("CREATE TEMPORARY TABLE tmp (a TEXT);")
+            .await?;
+
+        let mut file = ::std::fs::File::create("local_infile.txt").unwrap();
+        let _ = file.write(b"AAAAAA\n");
+        let _ = file.write(b"BBBBBB\n");
+        let _ = file.write(b"CCCCCC\n");
+        let conn = match conn
+            .drop_query("LOAD DATA LOCAL INFILE 'local_infile.txt' INTO TABLE tmp;")
+            .await
+        {
+            Ok(conn) => conn,
+            Err(super::Error::Server(ref err)) if err.code == 1148 => {
+                // The used command is not allowed with this MySQL version
+                return Ok(());
+            }
+            e @ Err(_) => e.unwrap(),
+        };
+        let (conn, result) = conn
+            .prep_exec("SELECT * FROM tmp;", ())
+            .await?
+            .map_and_drop(|row| from_row::<(String,)>(row).0)
+            .await?;
+        assert_eq!(result.len(), 3);
+        assert_eq!(result[0], "AAAAAA");
+        assert_eq!(result[1], "BBBBBB");
+        assert_eq!(result[2], "CCCCCC");
+        let result = conn.disconnect().await;
+        let _ = ::std::fs::remove_file("local_infile.txt");
+        if let Err(crate::error::Error::Server(ref err)) = result {
+            if err.code == 1148 {
+                // The used command is not allowed with this MySQL version
+                return Ok(());
+            }
+        }
+        result.unwrap();
+        Ok(())
     }
 
     #[cfg(feature = "nightly")]
     mod bench {
-        use futures::Future;
+        use futures_util::try_future::TryFutureExt;
 
-        use super::get_opts;
-        use crate::{conn::Conn, queryable::Queryable};
+        use crate::{conn::Conn, queryable::Queryable, test_misc::get_opts};
 
         #[bench]
         fn simple_exec(bencher: &mut test::Bencher) {
-            let mut runtime = tokio::runtime::Runtime::new().unwrap();
+            let runtime = tokio::runtime::Runtime::new().unwrap();
             let mut conn_opt = Some(runtime.block_on(Conn::new(get_opts())).unwrap());
 
             bencher.iter(|| {
@@ -1317,12 +1171,12 @@ mod test {
             runtime
                 .block_on(conn_opt.take().unwrap().disconnect())
                 .unwrap();
-            runtime.shutdown_on_idle().wait().unwrap();
+            runtime.shutdown_on_idle();
         }
 
         #[bench]
         fn select_large_string(bencher: &mut test::Bencher) {
-            let mut runtime = tokio::runtime::Runtime::new().unwrap();
+            let runtime = tokio::runtime::Runtime::new().unwrap();
             let mut conn_opt = Some(runtime.block_on(Conn::new(get_opts())).unwrap());
 
             bencher.iter(|| {
@@ -1337,12 +1191,12 @@ mod test {
             runtime
                 .block_on(conn_opt.take().unwrap().disconnect())
                 .unwrap();
-            runtime.shutdown_on_idle().wait().unwrap();
+            runtime.shutdown_on_idle();
         }
 
         #[bench]
         fn prepared_exec(bencher: &mut test::Bencher) {
-            let mut runtime = tokio::runtime::Runtime::new().unwrap();
+            let runtime = tokio::runtime::Runtime::new().unwrap();
             let mut stmt_opt = Some(
                 runtime
                     .block_on(Conn::new(get_opts()).and_then(|conn| conn.prepare("DO 1")))
@@ -1367,12 +1221,12 @@ mod test {
                         .and_then(|conn| conn.disconnect()),
                 )
                 .unwrap();
-            runtime.shutdown_on_idle().wait().unwrap();
+            runtime.shutdown_on_idle();
         }
 
         #[bench]
         fn prepare_and_exec(bencher: &mut test::Bencher) {
-            let mut runtime = tokio::runtime::Runtime::new().unwrap();
+            let runtime = tokio::runtime::Runtime::new().unwrap();
             let mut conn_opt = Some(runtime.block_on(Conn::new(get_opts())).unwrap());
 
             bencher.iter(|| {
@@ -1392,7 +1246,7 @@ mod test {
             runtime
                 .block_on(conn_opt.take().unwrap().disconnect())
                 .unwrap();
-            runtime.shutdown_on_idle().wait().unwrap();
+            runtime.shutdown_on_idle();
         }
     }
 }
diff --git a/src/conn/pool/futures/disconnect_pool.rs b/src/conn/pool/futures/disconnect_pool.rs
index 0be7a211..d5198003 100644
--- a/src/conn/pool/futures/disconnect_pool.rs
+++ b/src/conn/pool/futures/disconnect_pool.rs
@@ -6,7 +6,9 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{task, Async, Future, Poll};
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use crate::{
     conn::pool::{Inner, Pool},
@@ -30,16 +32,15 @@ pub fn new(pool: Pool) -> DisconnectPool {
 }
 
 impl Future for DisconnectPool {
-    type Item = ();
-    type Error = Error;
+    type Output = Result<(), Error>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        self.pool_inner.wake.push(task::current());
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.pool_inner.wake.push(cx.waker().clone());
 
         if self.pool_inner.closed.load(atomic::Ordering::Acquire) {
-            Ok(Async::Ready(()))
+            Poll::Ready(Ok(()))
         } else {
-            Ok(Async::NotReady)
+            Poll::Pending
         }
     }
 }
diff --git a/src/conn/pool/futures/get_conn.rs b/src/conn/pool/futures/get_conn.rs
index ed9610ab..ffdfa9e3 100644
--- a/src/conn/pool/futures/get_conn.rs
+++ b/src/conn/pool/futures/get_conn.rs
@@ -6,19 +6,22 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{try_ready, Async, Future, Poll};
+use futures_core::ready;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use crate::{
     conn::{pool::Pool, Conn},
     error::*,
-    MyFuture,
+    BoxFuture,
 };
 
 pub(crate) enum GetConnInner {
     New,
     Done(Option<Conn>),
     // TODO: one day this should be an existential
-    Connecting(Box<dyn MyFuture<Conn>>),
+    Connecting(BoxFuture<Conn>),
 }
 
 impl GetConnInner {
@@ -41,18 +44,20 @@ pub fn new(pool: &Pool) -> GetConn {
     }
 }
 
+// this manual implementation of Future may seem stupid, but we sort
+// of need it to get the dropping behavior we want.
 impl Future for GetConn {
-    type Item = Conn;
-    type Error = Error;
+    type Output = Result<Conn>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         loop {
             match self.inner {
-                GetConnInner::New => match try_ready!(self
-                    .pool
-                    .as_mut()
-                    .expect("GetConn::poll polled after returning Async::Ready")
-                    .poll_new_conn())
+                GetConnInner::New => match ready!(Pin::new(
+                    self.pool
+                        .as_mut()
+                        .expect("GetConn::poll polled after returning Async::Ready")
+                )
+                .poll_new_conn(cx))?
                 .inner
                 .take()
                 {
@@ -76,18 +81,13 @@ impl Future for GetConn {
                             .take()
                             .expect("GetConn::poll polled after returning Async::Ready"),
                     );
-                    return Ok(Async::Ready(c));
+                    return Poll::Ready(Ok(c));
                 }
                 GetConnInner::Done(None) => {
                     unreachable!("GetConn::poll polled after returning Async::Ready");
                 }
                 GetConnInner::Connecting(ref mut f) => {
-                    let result = match f.poll() {
-                        Ok(Async::NotReady) => return Ok(Async::NotReady),
-                        Ok(Async::Ready(c)) => Ok(c),
-                        Err(e) => Err(e),
-                    };
-
+                    let result = ready!(Pin::new(f).poll(cx));
                     let pool = self
                         .pool
                         .take()
@@ -96,11 +96,11 @@ impl Future for GetConn {
                     return match result {
                         Ok(mut c) => {
                             c.inner.pool = Some(pool);
-                            Ok(Async::Ready(c))
+                            Poll::Ready(Ok(c))
                         }
                         Err(e) => {
                             pool.cancel_connection();
-                            Err(e)
+                            Poll::Ready(Err(e))
                         }
                     };
                 }
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 4cfccce1..6e9098b2 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -6,13 +6,13 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use ::futures::stream::futures_unordered::FuturesUnordered;
-use ::futures::{
-    task::{self, Task},
-    try_ready, Async, Future, Poll, Stream,
-};
+use futures_core::{ready, stream::Stream};
+use futures_util::stream::futures_unordered::FuturesUnordered;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll, Waker};
 
-use tokio_sync::mpsc;
+use tokio::sync::mpsc;
 
 use std::{
     fmt,
@@ -28,7 +28,7 @@ use crate::{
         transaction::{Transaction, TransactionOptions},
         Queryable,
     },
-    BoxFuture, MyFuture,
+    BoxFuture,
 };
 
 // this is a really unfortunate name for a module
@@ -47,10 +47,9 @@ struct Recycler {
 }
 
 impl Future for Recycler {
-    type Item = ();
-    type Error = ();
+    type Output = ();
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let mut readied = 0;
         let mut close = self.inner.close.load(atomic::Ordering::Acquire);
 
@@ -58,11 +57,11 @@ impl Future for Recycler {
             ($self:ident, $readied:ident, $conn:ident) => {
                 if $conn.inner.stream.is_none() || $conn.inner.disconnected {
                     // drop unestablished connection
-                    $self.discard.push(Box::new(::futures::future::ok(())));
+                    $self.discard.push(Box::pin(::futures_util::future::ok(())));
                 } else if $conn.inner.in_transaction || $conn.inner.has_result.is_some() {
-                    $self.cleaning.push($conn.cleanup());
+                    $self.cleaning.push(Box::pin($conn.cleanup()));
                 } else if $conn.expired() || $self.inner.idle.len() >= $self.min || close {
-                    $self.discard.push(Box::new($conn.close()));
+                    $self.discard.push(Box::pin($conn.close()));
                 } else {
                     $self
                         .inner
@@ -76,23 +75,23 @@ impl Future for Recycler {
 
         while !self.eof {
             // see if there are more connections for us to recycle
-            match self.dropped.poll().unwrap() {
-                Async::Ready(Some(Some(conn))) => {
+            match Pin::new(&mut self.dropped).poll_next(cx) {
+                Poll::Ready(Some(Some(conn))) => {
                     conn_decision!(self, readied, conn);
                 }
-                Async::Ready(Some(None)) => {
+                Poll::Ready(Some(None)) => {
                     // someone signaled us that it's exit time
                     close = self.inner.close.load(atomic::Ordering::Acquire);
                     assert!(close);
                     continue;
                 }
-                Async::Ready(None) => {
+                Poll::Ready(None) => {
                     // no more connections are coming -- time to exit!
                     self.inner.close.store(true, atomic::Ordering::Release);
                     self.eof = true;
                     close = true;
                 }
-                Async::NotReady => {
+                Poll::Pending => {
                     // nope -- but let's still make progress on the ones we have
                     break;
                 }
@@ -108,10 +107,10 @@ impl Future for Recycler {
 
         // are any dirty connections ready for us to reclaim?
         loop {
-            match self.cleaning.poll() {
-                Ok(Async::NotReady) | Ok(Async::Ready(None)) => break,
-                Ok(Async::Ready(Some(conn))) => conn_decision!(self, readied, conn),
-                Err(e) => {
+            match Pin::new(&mut self.cleaning).poll_next(cx) {
+                Poll::Pending | Poll::Ready(None) => break,
+                Poll::Ready(Some(Ok(conn))) => conn_decision!(self, readied, conn),
+                Poll::Ready(Some(Err(e))) => {
                     // an error occurred while cleaning a connection.
                     // what do we do? replace it with a new connection?
                     self.discarded += 1;
@@ -123,13 +122,13 @@ impl Future for Recycler {
 
         // are there any torn-down connections for us to deal with?
         loop {
-            match self.discard.poll() {
-                Ok(Async::NotReady) | Ok(Async::Ready(None)) => break,
-                Ok(Async::Ready(Some(()))) => {
+            match Pin::new(&mut self.discard).poll_next(cx) {
+                Poll::Pending | Poll::Ready(None) => break,
+                Poll::Ready(Some(Ok(()))) => {
                     // yes! count it.
                     self.discarded += 1
                 }
-                Err(e) => {
+                Poll::Ready(Some(Err(e))) => {
                     // an error occurred while closing a connection.
                     // what do we do? we still replace it with a new connection..
                     self.discarded += 1;
@@ -177,9 +176,9 @@ impl Future for Recycler {
         if self.inner.closed.load(atomic::Ordering::Acquire) {
             // since there are no more Pools, we also know that no-one is waiting anymore,
             // so we don't have to worry about calling wake more times
-            Ok(Async::Ready(()))
+            Poll::Ready(())
         } else {
-            Ok(Async::NotReady)
+            Poll::Pending
         }
     }
 }
@@ -188,7 +187,7 @@ struct Inner {
     close: atomic::AtomicBool,
     closed: atomic::AtomicBool,
     idle: crossbeam::queue::ArrayQueue<Conn>,
-    wake: crossbeam::queue::SegQueue<Task>,
+    wake: crossbeam::queue::SegQueue<Waker>,
     exist: atomic::AtomicUsize,
     extra_wakeups: atomic::AtomicUsize,
 
@@ -203,7 +202,7 @@ impl Inner {
         }
 
         while let Ok(task) = self.wake.pop() {
-            task.notify();
+            task.wake();
             readied -= 1;
             if readied == 0 {
                 if self.close.load(atomic::Ordering::Acquire) {
@@ -283,12 +282,11 @@ impl Pool {
     }
 
     /// Shortcut for `get_conn` followed by `start_transaction`.
-    pub fn start_transaction(
+    pub async fn start_transaction(
         &self,
         options: TransactionOptions,
-    ) -> impl MyFuture<Transaction<Conn>> {
-        self.get_conn()
-            .and_then(|conn| Queryable::start_transaction(conn, options))
+    ) -> Result<Transaction<Conn>> {
+        Queryable::start_transaction(self.get_conn().await?, options).await
     }
 
     /// Returns future that disconnects this pool from server and resolves to `()`.
@@ -344,13 +342,17 @@ impl Pool {
     }
 
     /// Poll the pool for an available connection.
-    fn poll_new_conn(&mut self) -> Result<Async<GetConn>> {
-        self.poll_new_conn_inner(false)
+    fn poll_new_conn(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<GetConn>> {
+        self.poll_new_conn_inner(false, cx)
     }
 
-    fn poll_new_conn_inner(&mut self, retrying: bool) -> Result<Async<GetConn>> {
+    fn poll_new_conn_inner(
+        mut self: Pin<&mut Self>,
+        retrying: bool,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<GetConn>> {
         if self.inner.close.load(atomic::Ordering::Acquire) {
-            return Err(Error::Driver(DriverError::PoolDisconnected));
+            return Err(Error::Driver(DriverError::PoolDisconnected)).into();
         }
 
         loop {
@@ -362,7 +364,7 @@ impl Pool {
                         continue;
                     }
 
-                    return Ok(Async::Ready(GetConn {
+                    return Poll::Ready(Ok(GetConn {
                         pool: Some(self.clone()),
                         inner: GetConnInner::Done(Some(conn)),
                     }));
@@ -412,25 +414,25 @@ impl Pool {
                     self.drop
                         .try_send(None)
                         .expect("recycler is active as long as any Pool is");
-                    return Err(Error::Driver(DriverError::PoolDisconnected));
+                    return Err(Error::Driver(DriverError::PoolDisconnected)).into();
                 }
 
-                return Ok(Async::Ready(GetConn {
+                return Poll::Ready(Ok(GetConn {
                     pool: Some(self.clone()),
-                    inner: GetConnInner::Connecting(Box::new(Conn::new(self.opts.clone()))),
+                    inner: GetConnInner::Connecting(Box::pin(Conn::new(self.opts.clone()))),
                 }));
             }
 
             let exist = self.inner.exist.fetch_sub(1, atomic::Ordering::AcqRel);
             if exist < self.pool_constraints.max() {
                 // we'd _now_ be allowed to make a connection
-                return self.poll_new_conn_inner(retrying);
+                return self.poll_new_conn_inner(retrying, cx);
             }
         }
 
         if !retrying {
             // no go -- we have to wait
-            self.inner.wake.push(task::current());
+            self.inner.wake.push(cx.waker().clone());
 
             // there's a potential race here -- imagine another task releases a connection after we
             // try to poll .idle or check .exist, but before we push our task onto .wake. In that
@@ -440,7 +442,7 @@ impl Pool {
             // an alternative strategy would be to _always_ push to .wake and then do the checks,
             // but that would lead to a large number of spurious notifications/wakeups, as well as
             // needless contention on .wake.
-            let conn = try_ready!(self.poll_new_conn_inner(true));
+            let conn = ready!(self.as_mut().poll_new_conn_inner(true, cx))?;
 
             // this is a tricky case. we already registered ourselves as wanting to be woken up,
             // but we now have a connection, so we won't be waiting. this means that _if_ we were
@@ -456,9 +458,9 @@ impl Pool {
             // and also waking someone up (perhaps spuriously) in case we have already been
             // notified.
             if let Ok(task) = self.inner.wake.pop() {
-                if task.will_notify_current() {
+                if task.will_wake(cx.waker()) {
                     // phew -- we got out of that one easy!
-                    return Ok(Async::Ready(conn));
+                    return Poll::Ready(Ok(conn));
                 }
 
                 // if we _haven't_ been notified yet, someone else may be deciding who to wake up
@@ -486,16 +488,16 @@ impl Pool {
                 //
                 // in this case, task1 might never be awoken again, which is not okay.
                 // hence:
-                task.notify();
+                task.wake();
             } else {
                 // someone tried to notify us, but also, no-one else is waiting,
                 // so there's no-one to "forward" that wake-up to.
             }
 
-            return Ok(Async::Ready(conn));
+            return Poll::Ready(Ok(conn));
         }
 
-        Ok(Async::NotReady)
+        Poll::Pending
     }
 }
 
@@ -511,217 +513,182 @@ impl Drop for Conn {
 
 #[cfg(test)]
 mod test {
-    use futures::{collect, future, Future};
+    use futures_util::stream::StreamExt;
     use std::sync::atomic;
 
     use crate::{
-        conn::pool::Pool, queryable::Queryable, test_misc::DATABASE_URL, TransactionOptions,
+        conn::pool::Pool, queryable::Queryable, test_misc::get_opts, PoolConstraints,
+        TransactionOptions,
     };
 
-    /// Same as `tokio::run`, but will panic if future panics and will return the result
-    /// of future execution.
-    fn run<F, T, U>(future: F) -> Result<T, U>
-    where
-        F: Future<Item = T, Error = U> + Send + 'static,
-        T: Send + 'static,
-        U: Send + 'static,
-    {
-        let mut runtime = tokio::runtime::Runtime::new().unwrap();
-        let result = runtime.block_on(future);
-        runtime.shutdown_on_idle().wait().unwrap();
-        result
-    }
-
-    #[test]
-    fn should_connect() {
-        let pool = Pool::new(&**DATABASE_URL);
-        let fut = pool
-            .get_conn()
-            .and_then(|conn| conn.ping().map(|_| ()))
-            .and_then(|_| pool.disconnect());
-
-        run(fut).unwrap();
+    #[tokio::test]
+    async fn should_connect() -> super::Result<()> {
+        let pool = Pool::new(get_opts());
+        pool.get_conn().await?.ping().await?;
+        pool.disconnect().await?;
+        Ok(())
     }
 
-    #[test]
+    #[tokio::test]
     #[ignore]
-    fn can_handle_the_pressure() {
-        let mut runtime = tokio::runtime::Runtime::new().unwrap();
-        let pool = Pool::new(&**DATABASE_URL);
-        for _ in 0..10 {
-            use futures::{Sink, Stream};
-            let (tx, rx) = futures::sync::mpsc::unbounded();
+    async fn can_handle_the_pressure() {
+        let pool = Pool::new(get_opts());
+        for _ in 0..10i32 {
+            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
             for i in 0..10_000 {
                 let pool = pool.clone();
-                let tx = tx.clone();
-                runtime.spawn(futures::future::lazy(move || {
-                    pool.get_conn()
-                        .map_err(|e| unreachable!("{:?}", e))
-                        .and_then(move |_| tx.send(i).map_err(|e| unreachable!("{:?}", e)))
-                        .map(|_| ())
-                }));
+                let mut tx = tx.clone();
+                tokio::spawn(async move {
+                    let _ = pool.get_conn().await.unwrap();
+                    tx.try_send(i).unwrap();
+                });
             }
             drop(tx);
-            runtime.block_on(rx.fold(0, |_, _i| Ok(0))).unwrap();
+            // see that all the tx's eventually complete
+            while let Some(_) = rx.next().await {}
         }
         drop(pool);
-        runtime.shutdown_on_idle().wait().unwrap();
     }
 
-    #[test]
-    fn should_start_transaction() {
-        let pool = Pool::new(format!("{}?pool_min=1&pool_max=1", &**DATABASE_URL));
-        let fut = pool
+    #[tokio::test]
+    async fn should_start_transaction() -> super::Result<()> {
+        let mut opts = get_opts();
+        opts.pool_constraints(PoolConstraints::new(1, 1));
+        let pool = Pool::new(opts);
+        pool.get_conn()
+            .await?
+            .drop_query("CREATE TABLE IF NOT EXISTS tmp(id int)")
+            .await?;
+        let _ = pool
+            .start_transaction(TransactionOptions::default())
+            .await?
+            .batch_exec("INSERT INTO tmp (id) VALUES (?)", vec![(1,), (2,)])
+            .await?
+            .prep_exec("SELECT * FROM tmp", ())
+            .await?;
+        let row_opt = pool
             .get_conn()
-            .and_then(|conn| conn.drop_query("CREATE TABLE IF NOT EXISTS tmp(id int)"))
-            .and_then({
-                let pool = pool.clone();
-                move |_| pool.start_transaction(TransactionOptions::default())
-            })
-            .and_then(|transaction| {
-                transaction.batch_exec("INSERT INTO tmp (id) VALUES (?)", vec![(1,), (2,)])
-            })
-            .and_then(|transaction| transaction.prep_exec("SELECT * FROM tmp", ()))
-            .map(|_| ())
-            .and_then({
-                let pool = pool.clone();
-                move |_| pool.get_conn()
-            })
-            .and_then(|conn| conn.first("SELECT COUNT(*) FROM tmp"))
-            .and_then(|(_, row_opt)| {
-                assert_eq!(row_opt, Some((0u8,)));
-                pool.get_conn()
-                    .and_then(|conn| conn.drop_query("DROP TABLE tmp"))
-                    .and_then(move |_| pool.disconnect())
-            });
-
-        run(fut).unwrap();
+            .await?
+            .first("SELECT COUNT(*) FROM tmp")
+            .await?
+            .1;
+        assert_eq!(row_opt, Some((0u8,)));
+        pool.get_conn().await?.drop_query("DROP TABLE tmp").await?;
+        pool.disconnect().await?;
+        Ok(())
     }
 
-    #[test]
-    fn should_hold_bounds2() {
+    #[tokio::test]
+    async fn should_hold_bounds2() -> super::Result<()> {
         use std::cmp::min;
 
         const POOL_MIN: usize = 5;
         const POOL_MAX: usize = 10;
 
-        let url = format!(
-            "{}?pool_min={}&pool_max={}",
-            &**DATABASE_URL, POOL_MIN, POOL_MAX
-        );
-
         // Clean
-        let pool = Pool::new(url.clone());
+        let mut opts = get_opts();
+        opts.pool_constraints(PoolConstraints::new(POOL_MIN, POOL_MAX));
+        let pool = Pool::new(opts);
         let pool_clone = pool.clone();
         let conns = (0..POOL_MAX).map(|_| pool.get_conn()).collect::<Vec<_>>();
 
-        let fut = ::futures::future::join_all(conns)
-            .and_then(|conns| {
-                // we want to continuously drop connections
-                // and check that they are _actually_ dropped until we reach POOL_MIN
-                assert_eq!(
-                    pool_clone.inner.exist.load(atomic::Ordering::SeqCst),
-                    POOL_MAX
-                );
-
-                future::loop_fn((pool_clone, conns), move |(pool_clone, mut conns)| {
-                    // first, drop a connection
-                    let _ = conns.pop();
-
-                    // then, wait for a bit to let the connection be reclaimed
-                    tokio::timer::Delay::new(
-                        std::time::Instant::now() + std::time::Duration::from_millis(100),
-                    )
-                    .map_err(|e| unimplemented!("{:?}", e))
-                    .map(|_| {
-                        // now check that we have the expected # of connections
-                        // this may look a little funky, but think of it this way:
-                        //
-                        //  - if we hold all 10 connections, we expect 10
-                        //  - if we drop one,  we still expect 10, because POOL_MIN limits
-                        //    the number of _idle_ connections (of which there is only 1)
-                        //  - once we've dropped 5, there are now 5 idle connections. thus,
-                        //    if we drop one more, we _now_ expect there to be only 9
-                        //    connections total (no more connections should be pushed to
-                        //    idle).
-                        let dropped = POOL_MAX - conns.len();
-                        let idle = min(dropped, POOL_MIN);
-                        let expected = conns.len() + idle;
-                        let have = pool_clone.inner.exist.load(atomic::Ordering::SeqCst);
-                        assert_eq!(have, expected);
-
-                        if conns.is_empty() {
-                            future::Loop::Break(pool_clone)
-                        } else {
-                            future::Loop::Continue((pool_clone, conns))
-                        }
-                    })
-                })
-            })
-            .and_then(|pool| pool.disconnect());
-
-        run(fut).unwrap();
-    }
+        let mut conns = futures_util::try_future::try_join_all(conns).await?;
 
-    #[test]
-    fn should_hold_bounds1() {
-        let pool = Pool::new(format!("{}?pool_min=1&pool_max=2", &**DATABASE_URL));
-        let pool_clone = pool.clone();
-        let fut = pool
-            .get_conn()
-            .join(pool.get_conn())
-            .and_then(move |(conn1, _conn2)| {
-                let new_conn = pool_clone.get_conn();
-                assert_eq!(
-                    conn1
-                        .inner
-                        .pool
-                        .as_ref()
-                        .unwrap()
-                        .inner
-                        .exist
-                        .load(atomic::Ordering::SeqCst),
-                    2
-                );
-                assert_eq!(conn1.inner.pool.as_ref().unwrap().inner.idle.len(), 0);
-                // NOTE: conn1 and conn2 are both dropped here
-                new_conn
-            })
-            .and_then(|conn1| {
-                // only one of conn1 and conn2 should have gone to idle,
-                // and should have immediately been picked up by new_conn (now conn1)
-                assert_eq!(conn1.inner.pool.as_ref().unwrap().inner.idle.len(), 0);
-                // NOTE: new_conn (now conn1) is dropped here
-                Ok(())
-            })
-            .and_then(|_| {
-                // the connection should be returned to idle
-                // (but may not have been returned _yet_)
-                assert!(pool.inner.idle.len() <= 1);
-                pool.disconnect()
-            });
+        // we want to continuously drop connections
+        // and check that they are _actually_ dropped until we reach POOL_MIN
+        assert_eq!(
+            pool_clone.inner.exist.load(atomic::Ordering::SeqCst),
+            POOL_MAX
+        );
+
+        while !conns.is_empty() {
+            // first, drop a connection
+            let _ = conns.pop();
+
+            // then, wait for a bit to let the connection be reclaimed
+            tokio::timer::delay(std::time::Instant::now() + std::time::Duration::from_millis(100))
+                .await;
+
+            // now check that we have the expected # of connections
+            // this may look a little funky, but think of it this way:
+            //
+            //  - if we hold all 10 connections, we expect 10
+            //  - if we drop one,  we still expect 10, because POOL_MIN limits
+            //    the number of _idle_ connections (of which there is only 1)
+            //  - once we've dropped 5, there are now 5 idle connections. thus,
+            //    if we drop one more, we _now_ expect there to be only 9
+            //    connections total (no more connections should be pushed to
+            //    idle).
+            let dropped = POOL_MAX - conns.len();
+            let idle = min(dropped, POOL_MIN);
+            let expected = conns.len() + idle;
+            let have = pool_clone.inner.exist.load(atomic::Ordering::SeqCst);
+            assert_eq!(have, expected);
+        }
 
-        run(fut).unwrap();
+        pool.disconnect().await?;
+        Ok(())
     }
 
-    #[test]
-    fn should_hold_bounds_on_error() {
-        // Test that connections which err do not count towards the connection count in the pool.
+    #[tokio::test]
+    async fn should_hold_bounds1() -> super::Result<()> {
+        let mut opts = get_opts();
+        opts.pool_constraints(PoolConstraints::new(1, 2));
+        let pool = Pool::new(opts);
+        let pool_clone = pool.clone();
 
-        let mut runtime = tokio::runtime::Runtime::new().unwrap();
+        let (conn1, conn2) = futures_util::try_future::try_join(pool.get_conn(), pool.get_conn())
+            .await
+            .unwrap();
 
+        assert_eq!(
+            conn1
+                .inner
+                .pool
+                .as_ref()
+                .unwrap()
+                .inner
+                .exist
+                .load(atomic::Ordering::SeqCst),
+            2
+        );
+        assert_eq!(conn1.inner.pool.as_ref().unwrap().inner.idle.len(), 0);
+
+        drop(conn1);
+        drop(conn2);
+        // only one of conn1 and conn2 should have gone to idle,
+        // and should have immediately been picked up by new_conn (now conn1)
+        let conn1 = pool_clone.get_conn().await?;
+        assert_eq!(conn1.inner.pool.as_ref().unwrap().inner.idle.len(), 0);
+
+        drop(conn1);
+
+        // the connection should be returned to idle
+        // (but may not have been returned _yet_)
+        assert!(pool.inner.idle.len() <= 1);
+        pool.disconnect().await?;
+        Ok(())
+    }
+
+    // Test that connections which err do not count towards the connection count in the pool.
+    #[tokio::test]
+    async fn should_hold_bounds_on_error() -> super::Result<()> {
         // Should not be possible to connect to broadcast address.
         let pool = Pool::new(String::from("mysql://255.255.255.255"));
 
-        let result = runtime.block_on(pool.get_conn().join(pool.get_conn()));
-
-        assert!(result.is_err());
+        assert!(
+            futures_util::try_future::try_join(pool.get_conn(), pool.get_conn())
+                .await
+                .is_err()
+        );
         assert_eq!(pool.inner.exist.load(atomic::Ordering::SeqCst), 0);
+        Ok(())
     }
 
+    /*
     #[test]
     fn should_hold_bounds_on_get_conn_drop() {
-        let pool = Pool::new(format!("{}?pool_min=1&pool_max=2", &**DATABASE_URL));
+        let pool = Pool::new(format!("{}?pool_min=1&pool_max=2", get_opts()));
         let mut runtime = tokio::runtime::Runtime::new().unwrap();
 
         // This test is a bit more intricate: we need to poll the connection future once to get the
@@ -745,19 +712,20 @@ mod test {
             }))
             .unwrap();
     }
+    */
 
-    #[test]
-    fn droptest() {
-        let pool = Pool::new(&**DATABASE_URL);
-        run(
-            collect((0..10).map(|_| pool.get_conn()).collect::<Vec<_>>()).map(move |conns| {
-                drop(conns);
-                drop(pool);
-            }),
-        )
-        .unwrap();
+    #[tokio::test]
+    async fn droptest() -> super::Result<()> {
+        let pool = Pool::new(get_opts());
+        let conns = futures_util::try_future::try_join_all((0..10).map(|_| pool.get_conn()))
+            .await
+            .unwrap();
+        drop(conns);
+        drop(pool);
+        Ok(())
     }
 
+    /*
     #[test]
     #[ignore]
     fn should_not_panic_if_dropped_without_tokio_runtime() {
@@ -768,25 +736,28 @@ mod test {
         //  - Runtime::shutdown_now is called
         //
         // none of these are true in this test, which is why it's been ignored
-        let pool = Pool::new(&**DATABASE_URL);
+        let pool = Pool::new(get_opts());
         run(collect(
             (0..10).map(|_| pool.get_conn()).collect::<Vec<_>>(),
         ))
         .unwrap();
         // pool will drop here
     }
+    */
 
     #[cfg(feature = "nightly")]
     mod bench {
-        use futures::Future;
+        use futures_util::try_future::TryFutureExt;
         use tokio::runtime::Runtime;
 
-        use crate::{conn::pool::Pool, queryable::Queryable, test_misc::DATABASE_URL};
+        use crate::prelude::Queryable;
+        use crate::test_misc::get_opts;
+        use crate::Pool;
 
         #[bench]
         fn connect(bencher: &mut test::Bencher) {
-            let mut runtime = Runtime::new().expect("3");
-            let pool = Pool::new(&**DATABASE_URL);
+            let runtime = Runtime::new().expect("3");
+            let pool = Pool::new(get_opts());
 
             bencher.iter(|| {
                 let fut = pool.get_conn().and_then(|conn| conn.ping());
diff --git a/src/connection_like/mod.rs b/src/connection_like/mod.rs
index 5206bf5d..f30523fe 100644
--- a/src/connection_like/mod.rs
+++ b/src/connection_like/mod.rs
@@ -7,12 +7,12 @@
 // modified, or distributed except according to those terms.
 
 use byteorder::{ByteOrder, LittleEndian};
-use futures::future::{err, loop_fn, ok, Either::*, Future, IntoFuture, Loop};
+use futures_util::future::ok;
 use mysql_common::{
     io::ReadMysqlExt,
     packets::{column_from_payload, parse_local_infile_packet, Column, RawPacket},
 };
-use tokio_io::io::read;
+use tokio::prelude::*;
 
 use std::sync::Arc;
 
@@ -28,7 +28,7 @@ use crate::{
         stmt::InnerStmt,
         Protocol,
     },
-    BoxFuture, MyFuture, Opts,
+    BoxFuture, Opts,
 };
 
 pub mod read_packet;
@@ -217,18 +217,19 @@ pub trait ConnectionLike: Send {
     where
         Self: Sized + 'static,
     {
-        let fut = if self.get_opts().get_stmt_cache_size() > 0 {
+        if self.get_opts().get_stmt_cache_size() > 0 {
             if let Some(old_stmt) = self.stmt_cache_mut().put(query, stmt.clone()) {
-                A(self
-                    .close_stmt(old_stmt.statement_id)
-                    .map(|this| (this, StmtCacheResult::Cached)))
+                let f = self.close_stmt(old_stmt.statement_id);
+                Box::pin(async move { Ok((f.await?, StmtCacheResult::Cached)) })
             } else {
-                B(ok((self, StmtCacheResult::Cached)))
+                Box::pin(futures_util::future::ok((self, StmtCacheResult::Cached)))
             }
         } else {
-            B(ok((self, StmtCacheResult::NotCached(stmt.statement_id))))
-        };
-        Box::new(fut)
+            Box::pin(futures_util::future::ok((
+                self,
+                StmtCacheResult::NotCached(stmt.statement_id),
+            )))
+        }
     }
 
     fn get_cached_stmt(&mut self, query: &str) -> Option<&InnerStmt> {
@@ -249,21 +250,18 @@ pub trait ConnectionLike: Send {
         Self: Sized + 'static,
     {
         if n == 0 {
-            return Box::new(ok((self, Vec::new())));
+            return Box::pin(ok((self, Vec::new())));
         }
-        let fut = loop_fn((Vec::new(), n - 1, self), |(mut acc, num, conn_like)| {
-            conn_like
-                .read_packet()
-                .and_then(move |(conn_like, packet)| {
-                    acc.push(packet);
-                    if num == 0 {
-                        Ok(Loop::Break((conn_like, acc)))
-                    } else {
-                        Ok(Loop::Continue((acc, num - 1, conn_like)))
-                    }
-                })
-        });
-        Box::new(fut)
+        Box::pin(async move {
+            let mut acc = Vec::new();
+            let mut conn_like = self;
+            for _ in 0..n {
+                let (cl, packet) = conn_like.read_packet().await?;
+                conn_like = cl;
+                acc.push(packet);
+            }
+            Ok((conn_like, acc))
+        })
     }
 
     fn prepare_stmt<Q>(mut self, query: Q) -> BoxFuture<(Self, InnerStmt, StmtCacheResult)>
@@ -276,102 +274,59 @@ pub trait ConnectionLike: Send {
                 let query = query.into_owned();
                 if let Some(mut inner_stmt) = self.get_cached_stmt(&query).map(Clone::clone) {
                     inner_stmt.named_params = named_params.clone();
-                    Box::new(ok((self, inner_stmt, StmtCacheResult::Cached)))
+                    Box::pin(ok((self, inner_stmt, StmtCacheResult::Cached)))
                 } else {
-                    let fut = self
-                        .write_command_data(Command::COM_STMT_PREPARE, &*query)
-                        .and_then(|this| this.read_packet())
-                        .and_then(|(this, packet)| {
-                            InnerStmt::new(&*packet.0, named_params)
-                                .into_future()
-                                .map(|inner_stmt| (this, inner_stmt))
-                        })
-                        .and_then(|(this, mut inner_stmt)| {
-                            this.read_packets(inner_stmt.num_params as usize)
-                                .and_then(|(this, packets)| {
-                                    let params = if !packets.is_empty() {
-                                        let params = packets
-                                            .into_iter()
-                                            .map(|packet| {
-                                                column_from_payload(packet.0).map_err(Error::from)
-                                            })
-                                            .collect::<Result<Vec<Column>>>();
-                                        Some(params)
-                                    } else {
-                                        None
-                                    };
-                                    match params {
-                                        Some(Err(error)) => return A(err(error)),
-                                        Some(Ok(params)) => inner_stmt.params = Some(params),
-                                        _ => (),
-                                    }
-                                    B(ok((this, inner_stmt)))
-                                })
-                                .and_then(|(this, inner_stmt)| {
-                                    if inner_stmt.num_params > 0 {
-                                        if this
-                                            .get_capabilities()
-                                            .contains(CapabilityFlags::CLIENT_DEPRECATE_EOF)
-                                        {
-                                            A(ok((this, inner_stmt)))
-                                        } else {
-                                            B(this
-                                                .read_packet()
-                                                .map(|(this, _)| (this, inner_stmt)))
-                                        }
-                                    } else {
-                                        A(ok((this, inner_stmt)))
-                                    }
-                                })
-                        })
-                        .and_then(|(this, mut inner_stmt)| {
-                            this.read_packets(inner_stmt.num_columns as usize)
-                                .and_then(|(this, packets)| {
-                                    let columns = if !packets.is_empty() {
-                                        let columns = packets
-                                            .into_iter()
-                                            .map(|packet| {
-                                                column_from_payload(packet.0).map_err(Error::from)
-                                            })
-                                            .collect::<Result<Vec<Column>>>();
-                                        Some(columns)
-                                    } else {
-                                        None
-                                    };
-                                    match columns {
-                                        Some(Err(error)) => return A(err(error)),
-                                        Some(Ok(columns)) => inner_stmt.columns = Some(columns),
-                                        _ => (),
-                                    }
-                                    B(ok((this, inner_stmt)))
-                                })
-                                .and_then(|(this, inner_stmt)| {
-                                    if inner_stmt.num_columns > 0 {
-                                        if this
-                                            .get_capabilities()
-                                            .contains(CapabilityFlags::CLIENT_DEPRECATE_EOF)
-                                        {
-                                            A(ok((this, inner_stmt)))
-                                        } else {
-                                            B(this
-                                                .read_packet()
-                                                .map(|(this, _)| (this, inner_stmt)))
-                                        }
-                                    } else {
-                                        A(ok((this, inner_stmt)))
-                                    }
-                                })
-                        })
-                        .and_then(|(this, inner_stmt)| {
-                            this.cache_stmt(query, &inner_stmt)
-                                .map(|(this, stmt_cache_result)| {
-                                    (this, inner_stmt, stmt_cache_result)
-                                })
-                        });
-                    Box::new(fut)
+                    Box::pin(async move {
+                        let (this, packet) = self
+                            .write_command_data(Command::COM_STMT_PREPARE, &*query)
+                            .await?
+                            .read_packet()
+                            .await?;
+                        let mut inner_stmt = InnerStmt::new(&*packet.0, named_params)?;
+                        let (mut this, packets) =
+                            this.read_packets(inner_stmt.num_params as usize).await?;
+                        if !packets.is_empty() {
+                            let params = packets
+                                .into_iter()
+                                .map(|packet| column_from_payload(packet.0).map_err(Error::from))
+                                .collect::<Result<Vec<Column>>>()?;
+                            inner_stmt.params = Some(params);
+                        }
+
+                        if inner_stmt.num_params > 0 {
+                            if !this
+                                .get_capabilities()
+                                .contains(CapabilityFlags::CLIENT_DEPRECATE_EOF)
+                            {
+                                this = this.read_packet().await?.0;
+                            }
+                        }
+
+                        let (mut this, packets) =
+                            this.read_packets(inner_stmt.num_columns as usize).await?;
+                        if !packets.is_empty() {
+                            let columns = packets
+                                .into_iter()
+                                .map(|packet| column_from_payload(packet.0).map_err(Error::from))
+                                .collect::<Result<Vec<Column>>>()?;
+                            inner_stmt.columns = Some(columns);
+                        }
+
+                        if inner_stmt.num_columns > 0 {
+                            if !this
+                                .get_capabilities()
+                                .contains(CapabilityFlags::CLIENT_DEPRECATE_EOF)
+                            {
+                                this = this.read_packet().await?.0;
+                            }
+                        }
+
+                        let (this, stmt_cache_result) = this.cache_stmt(query, &inner_stmt).await?;
+                        Ok((this, inner_stmt, stmt_cache_result))
+                    })
                 }
             }
-            Err(err) => Box::new(Err(Error::from(err)).into_future()),
+            Err(err) => Box::pin(async move { Err(Error::from(err)) }),
         }
     }
 
@@ -391,14 +346,14 @@ pub trait ConnectionLike: Send {
         P: Protocol,
         P: Send + 'static,
     {
-        let fut = self
-            .read_packet()
-            .and_then(|(this, packet)| match packet.0[0] {
-                0x00 => A(A(ok(query_result::new(this, None, cached)))),
-                0xFB => A(B(handle_local_infile(this, packet, cached))),
-                _ => B(handle_result_set(this, packet, cached)),
-            });
-        Box::new(fut)
+        Box::pin(async move {
+            let (this, packet) = self.read_packet().await?;
+            match packet.0[0] {
+                0x00 => Ok(query_result::new(this, None, cached)),
+                0xFB => handle_local_infile(this, packet, cached).await,
+                _ => handle_result_set(this, packet, cached).await,
+            }
+        })
     }
 
     /// Returns future that writes packet to a server end resolves to `Self`.
@@ -425,89 +380,64 @@ pub trait ConnectionLike: Send {
 }
 
 /// Will handle local infile packet.
-fn handle_local_infile<T, P>(
-    this: T,
+async fn handle_local_infile<T, P>(
+    mut this: T,
     packet: RawPacket,
     cached: Option<StmtCacheResult>,
-) -> impl MyFuture<QueryResult<T, P>>
+) -> Result<QueryResult<T, P>>
 where
     P: Protocol + 'static,
     T: ConnectionLike,
     T: Send + Sized + 'static,
 {
-    parse_local_infile_packet(&*packet.0)
-        .map_err(Error::from)
-        .and_then(|local_infile| match this.get_local_infile_handler() {
-            Some(handler) => Ok((local_infile.into_owned(), handler)),
-            None => Err(DriverError::NoLocalInfileHandler.into()),
-        })
-        .into_future()
-        .and_then(|(local_infile, handler)| handler.handle(local_infile.file_name_ref()))
-        .and_then(|reader| {
-            let mut buf = Vec::with_capacity(4096);
-            unsafe {
-                buf.set_len(4096);
-            }
-            loop_fn((this, buf, reader), |(this, buf, reader)| {
-                read(reader, buf)
-                    .map_err(Into::into)
-                    .and_then(|(reader, mut buf, count)| {
-                        unsafe {
-                            buf.set_len(count);
-                        }
-                        this.write_packet(&buf[..count])
-                            .map(move |this| (this, buf, reader, count))
-                    })
-                    .map(|(this, buf, reader, count)| {
-                        if count > 0 {
-                            Loop::Continue((this, buf, reader))
-                        } else {
-                            Loop::Break(this)
-                        }
-                    })
-            })
-            .and_then(|this| this.read_packet())
-            .map(|(this, _)| query_result::new(this, None, cached))
-        })
+    let local_infile = parse_local_infile_packet(&*packet.0)?;
+    let (local_infile, handler) = match this.get_local_infile_handler() {
+        Some(handler) => ((local_infile.into_owned(), handler)),
+        None => return Err(DriverError::NoLocalInfileHandler.into()),
+    };
+    let mut reader = handler.handle(local_infile.file_name_ref()).await?;
+
+    let mut buf = [0; 4096];
+    loop {
+        let read = reader.read(&mut buf[..]).await?;
+        this = this.write_packet(&buf[..read]).await?;
+
+        if read == 0 {
+            break;
+        }
+    }
+
+    let (this, _) = this.read_packet().await?;
+    Ok(query_result::new(this, None, cached))
 }
 
 /// Will handle result set packet.
-fn handle_result_set<T, P>(
+async fn handle_result_set<T, P>(
     this: T,
     packet: RawPacket,
     cached: Option<StmtCacheResult>,
-) -> impl MyFuture<QueryResult<T, P>>
+) -> Result<QueryResult<T, P>>
 where
     P: Protocol,
     P: Send + 'static,
     T: ConnectionLike,
     T: Send + Sized + 'static,
 {
-    (&*packet.0)
-        .read_lenenc_int()
-        .map_err(Into::into)
-        .into_future()
-        .and_then(|column_count| this.read_packets(column_count as usize))
-        .and_then(|(this, packets)| {
-            packets
-                .into_iter()
-                .map(|packet| column_from_payload(packet.0).map_err(Error::from))
-                .collect::<Result<Vec<Column>>>()
-                .into_future()
-                .and_then(|columns| {
-                    if this
-                        .get_capabilities()
-                        .contains(CapabilityFlags::CLIENT_DEPRECATE_EOF)
-                    {
-                        A(ok((this, columns)))
-                    } else {
-                        B(this.read_packet().map(|(this, _)| (this, columns)))
-                    }
-                })
-        })
-        .map(|(mut this, columns)| {
-            let columns = Arc::new(columns);
-            this.set_pending_result(Some((Clone::clone(&columns), None)));
-            query_result::new(this, Some(columns), cached)
-        })
+    let column_count = (&*packet.0).read_lenenc_int()?;
+    let (mut this, packets) = this.read_packets(column_count as usize).await?;
+    let columns = packets
+        .into_iter()
+        .map(|packet| column_from_payload(packet.0).map_err(Error::from))
+        .collect::<Result<Vec<Column>>>()?;
+
+    if !this
+        .get_capabilities()
+        .contains(CapabilityFlags::CLIENT_DEPRECATE_EOF)
+    {
+        this = this.read_packet().await?.0;
+    }
+
+    let columns = Arc::new(columns);
+    this.set_pending_result(Some((Clone::clone(&columns), None)));
+    Ok(query_result::new(this, Some(columns), cached))
 }
diff --git a/src/connection_like/read_packet.rs b/src/connection_like/read_packet.rs
index 81984d62..5cde4922 100644
--- a/src/connection_like/read_packet.rs
+++ b/src/connection_like/read_packet.rs
@@ -6,13 +6,13 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{
-    stream::{Stream, StreamFuture},
-    try_ready,
-    Async::Ready,
-    Future, Poll,
-};
+use futures_core::ready;
+use futures_util::stream::{StreamExt, StreamFuture};
 use mysql_common::packets::{parse_err_packet, parse_ok_packet, RawPacket};
+use pin_project::pin_project;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use crate::{
     connection_like::{streamless::Streamless, ConnectionLike},
@@ -20,8 +20,10 @@ use crate::{
     io,
 };
 
+#[pin_project]
 pub struct ReadPacket<T> {
     conn_like: Option<Streamless<T>>,
+    #[pin]
     fut: StreamFuture<io::Stream>,
 }
 
@@ -36,12 +38,13 @@ impl<T: ConnectionLike> ReadPacket<T> {
 }
 
 impl<T: ConnectionLike> Future for ReadPacket<T> {
-    type Item = (T, RawPacket);
-    type Error = Error;
+    type Output = Result<(T, RawPacket)>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let (packet_opt, stream) = try_ready!(self.fut.poll());
-        let mut conn_like = self.conn_like.take().unwrap().return_stream(stream);
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let (packet_opt, stream) = ready!(this.fut.poll(cx));
+        let packet_opt = packet_opt.transpose()?;
+        let mut conn_like = this.conn_like.take().unwrap().return_stream(stream);
         match packet_opt {
             Some((packet, seq_id)) => {
                 if let Ok(ok_packet) = parse_ok_packet(&*packet.0, conn_like.get_capabilities()) {
@@ -52,14 +55,14 @@ impl<T: ConnectionLike> Future for ReadPacket<T> {
                 } else if let Ok(err_packet) =
                     parse_err_packet(&*packet.0, conn_like.get_capabilities())
                 {
-                    return Err(err_packet.into());
+                    return Err(err_packet.into()).into();
                 }
 
                 conn_like.touch();
                 conn_like.set_seq_id(seq_id.wrapping_add(1));
-                Ok(Ready((conn_like, packet)))
+                Poll::Ready(Ok((conn_like, packet)))
             }
-            None => Err(DriverError::ConnectionClosed.into()),
+            None => Poll::Ready(Err(DriverError::ConnectionClosed.into())),
         }
     }
 }
diff --git a/src/connection_like/write_packet.rs b/src/connection_like/write_packet.rs
index db1bce1b..747fc8ac 100644
--- a/src/connection_like/write_packet.rs
+++ b/src/connection_like/write_packet.rs
@@ -6,7 +6,11 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{try_ready, Async::Ready, Future, Poll};
+use futures_core::ready;
+use pin_project::pin_project;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use crate::{
     connection_like::{streamless::Streamless, ConnectionLike},
@@ -14,8 +18,10 @@ use crate::{
     io,
 };
 
+#[pin_project]
 pub struct WritePacket<T> {
     conn_like: Option<Streamless<T>>,
+    #[pin]
     fut: io::futures::WritePacket,
 }
 
@@ -31,14 +37,14 @@ impl<T: ConnectionLike> WritePacket<T> {
 }
 
 impl<T: ConnectionLike> Future for WritePacket<T> {
-    type Item = T;
-    type Error = Error;
+    type Output = Result<T>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let (stream, seq_id) = try_ready!(self.fut.poll());
-        let mut conn_like = self.conn_like.take().unwrap().return_stream(stream);
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let (stream, seq_id) = ready!(this.fut.poll(cx))?;
+        let mut conn_like = this.conn_like.take().unwrap().return_stream(stream);
         conn_like.set_seq_id(seq_id);
         conn_like.touch();
-        Ok(Ready(conn_like))
+        Poll::Ready(Ok(conn_like))
     }
 }
diff --git a/src/error.rs b/src/error.rs
index e9ea3d8f..c3405168 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -19,6 +19,8 @@ use std::{io, result};
 /// Result type alias for this library.
 pub type Result<T> = result::Result<T, Error>;
 
+pub(crate) type StdResult<T, E> = result::Result<T, E>;
+
 /// This type enumerates library errors.
 #[derive(Debug, Fail)]
 pub enum Error {
@@ -34,7 +36,6 @@ pub enum Error {
     #[fail(display = "Server error: `{}'", _0)]
     Server(#[cause] ServerError),
 
-    #[cfg(feature = "ssl")]
     #[fail(display = "TLS error: `{}'", _0)]
     Tls(#[cause] native_tls::Error),
 
@@ -166,7 +167,6 @@ impl From<UrlError> for Error {
     }
 }
 
-#[cfg(feature = "ssl")]
 impl From<native_tls::Error> for Error {
     fn from(err: native_tls::Error) -> Self {
         Error::Tls(err)
diff --git a/src/io/async_tls.rs b/src/io/async_tls.rs
index 38439b5d..99445cda 100644
--- a/src/io/async_tls.rs
+++ b/src/io/async_tls.rs
@@ -9,11 +9,14 @@
 //! This module implements async TLS streams and source
 //! of this module is mostly copyed from tokio_tls crate.
 
-use futures::{Async, Future, Poll};
-use native_tls::{self, Error, HandshakeError, TlsConnector};
-use tokio_io::{AsyncRead, AsyncWrite};
-
-use std::io::{self, Read, Write};
+use bytes::buf::BufMut;
+use native_tls::{Error, TlsConnector};
+use pin_project::pin_project;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::Error as IoError;
+use tokio::prelude::*;
+use tokio_tls::{self};
 
 /// A wrapper around an underlying raw stream which implements the TLS or SSL
 /// protocol.
@@ -22,84 +25,65 @@ use std::io::{self, Read, Write};
 /// and both the server and the client are ready for receiving and sending
 /// data. Bytes read from a `TlsStream` are decrypted from `S` and bytes written
 /// to a `TlsStream` are encrypted when passing through to `S`.
+#[pin_project]
 #[derive(Debug)]
 pub struct TlsStream<S> {
-    inner: native_tls::TlsStream<S>,
+    #[pin]
+    inner: tokio_tls::TlsStream<S>,
 }
 
 impl<S> TlsStream<S> {
-    /// Get access to the internal `native_tls::TlsStream` stream which also
+    /// Get access to the internal `tokio_tls::TlsStream` stream which also
     /// transitively allows access to `S`.
-    pub fn get_ref(&self) -> &native_tls::TlsStream<S> {
+    pub fn get_ref(&self) -> &tokio_tls::TlsStream<S> {
         &self.inner
     }
 }
 
-impl<S: Read + Write> Read for TlsStream<S> {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.inner.read(buf)
+impl<S> AsyncRead for TlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, IoError>> {
+        self.project().inner.poll_read(cx, buf)
     }
-}
 
-impl<S: Read + Write> Write for TlsStream<S> {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.inner.write(buf)
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.inner.prepare_uninitialized_buffer(buf)
     }
 
-    fn flush(&mut self) -> io::Result<()> {
-        self.inner.flush()
+    fn poll_read_buf<B>(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut B,
+    ) -> Poll<Result<usize, IoError>>
+    where
+        B: BufMut,
+    {
+        self.project().inner.poll_read_buf(cx, buf)
     }
 }
 
-impl<S: AsyncRead + AsyncWrite> AsyncRead for TlsStream<S> {}
-
-impl<S: AsyncRead + AsyncWrite> AsyncWrite for TlsStream<S> {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        match self.inner.shutdown() {
-            Ok(t) => t,
-            Err(ref e) if e.kind() == ::std::io::ErrorKind::WouldBlock => {
-                return Ok(futures::Async::NotReady);
-            }
-            Err(e) => return Err(e.into()),
-        }
-        self.inner.get_mut().shutdown()
+impl<S> AsyncWrite for TlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<Result<usize, IoError>> {
+        self.project().inner.poll_write(cx, buf)
     }
-}
-
-pub struct ConnectAsync<S> {
-    inner: MidHandshake<S>,
-}
-
-impl<S: Read + Write> Future for ConnectAsync<S> {
-    type Item = TlsStream<S>;
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<TlsStream<S>, Error> {
-        self.inner.poll()
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), IoError>> {
+        self.project().inner.poll_flush(cx)
     }
-}
-
-struct MidHandshake<S> {
-    inner: Option<Result<native_tls::TlsStream<S>, HandshakeError<S>>>,
-}
-
-impl<S: Read + Write> Future for MidHandshake<S> {
-    type Item = TlsStream<S>;
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<TlsStream<S>, Error> {
-        match self.inner.take().expect("cannot poll MidHandshake twice") {
-            Ok(stream) => Ok(TlsStream { inner: stream }.into()),
-            Err(HandshakeError::Failure(e)) => Err(e),
-            Err(HandshakeError::WouldBlock(s)) => match s.handshake() {
-                Ok(stream) => Ok(TlsStream { inner: stream }.into()),
-                Err(HandshakeError::Failure(e)) => Err(e),
-                Err(HandshakeError::WouldBlock(s)) => {
-                    self.inner = Some(Err(HandshakeError::WouldBlock(s)));
-                    Ok(Async::NotReady)
-                }
-            },
-        }
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), IoError>> {
+        self.project().inner.poll_shutdown(cx)
     }
 }
 
@@ -115,20 +99,17 @@ impl<S: Read + Write> Future for MidHandshake<S> {
 /// example, a TCP connection to a remote server. That stream is then
 /// provided here to perform the client half of a connection to a
 /// TLS-powered server.
-///
-/// # Compatibility notes
-///
-/// Note that this method currently requires `S: Read + Write` but it's
-/// highly recommended to ensure that the object implements the `AsyncRead`
-/// and `AsyncWrite` traits as well, otherwise this function will not work
-/// properly.
-pub fn connect_async<S>(connector: &TlsConnector, domain: &str, stream: S) -> ConnectAsync<S>
+pub async fn connect_async<S>(
+    connector: &TlsConnector,
+    domain: &str,
+    stream: S,
+) -> Result<TlsStream<S>, Error>
 where
-    S: Read + Write,
+    S: AsyncRead + AsyncWrite + Unpin,
 {
-    ConnectAsync {
-        inner: MidHandshake {
-            inner: Some(connector.connect(domain, stream)),
-        },
-    }
+    let connector = tokio_tls::TlsConnector::from(connector.clone());
+    let connect = connector.connect(domain, stream);
+    Ok(TlsStream {
+        inner: connect.await?,
+    })
 }
diff --git a/src/io/futures/connecting_tcp_stream.rs b/src/io/futures/connecting_tcp_stream.rs
index c8109057..3b70f09f 100644
--- a/src/io/futures/connecting_tcp_stream.rs
+++ b/src/io/futures/connecting_tcp_stream.rs
@@ -6,15 +6,9 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{
-    failed,
-    future::{select_ok, SelectOk},
-    try_ready,
-    Async::{self, Ready},
-    Failed, Future, Poll,
-};
-use tokio::net::{tcp::ConnectFuture, TcpStream};
-use tokio_codec::Framed;
+use futures_util::stream::{FuturesUnordered, StreamExt};
+use tokio::codec::Framed;
+use tokio::net::TcpStream;
 
 use std::{io, net::ToSocketAddrs};
 
@@ -23,61 +17,43 @@ use crate::{
     io::{packet_codec::PacketCodec, Stream},
 };
 
-steps! {
-    ConnectingTcpStream {
-        WaitForStream(SelectOk<ConnectFuture>),
-        Fail(Failed<(), Error>),
-    }
-}
-
-/// Future that resolves to a `Stream` connected to a MySql server.
-pub struct ConnectingTcpStream {
-    step: Step,
-}
-
-pub fn new<S>(addr: S) -> ConnectingTcpStream
+pub async fn new<S>(addr: S) -> Result<Stream>
 where
     S: ToSocketAddrs,
 {
     match addr.to_socket_addrs() {
         Ok(addresses) => {
-            let mut streams = Vec::new();
+            let mut streams = FuturesUnordered::new();
 
             for address in addresses {
-                streams.push(TcpStream::connect(&address));
+                streams.push(TcpStream::connect(address));
             }
 
-            if !streams.is_empty() {
-                ConnectingTcpStream {
-                    step: Step::WaitForStream(select_ok(streams)),
+            let mut err = None;
+            while let Some(stream) = streams.next().await {
+                match stream {
+                    Err(e) => {
+                        err = Some(e);
+                    }
+                    Ok(stream) => {
+                        return Ok(Stream {
+                            closed: false,
+                            codec: Box::new(Framed::new(stream.into(), PacketCodec::new())).into(),
+                        });
+                    }
                 }
+            }
+
+            if let Some(e) = err {
+                Err(e.into())
             } else {
-                let err = io::Error::new(
+                Err(io::Error::new(
                     io::ErrorKind::InvalidInput,
                     "could not resolve to any address",
-                );
-                ConnectingTcpStream {
-                    step: Step::Fail(failed(err.into())),
-                }
+                )
+                .into())
             }
         }
-        Err(err) => ConnectingTcpStream {
-            step: Step::Fail(failed(err.into())),
-        },
-    }
-}
-
-impl Future for ConnectingTcpStream {
-    type Item = Stream;
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        match try_ready!(self.either_poll()) {
-            Out::WaitForStream((stream, _)) => Ok(Ready(Stream {
-                closed: false,
-                codec: Box::new(Framed::new(stream.into(), PacketCodec::new())).into(),
-            })),
-            Out::Fail(_) => unreachable!(),
-        }
+        Err(err) => Err(err.into()),
     }
 }
diff --git a/src/io/futures/mod.rs b/src/io/futures/mod.rs
index c47232c3..ae9bcd07 100644
--- a/src/io/futures/mod.rs
+++ b/src/io/futures/mod.rs
@@ -7,7 +7,7 @@
 // modified, or distributed except according to those terms.
 
 pub use self::{
-    connecting_tcp_stream::{new as new_connecting_tcp_stream, ConnectingTcpStream},
+    connecting_tcp_stream::new as new_connecting_tcp_stream,
     write_packet::{new as new_write_packet, WritePacket},
 };
 
diff --git a/src/io/futures/write_packet.rs b/src/io/futures/write_packet.rs
index a0110434..be1c13f6 100644
--- a/src/io/futures/write_packet.rs
+++ b/src/io/futures/write_packet.rs
@@ -6,8 +6,12 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{try_ready, Async, AsyncSink, Future, Poll, Sink};
+use futures_core::ready;
+use futures_sink::Sink;
 use mysql_common::packets::RawPacket;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use crate::{consts::MAX_PAYLOAD_LEN, error::*, io::Stream};
 
@@ -42,37 +46,22 @@ pub fn new(stream: Stream, data: Vec<u8>, seq_id: u8) -> WritePacket {
 }
 
 impl Future for WritePacket {
-    type Item = (Stream, u8);
-    type Error = Error;
+    type Output = Result<(Stream, u8)>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.data.is_some() {
+            ready!(Pin::new(self.stream.as_mut().unwrap().codec.as_mut().unwrap()).poll_ready(cx))?;
+        }
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         if let Some(data) = self.data.take() {
-            let send_result = self
-                .stream
-                .as_mut()
-                .unwrap()
-                .codec
-                .as_mut()
-                .unwrap()
-                .start_send((data, self.seq_id))?;
-            if let AsyncSink::NotReady(data) = send_result {
-                self.data = Some(data.0);
-                return Ok(Async::NotReady);
-            }
+            // to get here, stream must be ready
+            let id = self.seq_id;
+            Pin::new(self.stream.as_mut().unwrap().codec.as_mut().unwrap())
+                .start_send((data, id))?;
         }
 
-        try_ready!(self
-            .stream
-            .as_mut()
-            .unwrap()
-            .codec
-            .as_mut()
-            .unwrap()
-            .poll_complete()
-            .map_err(Error::from));
-        Ok(Async::Ready((
-            self.stream.take().unwrap(),
-            self.resulting_seq_id,
-        )))
+        ready!(Pin::new(self.stream.as_mut().unwrap().codec.as_mut().unwrap()).poll_flush(cx))
+            .map_err(Error::from)?;
+        Poll::Ready(Ok((self.stream.take().unwrap(), self.resulting_seq_id)))
     }
 }
diff --git a/src/io/mod.rs b/src/io/mod.rs
index c2137b47..8504d1bc 100644
--- a/src/io/mod.rs
+++ b/src/io/mod.rs
@@ -6,51 +6,44 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-#[cfg(feature = "ssl")]
-use ::futures::{future::Either::*, IntoFuture};
-use ::futures::{
-    future::{ok, Future},
-    stream, Async, Poll,
-};
+use bytes::BufMut;
+use futures_core::{ready, stream};
 use mysql_common::packets::RawPacket;
-#[cfg(feature = "ssl")]
 use native_tls::{Certificate, Identity, TlsConnector};
+use pin_project::{pin_project, project};
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::codec::Framed;
+use tokio::codec::FramedParts;
 use tokio::net::TcpStream;
-use tokio_codec::Framed;
-#[cfg(feature = "ssl")]
-use tokio_codec::FramedParts;
-use tokio_io::{AsyncRead, AsyncWrite};
+use tokio::prelude::*;
 
-use std::{fmt, io, net::ToSocketAddrs, path::Path, time::Duration};
-#[cfg(feature = "ssl")]
+use std::{fmt, net::ToSocketAddrs, path::Path, time::Duration};
 use std::{fs::File, io::Read};
 
 use crate::{
     error::*,
     io::{
-        futures::{new_connecting_tcp_stream, new_write_packet, ConnectingTcpStream, WritePacket},
+        futures::{new_connecting_tcp_stream, new_write_packet, WritePacket},
         socket::Socket,
     },
     opts::SslOpts,
-    MyFuture,
 };
 
-#[cfg(feature = "ssl")]
 mod async_tls;
 pub mod futures;
 mod packet_codec;
 mod socket;
 
+#[pin_project]
 #[derive(Debug)]
 pub enum Endpoint {
-    Plain(TcpStream),
-    #[cfg(feature = "ssl")]
-    Secure(self::async_tls::TlsStream<TcpStream>),
-    Socket(Socket),
+    Plain(#[pin] TcpStream),
+    Secure(#[pin] self::async_tls::TlsStream<TcpStream>),
+    Socket(#[pin] Socket),
 }
 
 impl Endpoint {
-    #[cfg(feature = "ssl")]
     pub fn is_secure(&self) -> bool {
         if let Endpoint::Secure(_) = self {
             true
@@ -59,16 +52,10 @@ impl Endpoint {
         }
     }
 
-    #[cfg(not(feature = "ssl"))]
-    pub fn is_secure(&self) -> bool {
-        false
-    }
-
     pub fn set_keepalive_ms(&self, ms: Option<u32>) -> Result<()> {
         let ms = ms.map(|val| Duration::from_millis(u64::from(val)));
         match *self {
             Endpoint::Plain(ref stream) => stream.set_keepalive(ms)?,
-            #[cfg(feature = "ssl")]
             Endpoint::Secure(ref stream) => stream.get_ref().get_ref().set_keepalive(ms)?,
             Endpoint::Socket(_) => (/* inapplicable */),
         }
@@ -78,53 +65,44 @@ impl Endpoint {
     pub fn set_tcp_nodelay(&self, val: bool) -> Result<()> {
         match *self {
             Endpoint::Plain(ref stream) => stream.set_nodelay(val)?,
-            #[cfg(feature = "ssl")]
             Endpoint::Secure(ref stream) => stream.get_ref().get_ref().set_nodelay(val)?,
             Endpoint::Socket(_) => (/* inapplicable */),
         }
         Ok(())
     }
 
-    #[cfg(feature = "ssl")]
-    pub fn make_secure(self, domain: String, ssl_opts: SslOpts) -> impl MyFuture<Self> {
+    pub async fn make_secure(self, domain: String, ssl_opts: SslOpts) -> Result<Self> {
         if let Endpoint::Socket(_) = self {
             // inapplicable
-            return A(ok(self));
+            return Ok(self);
         }
 
-        let fut = (|| {
-            let mut builder = TlsConnector::builder();
-            match ssl_opts.root_cert_path() {
-                Some(root_cert_path) => {
-                    let mut root_cert_der = vec![];
-                    let mut root_cert_file = File::open(root_cert_path)?;
-                    root_cert_file.read_to_end(&mut root_cert_der)?;
-                    let root_cert = Certificate::from_der(&*root_cert_der).map_err(Error::from)?;
-                    builder.add_root_certificate(root_cert);
-                }
-                None => (),
+        let mut builder = TlsConnector::builder();
+        match ssl_opts.root_cert_path() {
+            Some(root_cert_path) => {
+                let mut root_cert_der = vec![];
+                let mut root_cert_file = File::open(root_cert_path)?;
+                root_cert_file.read_to_end(&mut root_cert_der)?;
+                let root_cert = Certificate::from_der(&*root_cert_der)?;
+                builder.add_root_certificate(root_cert);
             }
-            if let Some(pkcs12_path) = ssl_opts.pkcs12_path() {
-                let der = std::fs::read(pkcs12_path)?;
-                let identity = Identity::from_pkcs12(&*der, ssl_opts.password().unwrap_or(""))
-                    .map_err(Error::from)?;
-                builder.identity(identity);
-            }
-            builder.danger_accept_invalid_hostnames(ssl_opts.skip_domain_validation());
-            builder.danger_accept_invalid_certs(ssl_opts.accept_invalid_certs());
-            builder.build().map_err(Error::from)
-        })()
-        .into_future()
-        .and_then(move |tls_connector| match self {
+            None => (),
+        }
+        if let Some(pkcs12_path) = ssl_opts.pkcs12_path() {
+            let der = std::fs::read(pkcs12_path)?;
+            let identity = Identity::from_pkcs12(&*der, ssl_opts.password().unwrap_or(""))?;
+            builder.identity(identity);
+        }
+        builder.danger_accept_invalid_hostnames(ssl_opts.skip_domain_validation());
+        builder.danger_accept_invalid_certs(ssl_opts.accept_invalid_certs());
+        let tls_connector = builder.build()?;
+        let tls_stream = match self {
             Endpoint::Plain(stream) => {
-                self::async_tls::connect_async(&tls_connector, &*domain, stream)
-                    .map_err(Error::from)
+                self::async_tls::connect_async(&tls_connector, &*domain, stream).await?
             }
             Endpoint::Secure(_) | Endpoint::Socket(_) => unreachable!(),
-        })
-        .map(|tls_stream| Endpoint::Secure(tls_stream));
-
-        B(fut)
+        };
+        Ok(Endpoint::Secure(tls_stream))
     }
 }
 
@@ -140,62 +118,91 @@ impl From<Socket> for Endpoint {
     }
 }
 
-#[cfg(feature = "ssl")]
 impl From<self::async_tls::TlsStream<TcpStream>> for Endpoint {
     fn from(stream: self::async_tls::TlsStream<TcpStream>) -> Self {
         Endpoint::Secure(stream)
     }
 }
 
-impl io::Read for Endpoint {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        match *self {
-            Endpoint::Plain(ref mut stream) => stream.read(buf),
-            #[cfg(feature = "ssl")]
-            Endpoint::Secure(ref mut stream) => stream.read(buf),
-            Endpoint::Socket(ref mut stream) => stream.read(buf),
+impl AsyncRead for Endpoint {
+    #[project]
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<StdResult<usize, tokio::io::Error>> {
+        #[project]
+        match self.project() {
+            Endpoint::Plain(stream) => stream.poll_read(cx, buf),
+            Endpoint::Secure(stream) => stream.poll_read(cx, buf),
+            Endpoint::Socket(stream) => stream.poll_read(cx, buf),
         }
     }
-}
 
-impl io::Write for Endpoint {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        match *self {
-            Endpoint::Plain(ref mut stream) => stream.write(buf),
-            #[cfg(feature = "ssl")]
-            Endpoint::Secure(ref mut stream) => stream.write(buf),
-            Endpoint::Socket(ref mut stream) => stream.write(buf),
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        match self {
+            Endpoint::Plain(stream) => stream.prepare_uninitialized_buffer(buf),
+            Endpoint::Secure(stream) => stream.prepare_uninitialized_buffer(buf),
+            Endpoint::Socket(stream) => stream.prepare_uninitialized_buffer(buf),
         }
     }
 
-    fn flush(&mut self) -> io::Result<()> {
-        match *self {
-            Endpoint::Plain(ref mut stream) => stream.flush(),
-            #[cfg(feature = "ssl")]
-            Endpoint::Secure(ref mut stream) => stream.flush(),
-            Endpoint::Socket(ref mut stream) => stream.flush(),
+    #[project]
+    fn poll_read_buf<B>(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut B,
+    ) -> Poll<StdResult<usize, tokio::io::Error>>
+    where
+        B: BufMut,
+    {
+        #[project]
+        match self.project() {
+            Endpoint::Plain(stream) => stream.poll_read_buf(cx, buf),
+            Endpoint::Secure(stream) => stream.poll_read_buf(cx, buf),
+            Endpoint::Socket(stream) => stream.poll_read_buf(cx, buf),
         }
     }
 }
 
-impl AsyncRead for Endpoint {
-    unsafe fn prepare_uninitialized_buffer(&self, _buf: &mut [u8]) -> bool {
-        match *self {
-            Endpoint::Plain(ref stream) => stream.prepare_uninitialized_buffer(_buf),
-            #[cfg(feature = "ssl")]
-            Endpoint::Secure(ref stream) => stream.prepare_uninitialized_buffer(_buf),
-            Endpoint::Socket(ref stream) => stream.prepare_uninitialized_buffer(_buf),
+impl AsyncWrite for Endpoint {
+    #[project]
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<StdResult<usize, tokio::io::Error>> {
+        #[project]
+        match self.project() {
+            Endpoint::Plain(stream) => stream.poll_write(cx, buf),
+            Endpoint::Secure(stream) => stream.poll_write(cx, buf),
+            Endpoint::Socket(stream) => stream.poll_write(cx, buf),
         }
     }
-}
 
-impl AsyncWrite for Endpoint {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        match *self {
-            Endpoint::Plain(ref mut stream) => stream.shutdown(),
-            #[cfg(feature = "ssl")]
-            Endpoint::Secure(ref mut stream) => stream.shutdown(),
-            Endpoint::Socket(ref mut stream) => stream.shutdown(),
+    #[project]
+    fn poll_flush(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+    ) -> Poll<StdResult<(), tokio::io::Error>> {
+        #[project]
+        match self.project() {
+            Endpoint::Plain(stream) => stream.poll_flush(cx),
+            Endpoint::Secure(stream) => stream.poll_flush(cx),
+            Endpoint::Socket(stream) => stream.poll_flush(cx),
+        }
+    }
+
+    #[project]
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+    ) -> Poll<StdResult<(), tokio::io::Error>> {
+        #[project]
+        match self.project() {
+            Endpoint::Plain(stream) => stream.poll_shutdown(cx),
+            Endpoint::Secure(stream) => stream.poll_shutdown(cx),
+            Endpoint::Socket(stream) => stream.poll_shutdown(cx),
         }
     }
 }
@@ -226,15 +233,15 @@ impl Stream {
         }
     }
 
-    pub fn connect_tcp<S>(addr: S) -> ConnectingTcpStream
+    pub async fn connect_tcp<S>(addr: S) -> Result<Stream>
     where
         S: ToSocketAddrs,
     {
-        new_connecting_tcp_stream(addr)
+        new_connecting_tcp_stream(addr).await
     }
 
-    pub fn connect_socket<P: AsRef<Path>>(path: P) -> impl Future<Item = Stream, Error = Error> {
-        Socket::new(path).map(Stream::new).map_err(Error::from)
+    pub async fn connect_socket<P: AsRef<Path>>(path: P) -> Result<Stream> {
+        Ok(Stream::new(Socket::new(path).await?))
     }
 
     pub fn write_packet(self, data: Vec<u8>, seq_id: u8) -> WritePacket {
@@ -249,21 +256,13 @@ impl Stream {
         self.codec.as_ref().unwrap().get_ref().set_tcp_nodelay(val)
     }
 
-    #[cfg(not(feature = "ssl"))]
-    #[allow(unused)]
-    pub fn make_secure(self, domain: String, ssl_opts: SslOpts) -> impl MyFuture<Self> {
-        ok(panic!("Ssl connection requires `ssl` feature"))
-    }
-
-    #[cfg(feature = "ssl")]
-    pub fn make_secure(mut self, domain: String, ssl_opts: SslOpts) -> impl MyFuture<Self> {
+    pub async fn make_secure(mut self, domain: String, ssl_opts: SslOpts) -> Result<Self> {
         let codec = self.codec.take().unwrap();
         let FramedParts { io, codec, .. } = codec.into_parts();
-        io.make_secure(domain, ssl_opts).map(move |endpoint| {
-            let codec = Framed::new(endpoint, codec);
-            self.codec = Some(Box::new(codec));
-            self
-        })
+        let endpoint = io.make_secure(domain, ssl_opts).await?;
+        let codec = Framed::new(endpoint, codec);
+        self.codec = Some(Box::new(codec));
+        Ok(self)
     }
 
     pub fn is_secure(&self) -> bool {
@@ -272,14 +271,16 @@ impl Stream {
 }
 
 impl stream::Stream for Stream {
-    type Item = (RawPacket, u8);
-    type Error = Error;
+    type Item = Result<(RawPacket, u8)>;
 
-    fn poll(&mut self) -> Poll<Option<(RawPacket, u8)>, Error> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         if !self.closed {
-            self.codec.as_mut().unwrap().poll().map_err(Error::from)
+            Poll::Ready(
+                Ok(ready!(Pin::new(self.codec.as_mut().unwrap()).poll_next(cx)).transpose()?)
+                    .transpose(),
+            )
         } else {
-            Ok(Async::Ready(None))
+            Poll::Ready(None)
         }
     }
 }
diff --git a/src/io/packet_codec.rs b/src/io/packet_codec.rs
index 191647a6..4e72d736 100644
--- a/src/io/packet_codec.rs
+++ b/src/io/packet_codec.rs
@@ -9,7 +9,7 @@
 use byteorder::{ByteOrder, LittleEndian};
 use bytes::{BufMut, BytesMut};
 use mysql_common::{constants::MAX_PAYLOAD_LEN, packets::RawPacket};
-use tokio_codec::{Decoder, Encoder};
+use tokio::codec::{Decoder, Encoder};
 
 use std::io;
 
diff --git a/src/io/socket.rs b/src/io/socket.rs
index 80a2261d..9fa81bc9 100644
--- a/src/io/socket.rs
+++ b/src/io/socket.rs
@@ -6,85 +6,84 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use bytes::{Buf, BufMut};
-use futures::{future::Future, Async};
-use tokio::io::{AsyncRead, AsyncWrite};
+use bytes::BufMut;
+use pin_project::pin_project;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::Error;
+use tokio::prelude::*;
 
-use std::{
-    io::{self, Read, Write},
-    path::Path,
-};
+use std::{io, path::Path};
 
 /// Unix domain socket connection on unix, or named pipe connection on windows.
+#[pin_project]
 #[derive(Debug)]
 pub struct Socket {
+    #[pin]
     #[cfg(unix)]
-    inner: tokio_uds::UnixStream,
+    inner: tokio::net::unix::UnixStream,
+    #[pin]
     #[cfg(windows)]
-    inner: tokio_named_pipes::NamedPipe,
+    inner: tokio_net::util::PollEvented<mio_named_pipes::NamedPipe>,
 }
 
 impl Socket {
     /// Connects a new socket.
     #[cfg(unix)]
-    pub fn new<P: AsRef<Path>>(path: P) -> impl Future<Item = Socket, Error = io::Error> {
-        tokio_uds::UnixStream::connect(path).map(|socket| Socket { inner: socket })
+    pub async fn new<P: AsRef<Path>>(path: P) -> Result<Socket, io::Error> {
+        Ok(Socket {
+            inner: tokio::net::unix::UnixStream::connect(path).await?,
+        })
     }
 
     /// Connects a new socket.
     #[cfg(windows)]
-    pub fn new<P: AsRef<Path>>(path: P) -> impl Future<Item = Socket, Error = io::Error> {
-        use futures::future::IntoFuture;
-        use tokio_named_pipes::NamedPipe;
-
-        let handle = tokio::reactor::Handle::default();
-        NamedPipe::new(path.as_ref(), &handle)
-            .and_then(|pipe| {
-                pipe.connect()?;
-                Ok(Socket { inner: pipe })
-            })
-            .into_future()
-    }
-}
-
-impl Read for Socket {
-    fn read(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> {
-        self.inner.read(buf)
+    pub async fn new<P: AsRef<Path>>(path: P) -> Result<Socket, io::Error> {
+        let pipe = mio_named_pipes::NamedPipe::new(path.as_ref())?;
+        pipe.connect()?;
+        Ok(Socket {
+            inner: tokio_net::util::PollEvented::new(pipe),
+        })
     }
 }
 
-impl Write for Socket {
-    fn write(&mut self, buf: &[u8]) -> Result<usize, io::Error> {
-        self.inner.write(buf)
+impl AsyncRead for Socket {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut [u8],
+    ) -> Poll<Result<usize, Error>> {
+        self.project().inner.poll_read(cx, buf)
     }
 
-    fn flush(&mut self) -> Result<(), io::Error> {
-        self.inner.flush()
-    }
-}
-
-impl AsyncRead for Socket {
     unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
         self.inner.prepare_uninitialized_buffer(buf)
     }
 
-    fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Result<Async<usize>, io::Error>
+    fn poll_read_buf<B>(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut B,
+    ) -> Poll<Result<usize, Error>>
     where
-        Self: Sized,
+        B: BufMut,
     {
-        self.inner.read_buf(buf)
+        self.project().inner.poll_read_buf(cx, buf)
     }
 }
 
 impl AsyncWrite for Socket {
-    fn shutdown(&mut self) -> Result<Async<()>, io::Error> {
-        AsyncWrite::shutdown(&mut self.inner)
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<Result<usize, Error>> {
+        self.project().inner.poll_write(cx, buf)
     }
-
-    fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<Async<usize>, io::Error>
-    where
-        Self: Sized,
-    {
-        self.inner.write_buf(buf)
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
+        self.project().inner.poll_flush(cx)
+    }
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
+        self.project().inner.poll_shutdown(cx)
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 950fc02f..9274ca5f 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -20,14 +20,7 @@
 //! ### Example
 //!
 //! ```rust
-//! extern crate futures;
-//! #[macro_use]
-//! extern crate mysql_async as my;
-//! extern crate tokio;
-//! // ...
-//!
-//! use futures::Future;
-//! use my::prelude::*;
+//! use mysql_async::prelude::*;
 //! # use std::env;
 //!
 //! #[derive(Debug, PartialEq, Eq, Clone)]
@@ -37,21 +30,8 @@
 //!     account_name: Option<String>,
 //! }
 //!
-//! /// Same as `tokio::run`, but will panic if future panics and will return the result
-//! /// of future execution.
-//! pub fn run<F, T, U>(future: F) -> Result<T, U>
-//! where
-//!     F: Future<Item = T, Error = U> + Send + 'static,
-//!     T: Send + 'static,
-//!     U: Send + 'static,
-//! {
-//!     let mut runtime = tokio::runtime::Runtime::new().unwrap();
-//!     let result = runtime.block_on(future);
-//!     runtime.shutdown_on_idle().wait().unwrap();
-//!     result
-//! }
-//!
-//! fn main() {
+//! #[tokio::main]
+//! async fn main() -> Result<(), mysql_async::error::Error> {
 //!     let payments = vec![
 //!         Payment { customer_id: 1, amount: 2, account_name: None },
 //!         Payment { customer_id: 3, amount: 4, account_name: Some("foo".into()) },
@@ -61,60 +41,63 @@
 //!     ];
 //!     let payments_clone = payments.clone();
 //!
-//!     # let database_url: String = if let Ok(url) = env::var("DATABASE_URL") {
-//!     #     let opts = my::Opts::from_url(&url).expect("DATABASE_URL invalid");
+//!     let database_url = /* ... */
+//!     # if let Ok(url) = env::var("DATABASE_URL") {
+//!     #     let opts = mysql_async::Opts::from_url(&url).expect("DATABASE_URL invalid");
 //!     #     if opts.get_db_name().expect("a database name is required").is_empty() {
 //!     #         panic!("database name is empty");
 //!     #     }
 //!     #     url
 //!     # } else {
-//!     #     "mysql://root:password@127.0.0.1:3307/mysql".into()
+//!     #     "mysql://root:password@127.0.0.1:3307/mysql".to_string()
 //!     # };
 //!
-//!     let pool = my::Pool::new(database_url);
-//!     let future = pool.get_conn().and_then(|conn| {
-//!         // Create temporary table
-//!         conn.drop_query(
-//!             r"CREATE TEMPORARY TABLE payment (
-//!                 customer_id int not null,
-//!                 amount int not null,
-//!                 account_name text
-//!             )"
-//!         )
-//!     }).and_then(move |conn| {
-//!         // Save payments
-//!         let params = payments_clone.into_iter().map(|payment| {
-//!             params! {
-//!                 "customer_id" => payment.customer_id,
-//!                 "amount" => payment.amount,
-//!                 "account_name" => payment.account_name.clone(),
-//!             }
-//!         });
+//!     let pool = mysql_async::Pool::new(database_url);
+//!     let conn = pool.get_conn().await?;
+//!
+//!     // Create temporary table
+//!     let conn = conn.drop_query(
+//!         r"CREATE TEMPORARY TABLE payment (
+//!             customer_id int not null,
+//!             amount int not null,
+//!             account_name text
+//!         )"
+//!     ).await?;
 //!
-//!         conn.batch_exec(r"INSERT INTO payment (customer_id, amount, account_name)
-//!                         VALUES (:customer_id, :amount, :account_name)", params)
-//!     }).and_then(|conn| {
-//!         // Load payments from database.
-//!         conn.prep_exec("SELECT customer_id, amount, account_name FROM payment", ())
-//!     }).and_then(|result| {
-//!         // Collect payments
-//!         result.map_and_drop(|row| {
-//!             let (customer_id, amount, account_name) = my::from_row(row);
-//!             Payment {
-//!                 customer_id: customer_id,
-//!                 amount: amount,
-//!                 account_name: account_name,
-//!             }
-//!         })
-//!     }).and_then(|(_ /* conn */, payments)| {
-//!         // The destructor of a connection will return it to the pool,
-//!         // but pool should be disconnected explicitly because it's
-//!         // an asynchronous procedure.
-//!         pool.disconnect().map(|_| payments)
+//!     // Save payments
+//!     let params = payments_clone.into_iter().map(|payment| {
+//!         params! {
+//!             "customer_id" => payment.customer_id,
+//!             "amount" => payment.amount,
+//!             "account_name" => payment.account_name.clone(),
+//!         }
 //!     });
 //!
-//!     let loaded_payments = run(future).unwrap();
+//!     let conn = conn.batch_exec(r"INSERT INTO payment (customer_id, amount, account_name)
+//!                     VALUES (:customer_id, :amount, :account_name)", params).await?;
+//!
+//!     // Load payments from database.
+//!     let result = conn.prep_exec("SELECT customer_id, amount, account_name FROM payment", ()).await?;
+//!
+//!     // Collect payments
+//!     let (_ /* conn */, loaded_payments) = result.map_and_drop(|row| {
+//!         let (customer_id, amount, account_name) = mysql_async::from_row(row);
+//!         Payment {
+//!             customer_id: customer_id,
+//!             amount: amount,
+//!             account_name: account_name,
+//!         }
+//!     }).await?;
+//!
+//!     // The destructor of a connection will return it to the pool,
+//!     // but pool should be disconnected explicitly because it's
+//!     // an asynchronous procedure.
+//!     pool.disconnect().await?;
+//!
 //!     assert_eq!(loaded_payments, payments);
+//!
+//!     // the async fn returns Result, so
+//!     Ok(())
 //! }
 //! ```
 
@@ -137,14 +120,9 @@ mod local_infile_handler;
 mod opts;
 mod queryable;
 
-pub type BoxFuture<T> = Box<dyn ::futures::Future<Item = T, Error = error::Error> + Send + 'static>;
-
-/// Alias for `Future` with library error as `Future::Error`.
-pub trait MyFuture<T>: ::futures::Future<Item = T, Error = error::Error> + Send + 'static {}
-impl<T, U> MyFuture<T> for U where
-    U: ::futures::Future<Item = T, Error = error::Error> + Send + 'static
-{
-}
+pub type BoxFuture<T> = ::std::pin::Pin<
+    Box<dyn ::std::future::Future<Output = Result<T, error::Error>> + Send + 'static>,
+>;
 
 #[doc(inline)]
 pub use self::conn::Conn;
@@ -159,7 +137,7 @@ pub use self::queryable::transaction::IsolationLevel;
 pub use self::opts::{Opts, OptsBuilder, PoolConstraints, SslOpts};
 
 #[doc(inline)]
-pub use self::local_infile_handler::builtin::WhiteListFsLocalInfileHandler;
+pub use self::local_infile_handler::{builtin::WhiteListFsLocalInfileHandler, InfileHandlerFuture};
 
 #[doc(inline)]
 pub use mysql_common::packets::Column;
@@ -197,9 +175,6 @@ pub use self::queryable::stmt::Stmt;
 /// Futures used in this crate
 pub mod futures {
     pub use crate::conn::pool::futures::{DisconnectPool, GetConn};
-    pub use crate::queryable::query_result::{
-        ForEach, ForEachAndDrop, Map, MapAndDrop, Reduce, ReduceAndDrop,
-    };
 }
 
 /// Traits used in this crate
@@ -220,6 +195,8 @@ pub mod prelude {
     /// Trait for protocol markers [`TextProtocol`] and [`BinaryProtocol`].
     pub trait Protocol: crate::queryable::Protocol {}
     impl<T: crate::queryable::Protocol> Protocol for T {}
+
+    pub use mysql_common::params;
 }
 
 #[cfg(test)]
@@ -228,7 +205,7 @@ mod test_misc {
 
     use std::env;
 
-    use crate::opts;
+    use crate::opts::{Opts, OptsBuilder, SslOpts};
 
     #[allow(dead_code)]
     fn error_should_implement_send_and_sync() {
@@ -239,7 +216,7 @@ mod test_misc {
     lazy_static! {
         pub static ref DATABASE_URL: String = {
             if let Ok(url) = env::var("DATABASE_URL") {
-                let opts = opts::Opts::from_url(&url).expect("DATABASE_URL invalid");
+                let opts = Opts::from_url(&url).expect("DATABASE_URL invalid");
                 if opts
                     .get_db_name()
                     .expect("a database name is required")
@@ -253,4 +230,18 @@ mod test_misc {
             }
         };
     }
+
+    pub fn get_opts() -> OptsBuilder {
+        let mut builder = OptsBuilder::from_opts(&**DATABASE_URL);
+        // to suppress warning on unused mut
+        builder.stmt_cache_size(None);
+        if ["true", "1"].contains(&&*env::var("SSL").unwrap_or("".into())) {
+            builder.prefer_socket(false);
+            let mut ssl_opts = SslOpts::default();
+            ssl_opts.set_danger_skip_domain_validation(true);
+            ssl_opts.set_danger_accept_invalid_certs(true);
+            builder.ssl_opts(ssl_opts);
+        }
+        builder
+    }
 }
diff --git a/src/local_infile_handler/builtin.rs b/src/local_infile_handler/builtin.rs
index e1b6da55..a660265b 100644
--- a/src/local_infile_handler/builtin.rs
+++ b/src/local_infile_handler/builtin.rs
@@ -6,13 +6,11 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{Future, IntoFuture};
 use tokio::fs::File;
-use tokio_io::AsyncRead;
 
 use std::{collections::HashSet, path::PathBuf, str::from_utf8};
 
-use crate::{local_infile_handler::LocalInfileHandler, BoxFuture};
+use crate::local_infile_handler::LocalInfileHandler;
 
 /// Handles local infile requests from filesystem using explicit path white list.
 ///
@@ -49,21 +47,17 @@ impl WhiteListFsLocalInfileHandler {
 }
 
 impl LocalInfileHandler for WhiteListFsLocalInfileHandler {
-    fn handle(&self, file_name: &[u8]) -> BoxFuture<Box<dyn AsyncRead + Send + 'static>> {
+    fn handle(&self, file_name: &[u8]) -> super::InfileHandlerFuture {
         let path: PathBuf = match from_utf8(file_name) {
             Ok(path_str) => path_str.into(),
-            Err(_) => return Box::new(Err("Invalid file name".into()).into_future()),
+            Err(_) => return Box::pin(futures_util::future::err("Invalid file name".into())),
         };
 
         if !self.white_list.contains(&path) {
             let err_msg = format!("Path `{}' is not in white list", path.display());
-            return Box::new(Err(err_msg.into()).into_future());
+            return Box::pin(futures_util::future::err(err_msg.into()));
         }
 
-        let future = File::open(path.to_owned())
-            .map(|file| Box::new(file) as Box<_>)
-            .map_err(|e| e.into());
-
-        Box::new(future)
+        Box::pin(async move { Ok(Box::new(File::open(path.to_owned()).await?) as Box<_>) })
     }
 }
diff --git a/src/local_infile_handler/mod.rs b/src/local_infile_handler/mod.rs
index 93c5bc1b..0b418d0e 100644
--- a/src/local_infile_handler/mod.rs
+++ b/src/local_infile_handler/mod.rs
@@ -6,12 +6,14 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use tokio_io::AsyncRead;
+use crate::error;
+use tokio::prelude::*;
 
+use std::future::Future;
+use std::marker::Unpin;
+use std::pin::Pin;
 use std::{fmt, sync::Arc};
 
-use crate::BoxFuture;
-
 pub mod builtin;
 
 /// Trait used to handle local infile requests.
@@ -22,39 +24,23 @@ pub mod builtin;
 /// Simple handler example:
 ///
 /// ```rust
-/// # extern crate futures;
-/// # extern crate mysql_async as my;
-/// # extern crate tokio;
-/// # extern crate tokio_io;
-///
-/// # use futures::Future;
-/// # use my::prelude::*;
-/// # use tokio_io::AsyncRead;
+/// # use mysql_async::prelude::*;
+/// # use tokio::prelude::*;
 /// # use std::env;
-/// # fn main() {
-///
-/// # pub fn run<F, T, U>(future: F) -> Result<T, U>
-/// # where
-/// #     F: Future<Item = T, Error = U> + Send + 'static,
-/// #     T: Send + 'static,
-/// #     U: Send + 'static,
-/// # {
-/// #     let mut runtime = tokio::runtime::Runtime::new().unwrap();
-/// #     let result = runtime.block_on(future);
-/// #     runtime.shutdown_on_idle().wait().unwrap();
-/// #     result
-/// # }
-///
+/// # #[tokio::main]
+/// # async fn main() -> Result<(), mysql_async::error::Error> {
+/// #
 /// struct ExampleHandler(&'static [u8]);
 ///
 /// impl LocalInfileHandler for ExampleHandler {
-///     fn handle(&self, _: &[u8]) -> Box<Future<Item=Box<AsyncRead + Send>, Error=my::error::Error> + Send> {
-///         Box::new(futures::future::ok(Box::new(self.0) as Box<_>))
+///     fn handle(&self, _: &[u8]) -> mysql_async::InfileHandlerFuture {
+///         let handler = Box::new(self.0) as Box<_>;
+///         Box::pin(async move { Ok(handler) })
 ///     }
 /// }
 ///
 /// # let database_url: String = if let Ok(url) = env::var("DATABASE_URL") {
-/// #     let opts = my::Opts::from_url(&url).expect("DATABASE_URL invalid");
+/// #     let opts = mysql_async::Opts::from_url(&url).expect("DATABASE_URL invalid");
 /// #     if opts.get_db_name().expect("a database name is required").is_empty() {
 /// #         panic!("database name is empty");
 /// #     }
@@ -63,29 +49,30 @@ pub mod builtin;
 /// #     "mysql://root:password@127.0.0.1:3307/mysql".into()
 /// # };
 ///
-/// let mut opts = my::OptsBuilder::from_opts(&*database_url);
+/// let mut opts = mysql_async::OptsBuilder::from_opts(&*database_url);
 /// opts.local_infile_handler(Some(ExampleHandler(b"foobar")));
 ///
-/// let pool = my::Pool::new(opts);
+/// let pool = mysql_async::Pool::new(opts);
 ///
-/// let future = pool.get_conn()
-///     .and_then(|conn| conn.drop_query("CREATE TEMPORARY TABLE tmp (a TEXT);"))
-///     .and_then(|conn| conn.drop_query("LOAD DATA LOCAL INFILE 'baz' INTO TABLE tmp;"))
-///     .and_then(|conn| conn.prep_exec("SELECT * FROM tmp;", ()))
-///     .and_then(|result| result.map_and_drop(|row| my::from_row::<(String,)>(row).0))
-///     .map(|(_ /* conn */, result)| {
-///         assert_eq!(result.len(), 1);
-///         assert_eq!(result[0], "foobar");
-///     })
-///     .and_then(|_| pool.disconnect())
-///     .map_err(|err| match err {
-///         my::error::Error::Server(ref err) if err.code == 1148 => {
-///             // The used command is not allowed with this MySQL version
-///         },
-///         _ => panic!("{}", err),
-///     });
+/// let conn = pool.get_conn().await?;
+/// let conn = conn.drop_query("CREATE TEMPORARY TABLE tmp (a TEXT);").await?;
+/// let conn = match conn.drop_query("LOAD DATA LOCAL INFILE 'baz' INTO TABLE tmp;").await {
+///     Ok(conn) => conn,
+///     Err(mysql_async::error::Error::Server(ref err)) if err.code == 1148 => {
+///         // The used command is not allowed with this MySQL version
+///         return Ok(());
+///     },
+///     e@Err(_) => e.unwrap(),
+/// };
+/// let result = conn.prep_exec("SELECT * FROM tmp;", ()).await?;
+/// let (_ /* conn */, result) = result.map_and_drop(|row| {
+///     mysql_async::from_row::<(String,)>(row).0
+/// }).await?;
 ///
-///     run(future);
+/// assert_eq!(result.len(), 1);
+/// assert_eq!(result[0], "foobar");
+/// pool.disconnect().await?;
+/// # Ok(())
 /// # }
 /// ```
 ///
@@ -93,9 +80,17 @@ pub mod builtin;
 pub trait LocalInfileHandler: Sync + Send {
     /// `file_name` is the file name in `LOAD DATA LOCAL INFILE '<file name>' INTO TABLE ...;`
     /// query.
-    fn handle(&self, file_name: &[u8]) -> BoxFuture<Box<dyn AsyncRead + Send + 'static>>;
+    fn handle(&self, file_name: &[u8]) -> InfileHandlerFuture;
 }
 
+pub type InfileHandlerFuture = Pin<
+    Box<
+        dyn Future<Output = Result<Box<dyn AsyncRead + Send + Unpin + 'static>, error::Error>>
+            + Send
+            + 'static,
+    >,
+>;
+
 /// Object used to wrap `T: LocalInfileHandler` inside of Opts.
 #[derive(Clone)]
 pub struct LocalInfileHandlerObject(Arc<dyn LocalInfileHandler>);
diff --git a/src/macros.rs b/src/macros.rs
index 8401c095..555fcffc 100644
--- a/src/macros.rs
+++ b/src/macros.rs
@@ -12,23 +12,3 @@ macro_rules! const_assert {
         const $name: [(); 0 - !($($xs)&&+) as usize] = [];
     };
 }
-
-macro_rules! steps {
-    ($fut:ty { $($step:ident($ty:ty),)+ }) => (
-        enum Step {
-            $($step($ty),)+
-        }
-
-        enum Out {
-            $($step(<$ty as Future>::Item),)+
-        }
-
-        impl $fut {
-            fn either_poll(&mut self) -> Result<Async<Out>> {
-                match self.step {
-                    $(Step::$step(ref mut fut) => Ok(Ready(Out::$step(::futures::try_ready!(fut.poll())))),)+
-                }
-            }
-        }
-    );
-}
diff --git a/src/opts.rs b/src/opts.rs
index 90a14a98..122211e1 100644
--- a/src/opts.rs
+++ b/src/opts.rs
@@ -146,8 +146,6 @@ pub struct InnerOpts {
     stmt_cache_size: usize,
 
     /// Driver will require SSL connection if this option isn't `None` (default to `None`).
-    ///
-    /// This option requires `ssl` feature to work.
     ssl_opts: Option<SslOpts>,
 
     /// Prefer socket connection (defaults to `true`).
@@ -260,8 +258,6 @@ impl Opts {
     }
 
     /// Driver will require SSL connection if this option isn't `None` (default to `None`).
-    ///
-    /// This option requires `ssl` feature to work.
     pub fn get_ssl_opts(&self) -> Option<&SslOpts> {
         self.inner.ssl_opts.as_ref()
     }
@@ -494,8 +490,6 @@ impl OptsBuilder {
     }
 
     /// Driver will require SSL connection if this option isn't `None` (default to `None`).
-    ///
-    /// This option requires `ssl` feature to work.
     pub fn ssl_opts<T: Into<Option<SslOpts>>>(&mut self, ssl_opts: T) -> &mut Self {
         self.opts.ssl_opts = ssl_opts.into();
         self
diff --git a/src/queryable/mod.rs b/src/queryable/mod.rs
index f512cc3d..e70677ff 100644
--- a/src/queryable/mod.rs
+++ b/src/queryable/mod.rs
@@ -6,7 +6,6 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::future::Future;
 use mysql_common::{
     packets::{parse_ok_packet, RawPacket},
     row::new_row,
@@ -74,26 +73,30 @@ where
 {
     /// Returns future that resolves to `Conn` if `COM_PING` executed successfully.
     fn ping(self) -> BoxFuture<Self> {
-        let fut = self
-            .write_command_data(Command::COM_PING, &[])
-            .and_then(|this| this.read_packet())
-            .map(|(this, _)| this);
-        Box::new(fut)
+        Box::pin(async move {
+            Ok(self
+                .write_command_data(Command::COM_PING, &[])
+                .await?
+                .read_packet()
+                .await?
+                .0)
+        })
     }
 
     /// Returns future that disconnects this connection from a server.
     fn disconnect(mut self) -> BoxFuture<()> {
         self.on_disconnect();
-        let fut = self.write_command_data(Command::COM_QUIT, &[]).map(|_| ());
-        Box::new(fut)
+        let f = self.write_command_data(Command::COM_QUIT, &[]);
+        Box::pin(async move {
+            f.await?;
+            Ok(())
+        })
     }
 
     /// Returns future that performs `query`.
     fn query<Q: AsRef<str>>(self, query: Q) -> BoxFuture<QueryResult<Self, TextProtocol>> {
-        let fut = self
-            .write_command_data(Command::COM_QUERY, query.as_ref().as_bytes())
-            .and_then(|conn_like| conn_like.read_result_set(None));
-        Box::new(fut)
+        let f = self.write_command_data(Command::COM_QUERY, query.as_ref().as_bytes());
+        Box::pin(async move { f.await?.read_result_set(None).await })
     }
 
     /// Returns future that resolves to a first row of result of a `query` execution (if any).
@@ -104,33 +107,30 @@ where
         Q: AsRef<str>,
         R: FromRow,
     {
-        let fut = self
-            .query(query)
-            .and_then(|result| result.collect_and_drop::<Row>())
-            .map(|(this, mut rows)| {
-                if rows.len() > 1 {
-                    (this, Some(FromRow::from_row(rows.swap_remove(0))))
-                } else {
-                    (this, rows.pop().map(FromRow::from_row))
-                }
-            });
-        Box::new(fut)
+        let f = self.query(query);
+        Box::pin(async move {
+            let (this, mut rows) = f.await?.collect_and_drop::<Row>().await?;
+            if rows.len() > 1 {
+                Ok((this, Some(FromRow::from_row(rows.swap_remove(0)))))
+            } else {
+                Ok((this, rows.pop().map(FromRow::from_row)))
+            }
+        })
     }
 
     /// Returns future that performs query. Result will be dropped.
     fn drop_query<Q: AsRef<str>>(self, query: Q) -> BoxFuture<Self> {
-        let fut = self.query(query).and_then(|result| result.drop_result());
-        Box::new(fut)
+        let f = self.query(query);
+        Box::pin(async move { f.await?.drop_result().await })
     }
 
     /// Returns future that prepares statement.
     fn prepare<Q: AsRef<str>>(self, query: Q) -> BoxFuture<Stmt<Self>> {
-        let fut = self
-            .prepare_stmt(query)
-            .map(|(this, inner_stmt, stmt_cache_result)| {
-                stmt::new(this, inner_stmt, stmt_cache_result)
-            });
-        Box::new(fut)
+        let f = self.prepare_stmt(query);
+        Box::pin(async move {
+            let (this, inner_stmt, stmt_cache_result) = f.await?;
+            Ok(stmt::new(this, inner_stmt, stmt_cache_result))
+        })
     }
 
     /// Returns future that prepares and executes statement in one pass.
@@ -140,15 +140,13 @@ where
         P: Into<Params>,
     {
         let params: Params = params.into();
-        let fut = self
-            .prepare(query)
-            .and_then(|stmt| stmt.execute(params))
-            .map(|result| {
-                let (stmt, columns, _) = query_result::disassemble(result);
-                let (conn_like, cached) = stmt.unwrap();
-                query_result::assemble(conn_like, columns, cached)
-            });
-        Box::new(fut)
+        let f = self.prepare(query);
+        Box::pin(async move {
+            let result = f.await?.execute(params).await?;
+            let (stmt, columns, _) = query_result::disassemble(result);
+            let (conn_like, cached) = stmt.unwrap();
+            Ok(query_result::assemble(conn_like, columns, cached))
+        })
     }
 
     /// Returns future that resolves to a first row of result of a statement execution (if any).
@@ -160,17 +158,15 @@ where
         P: Into<Params>,
         R: FromRow,
     {
-        let fut = self
-            .prep_exec(query, params)
-            .and_then(|result| result.collect_and_drop::<Row>())
-            .map(|(this, mut rows)| {
-                if rows.len() > 1 {
-                    (this, Some(FromRow::from_row(rows.swap_remove(0))))
-                } else {
-                    (this, rows.pop().map(FromRow::from_row))
-                }
-            });
-        Box::new(fut)
+        let f = self.prep_exec(query, params);
+        Box::pin(async move {
+            let (this, mut rows) = f.await?.collect_and_drop::<Row>().await?;
+            if rows.len() > 1 {
+                Ok((this, Some(FromRow::from_row(rows.swap_remove(0)))))
+            } else {
+                Ok((this, rows.pop().map(FromRow::from_row)))
+            }
+        })
     }
 
     /// Returns future that prepares and executes statement. Result will be dropped.
@@ -179,10 +175,8 @@ where
         Q: AsRef<str>,
         P: Into<Params>,
     {
-        let fut = self
-            .prep_exec(query, params)
-            .and_then(|result| result.drop_result());
-        Box::new(fut)
+        let f = self.prep_exec(query, params);
+        Box::pin(async move { f.await?.drop_result().await })
     }
 
     /// Returns future that prepares statement and performs batch execution.
@@ -195,16 +189,13 @@ where
         Params: From<P>,
         P: Send + 'static,
     {
-        let fut = self
-            .prepare(query)
-            .and_then(|stmt| stmt.batch(params_iter))
-            .and_then(|stmt| stmt.close());
-        Box::new(fut)
+        let f = self.prepare(query);
+        Box::pin(async move { f.await?.batch(params_iter).await?.close().await })
     }
 
     /// Returns future that starts transaction.
     fn start_transaction(self, options: TransactionOptions) -> BoxFuture<Transaction<Self>> {
-        Box::new(transaction::new(self, options))
+        Box::pin(transaction::new(self, options))
     }
 }
 
diff --git a/src/queryable/query_result/for_each.rs b/src/queryable/query_result/for_each.rs
deleted file mode 100644
index 5a9be84c..00000000
--- a/src/queryable/query_result/for_each.rs
+++ /dev/null
@@ -1,61 +0,0 @@
-// Copyright (c) 2017 Anatoly Ikorsky
-//
-// Licensed under the Apache License, Version 2.0
-// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
-// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
-// option. All files in the project carrying such notice may not be copied,
-// modified, or distributed except according to those terms.
-
-use futures::{try_ready, Async::Ready, Future, Poll};
-
-use crate::{
-    connection_like::ConnectionLike,
-    error::*,
-    queryable::{query_result::QueryResult, Protocol},
-    BoxFuture, Row,
-};
-
-pub struct ForEach<T, P, F> {
-    fut: BoxFuture<(QueryResult<T, P>, Option<Row>)>,
-    fun: F,
-}
-
-impl<T, P, F> ForEach<T, P, F>
-where
-    F: FnMut(Row),
-    P: Send + Protocol + 'static,
-    T: ConnectionLike + Sized + 'static,
-{
-    pub fn new(query_result: QueryResult<T, P>, fun: F) -> ForEach<T, P, F> {
-        ForEach {
-            fut: Box::new(query_result.get_row()),
-            fun,
-        }
-    }
-}
-
-impl<T, P, F> Future for ForEach<T, P, F>
-where
-    F: FnMut(Row),
-    P: Protocol,
-    P: Send + 'static,
-    T: ConnectionLike + Sized + 'static,
-{
-    type Item = (QueryResult<T, P>);
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        loop {
-            let (query_result, row_opt) = try_ready!(self.fut.poll());
-            match row_opt {
-                Some(row) => {
-                    (self.fun)(row);
-                }
-                None => {
-                    return Ok(Ready(query_result));
-                }
-            }
-            self.fut = Box::new(query_result.get_row());
-        }
-    }
-}
diff --git a/src/queryable/query_result/map.rs b/src/queryable/query_result/map.rs
deleted file mode 100644
index 475b89d6..00000000
--- a/src/queryable/query_result/map.rs
+++ /dev/null
@@ -1,70 +0,0 @@
-// Copyright (c) 2017 Anatoly Ikorsky
-//
-// Licensed under the Apache License, Version 2.0
-// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
-// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
-// option. All files in the project carrying such notice may not be copied,
-// modified, or distributed except according to those terms.
-
-use futures::{try_ready, Async::Ready, Future, Poll};
-
-use std::mem;
-
-use crate::{
-    connection_like::ConnectionLike,
-    error::*,
-    queryable::{query_result::QueryResult, Protocol},
-    BoxFuture, Row,
-};
-
-pub struct Map<T, P, F, U> {
-    fut: BoxFuture<(QueryResult<T, P>, Option<Row>)>,
-    acc: Vec<U>,
-    fun: F,
-}
-
-impl<T, P, F, U> Map<T, P, F, U>
-where
-    F: FnMut(Row) -> U,
-    P: Protocol,
-    P: Send + 'static,
-    T: ConnectionLike + Sized + 'static,
-{
-    pub fn new(query_result: QueryResult<T, P>, fun: F) -> Map<T, P, F, U> {
-        Map {
-            fut: Box::new(query_result.get_row()),
-            acc: Vec::new(),
-            fun,
-        }
-    }
-}
-
-impl<T, P, F, U> Future for Map<T, P, F, U>
-where
-    F: FnMut(Row) -> U,
-    P: Protocol,
-    P: Send + 'static,
-    T: ConnectionLike + Sized + 'static,
-{
-    type Item = (QueryResult<T, P>, Vec<U>);
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        loop {
-            let (query_result, row_opt) = try_ready!(self.fut.poll());
-            match row_opt {
-                Some(row) => {
-                    let val = (self.fun)(row);
-                    self.acc.push(val);
-                }
-                None => {
-                    return Ok(Ready((
-                        query_result,
-                        mem::replace(&mut self.acc, Vec::new()),
-                    )));
-                }
-            }
-            self.fut = Box::new(query_result.get_row());
-        }
-    }
-}
diff --git a/src/queryable/query_result/mod.rs b/src/queryable/query_result/mod.rs
index bb1c73d7..9e418937 100644
--- a/src/queryable/query_result/mod.rs
+++ b/src/queryable/query_result/mod.rs
@@ -6,13 +6,7 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-pub use self::{for_each::ForEach, map::Map, reduce::Reduce};
-
-use futures::future::{
-    loop_fn, ok, AndThen,
-    Either::{self, *},
-    Future, FutureResult, Loop,
-};
+use futures_util::future::Either;
 use mysql_common::packets::RawPacket;
 use mysql_common::row::convert::FromRowError;
 
@@ -30,28 +24,9 @@ use crate::{
     io,
     prelude::FromRow,
     queryable::Protocol,
-    BoxFuture, Column, MyFuture, Row,
+    Column, Row,
 };
 
-mod for_each;
-mod map;
-mod reduce;
-
-pub type ForEachAndDrop<S, T, P, F> =
-    AndThen<Either<FutureResult<S, Error>, ForEach<T, P, F>>, BoxFuture<T>, fn(S) -> BoxFuture<T>>;
-
-pub type MapAndDrop<S, T, P, F, U> = AndThen<
-    Either<FutureResult<(S, Vec<U>), Error>, Map<T, P, F, U>>,
-    (BoxFuture<T>, FutureResult<Vec<U>, Error>),
-    fn((S, Vec<U>)) -> (BoxFuture<T>, FutureResult<Vec<U>, Error>),
->;
-
-pub type ReduceAndDrop<S, T, P, F, U> = AndThen<
-    Either<FutureResult<(S, U), Error>, Reduce<T, P, F, U>>,
-    (BoxFuture<T>, FutureResult<U, Error>),
-    fn((S, U)) -> (BoxFuture<T>, FutureResult<U, Error>),
->;
-
 pub fn new<T, P>(
     conn_like: T,
     columns: Option<Arc<Vec<Column>>>,
@@ -69,8 +44,8 @@ pub fn disassemble<T, P>(
     query_result: QueryResult<T, P>,
 ) -> (T, Option<Arc<Vec<Column>>>, Option<StmtCacheResult>) {
     match query_result {
-        QueryResult(Empty(Some(A(conn_like)), cached, _)) => (conn_like, None, cached),
-        QueryResult(WithRows(Some(A(conn_like)), columns, cached, _)) => {
+        QueryResult(Empty(Some(Either::Left(conn_like)), cached, _)) => (conn_like, None, cached),
+        QueryResult(WithRows(Some(Either::Left(conn_like)), columns, cached, _)) => {
             (conn_like, Some(columns), cached)
         }
         _ => unreachable!(),
@@ -87,8 +62,13 @@ where
     P: Protocol + 'static,
 {
     match columns {
-        Some(columns) => QueryResult(WithRows(Some(A(conn_like)), columns, cached, PhantomData)),
-        None => QueryResult(Empty(Some(A(conn_like)), cached, PhantomData)),
+        Some(columns) => QueryResult(WithRows(
+            Some(Either::Left(conn_like)),
+            columns,
+            cached,
+            PhantomData,
+        )),
+        None => QueryResult(Empty(Some(Either::Left(conn_like)), cached, PhantomData)),
     }
 }
 
@@ -130,48 +110,46 @@ where
         match self {
             QueryResult(Empty(conn_like, cached, _))
             | QueryResult(WithRows(conn_like, _, cached, _)) => match conn_like {
-                Some(A(conn_like)) => (conn_like, cached),
+                Some(Either::Left(conn_like)) => (conn_like, cached),
                 _ => unreachable!(),
             },
         }
     }
 
-    fn get_row_raw(self) -> impl MyFuture<(Self, Option<RawPacket>)> {
+    async fn get_row_raw(self) -> Result<(Self, Option<RawPacket>)> {
         if self.is_empty() {
-            return A(ok((self, None)));
+            return Ok((self, None));
         }
-        let fut = self.read_packet().and_then(|(this, packet)| {
-            if P::is_last_result_set_packet(&this, &packet) {
-                if this
-                    .get_status()
-                    .contains(StatusFlags::SERVER_MORE_RESULTS_EXISTS)
-                {
-                    let (inner, cached) = this.into_inner();
-                    A(A(inner
-                        .read_result_set(cached)
-                        .map(|new_this| (new_this, None))))
-                } else {
-                    A(B(ok((this.into_empty(), None))))
-                }
+
+        let (this, packet) = self.read_packet().await?;
+        if P::is_last_result_set_packet(&this, &packet) {
+            if this
+                .get_status()
+                .contains(StatusFlags::SERVER_MORE_RESULTS_EXISTS)
+            {
+                let (inner, cached) = this.into_inner();
+                let this = inner.read_result_set(cached).await?;
+                Ok((this, None))
             } else {
-                B(ok((this, Some(packet))))
+                Ok((this.into_empty(), None))
             }
-        });
-        B(fut)
+        } else {
+            Ok((this, Some(packet)))
+        }
     }
 
-    fn get_row(self) -> impl MyFuture<(Self, Option<Row>)> {
-        self.get_row_raw()
-            .and_then(|(this, packet_opt)| match packet_opt {
-                Some(packet) => match this {
-                    QueryResult(WithRows(_, ref columns, ..)) => {
-                        P::read_result_set_row(&packet, columns.clone())
-                    }
-                    _ => unreachable!(),
-                }
-                .map(|row| (this, Some(row))),
-                None => Ok((this, None)),
-            })
+    async fn get_row(self) -> Result<(Self, Option<Row>)> {
+        let (this, packet) = self.get_row_raw().await?;
+        if let Some(packet) = packet {
+            if let QueryResult(WithRows(_, ref columns, ..)) = this {
+                let row = P::read_result_set_row(&packet, columns.clone())?;
+                Ok((this, Some(row)))
+            } else {
+                unreachable!()
+            }
+        } else {
+            Ok((this, None))
+        }
     }
 
     fn new(
@@ -180,10 +158,13 @@ where
         cached: Option<StmtCacheResult>,
     ) -> QueryResult<T, P> {
         match columns {
-            Some(columns) => {
-                QueryResult(WithRows(Some(A(conn_like)), columns, cached, PhantomData))
-            }
-            None => QueryResult(Empty(Some(A(conn_like)), cached, PhantomData)),
+            Some(columns) => QueryResult(WithRows(
+                Some(Either::Left(conn_like)),
+                columns,
+                cached,
+                PhantomData,
+            )),
+            None => QueryResult(Empty(Some(Either::Left(conn_like)), cached, PhantomData)),
         }
     }
 
@@ -238,7 +219,7 @@ where
     /// It'll panic if any row isn't convertible to `R` (i.e. programmer error or unknown schema).
     /// * In case of programmer error see `FromRow` docs;
     /// * In case of unknown schema use [`QueryResult::try_collect`].
-    pub fn collect<R>(self) -> impl MyFuture<(Self, Vec<R>)>
+    pub async fn collect<R>(self) -> Result<(Self, Vec<R>)>
     where
         R: FromRow,
         R: Send + 'static,
@@ -247,13 +228,14 @@ where
             acc.push(FromRow::from_row(row));
             acc
         })
+        .await
     }
 
     /// Returns future that collects result set of this query.
     ///
     /// It works the same way as [`QueryResult::collect`] but won't panic
     /// if row isn't convertible to `R`.
-    pub fn try_collect<R>(self) -> impl MyFuture<(Self, Vec<StdResult<R, FromRowError>>)>
+    pub async fn try_collect<R>(self) -> Result<(Self, Vec<StdResult<R, FromRowError>>)>
     where
         R: FromRow,
         R: Send + 'static,
@@ -262,6 +244,7 @@ where
             acc.push(FromRow::from_row_opt(row));
             acc
         })
+        .await
     }
 
     /// Returns future that collects result set of a query result and drops everything else.
@@ -272,13 +255,14 @@ where
     /// It'll panic if any row isn't convertible to `R` (i.e. programmer error or unknown schema).
     /// * In case of programmer error see `FromRow` docs;
     /// * In case of unknown schema use [`QueryResult::try_collect`].
-    pub fn collect_and_drop<R>(self) -> impl MyFuture<(T, Vec<R>)>
+    pub async fn collect_and_drop<R>(self) -> Result<(T, Vec<R>)>
     where
         R: FromRow,
         R: Send + 'static,
     {
-        self.collect()
-            .and_then(|(this, output)| (this.drop_result(), ok(output)))
+        let (this, output) = self.collect().await?;
+        let conn = this.drop_result().await?;
+        Ok((conn, output))
     }
 
     /// Returns future that collects result set of a query result and drops everything else.
@@ -286,133 +270,140 @@ where
     ///
     /// It works the same way as [`QueryResult::collect_and_drop`] but won't panic
     /// if row isn't convertible to `R`.
-    pub fn try_collect_and_drop<R>(self) -> impl MyFuture<(T, Vec<StdResult<R, FromRowError>>)>
+    pub async fn try_collect_and_drop<R>(self) -> Result<(T, Vec<StdResult<R, FromRowError>>)>
     where
         R: FromRow,
         R: Send + 'static,
     {
-        self.try_collect()
-            .and_then(|(this, output)| (this.drop_result(), ok(output)))
+        let (this, output) = self.try_collect().await?;
+        let conn = this.drop_result().await?;
+        Ok((conn, output))
     }
 
     /// Returns future that will execute `fun` on every row of current result set.
     ///
     /// It will stop on result set boundary (see `QueryResult::collect` docs).
-    pub fn for_each<F>(self, fun: F) -> Either<FutureResult<Self, Error>, ForEach<T, P, F>>
+    pub async fn for_each<F>(self, mut fun: F) -> Result<Self>
     where
         F: FnMut(Row),
     {
         if self.is_empty() {
-            A(ok(self))
+            Ok(self)
         } else {
-            B(ForEach::new(self, fun))
+            let mut qr = self;
+            loop {
+                let (qr_, row) = qr.get_row().await?;
+                qr = qr_;
+                if let Some(row) = row {
+                    fun(row);
+                } else {
+                    break Ok(qr);
+                }
+            }
         }
     }
 
     /// Returns future that will execute `fun` on every row of current result set and drop
     /// everything else. It will resolve to a wrapped `Queryable`.
-    pub fn for_each_and_drop<F>(self, fun: F) -> ForEachAndDrop<Self, T, P, F>
+    pub async fn for_each_and_drop<F>(self, fun: F) -> Result<T>
     where
         F: FnMut(Row),
     {
-        self.for_each(fun)
-            .and_then(|x| Box::new(QueryResult::drop_result(x)))
+        self.for_each(fun).await?.drop_result().await
     }
 
     /// Returns future that will map every row of current result set to `U` using `fun`.
     ///
     /// It will stop on result set boundary (see `QueryResult::collect` docs).
-    pub fn map<F, U>(self, fun: F) -> Either<FutureResult<(Self, Vec<U>), Error>, Map<T, P, F, U>>
+    pub async fn map<F, U>(self, mut fun: F) -> Result<(Self, Vec<U>)>
     where
         F: FnMut(Row) -> U,
     {
         if self.is_empty() {
-            A(ok((self, Vec::new())))
+            Ok((self, Vec::new()))
         } else {
-            B(Map::new(self, fun))
+            let mut qr = self;
+            let mut rows = Vec::new();
+            loop {
+                let (qr_, row) = qr.get_row().await?;
+                qr = qr_;
+                if let Some(row) = row {
+                    rows.push(fun(row));
+                } else {
+                    break Ok((qr, rows));
+                }
+            }
         }
     }
 
     /// Returns future that will map every row of current result set to `U` using `fun` and drop
     /// everything else. It will resolve to a pair of wrapped `Queryable` and mapped result set.
-    pub fn map_and_drop<F, U>(self, fun: F) -> MapAndDrop<Self, T, P, F, U>
+    pub async fn map_and_drop<F, U>(self, fun: F) -> Result<(T, Vec<U>)>
     where
         F: FnMut(Row) -> U,
     {
-        fn join_drop<T, P, U>(
-            (this, output): (QueryResult<T, P>, Vec<U>),
-        ) -> (BoxFuture<T>, FutureResult<Vec<U>, Error>)
-        where
-            T: ConnectionLike + Sized + 'static,
-            P: Protocol + 'static,
-        {
-            (Box::new(QueryResult::drop_result(this)), ok(output))
-        }
-
-        self.map(fun).and_then(join_drop)
+        let (this, rows) = self.map(fun).await?;
+        let this = this.drop_result().await?;
+        Ok((this, rows))
     }
 
     /// Returns future that will reduce rows of current result set to `U` using `fun`.
     ///
     /// It will stop on result set boundary (see `QueryResult::collect` docs).
-    pub fn reduce<F, U>(
-        self,
-        init: U,
-        fun: F,
-    ) -> Either<FutureResult<(Self, U), Error>, Reduce<T, P, F, U>>
+    pub async fn reduce<F, U>(self, init: U, mut fun: F) -> Result<(Self, U)>
     where
         F: FnMut(U, Row) -> U,
     {
         if self.is_empty() {
-            A(ok((self, init)))
+            Ok((self, init))
         } else {
-            B(Reduce::new(self, init, fun))
+            let mut qr = self;
+            let mut acc = init;
+            loop {
+                let (qr_, row) = qr.get_row().await?;
+                qr = qr_;
+                if let Some(row) = row {
+                    acc = fun(acc, row);
+                } else {
+                    break Ok((qr, acc));
+                }
+            }
         }
     }
 
     /// Returns future that will reduce rows of current result set to `U` using `fun` and drop
     /// everything else. It will resolve to a pair of wrapped `Queryable` and `U`.
-    pub fn reduce_and_drop<F, U>(self, init: U, fun: F) -> ReduceAndDrop<Self, T, P, F, U>
+    pub async fn reduce_and_drop<F, U>(self, init: U, fun: F) -> Result<(T, U)>
     where
         F: FnMut(U, Row) -> U,
     {
-        fn join_drop<T, P, U>(
-            (this, output): (QueryResult<T, P>, U),
-        ) -> (BoxFuture<T>, FutureResult<U, Error>)
-        where
-            T: ConnectionLike + Sized + 'static,
-            P: Protocol + 'static,
-        {
-            (Box::new(QueryResult::drop_result(this)), ok(output))
-        }
-
-        self.reduce(init, fun).and_then(join_drop)
+        let (this, acc) = self.reduce(init, fun).await?;
+        let this = this.drop_result().await?;
+        Ok((this, acc))
     }
 
     /// Returns future that will drop this query result end resolve to a wrapped `Queryable`.
-    pub fn drop_result(self) -> impl MyFuture<T> {
-        let fut = loop_fn(self, |this| {
+    pub async fn drop_result(self) -> Result<T> {
+        let mut this = self;
+        let (conn_like, cached) = loop {
             if !this.has_rows() {
                 if this.more_results_exists() {
                     let (inner, cached) = this.into_inner();
-                    A(A(inner
-                        .read_result_set(cached)
-                        .map(|new_this| Loop::Continue(new_this))))
+                    this = inner.read_result_set(cached).await?;
                 } else {
-                    A(B(ok(Loop::Break(this.into_inner()))))
+                    break this.into_inner();
                 }
             } else {
-                B(this.get_row_raw().map(|(this, _)| Loop::Continue(this)))
+                let (this_, _) = this.get_row_raw().await?;
+                this = this_;
             }
-        });
+        };
 
-        fut.and_then(|(conn_like, cached)| {
-            if let Some(StmtCacheResult::NotCached(statement_id)) = cached {
-                A(conn_like.close_stmt(statement_id))
-            } else {
-                B(ok(conn_like))
-            }
-        })
+        if let Some(StmtCacheResult::NotCached(statement_id)) = cached {
+            conn_like.close_stmt(statement_id).await
+        } else {
+            Ok(conn_like)
+        }
     }
 
     /// Returns reference to columns in this query result.
@@ -444,30 +435,30 @@ impl<T: ConnectionLike + 'static, P: Protocol> ConnectionLikeWrapper for QueryRe
     {
         match self {
             QueryResult(Empty(conn_like, cached, _)) => match conn_like {
-                Some(A(conn_like)) => {
+                Some(Either::Left(conn_like)) => {
                     let (streamless, stream) = conn_like.take_stream();
                     let self_streamless = Streamless::new(QueryResult(Empty(
-                        Some(B(streamless)),
+                        Some(Either::Right(streamless)),
                         cached,
                         PhantomData,
                     )));
                     (self_streamless, stream)
                 }
-                Some(B(..)) => panic!("Logic error: stream taken"),
+                Some(Either::Right(..)) => panic!("Logic error: stream taken"),
                 None => unreachable!(),
             },
             QueryResult(WithRows(conn_like, columns, cached, _)) => match conn_like {
-                Some(A(conn_like)) => {
+                Some(Either::Left(conn_like)) => {
                     let (streamless, stream) = conn_like.take_stream();
                     let self_streamless = Streamless::new(QueryResult(WithRows(
-                        Some(B(streamless)),
+                        Some(Either::Right(streamless)),
                         columns,
                         cached,
                         PhantomData,
                     )));
                     (self_streamless, stream)
                 }
-                Some(B(..)) => panic!("Logic error: stream taken"),
+                Some(Either::Right(..)) => panic!("Logic error: stream taken"),
                 None => unreachable!(),
             },
         }
@@ -477,9 +468,9 @@ impl<T: ConnectionLike + 'static, P: Protocol> ConnectionLikeWrapper for QueryRe
         match *self {
             QueryResult(Empty(ref mut conn_like, ..))
             | QueryResult(WithRows(ref mut conn_like, ..)) => match conn_like.take() {
-                Some(A(..)) => panic!("Logic error: stream exists"),
-                Some(B(streamless)) => {
-                    *conn_like = Some(A(streamless.return_stream(stream)));
+                Some(Either::Left(..)) => panic!("Logic error: stream exists"),
+                Some(Either::Right(streamless)) => {
+                    *conn_like = Some(Either::Left(streamless.return_stream(stream)));
                 }
                 None => unreachable!(),
             },
@@ -490,7 +481,7 @@ impl<T: ConnectionLike + 'static, P: Protocol> ConnectionLikeWrapper for QueryRe
         match *self {
             QueryResult(Empty(ref conn_like, ..)) | QueryResult(WithRows(ref conn_like, ..)) => {
                 match *conn_like {
-                    Some(A(ref conn_like)) => conn_like,
+                    Some(Either::Left(ref conn_like)) => conn_like,
                     _ => unreachable!(),
                 }
             }
@@ -501,7 +492,7 @@ impl<T: ConnectionLike + 'static, P: Protocol> ConnectionLikeWrapper for QueryRe
         match *self {
             QueryResult(Empty(ref mut conn_like, ..))
             | QueryResult(WithRows(ref mut conn_like, ..)) => match *conn_like {
-                Some(A(ref mut conn_like)) => conn_like,
+                Some(Either::Left(ref mut conn_like)) => conn_like,
                 _ => unreachable!(),
             },
         }
diff --git a/src/queryable/query_result/reduce.rs b/src/queryable/query_result/reduce.rs
deleted file mode 100644
index 83c18ebb..00000000
--- a/src/queryable/query_result/reduce.rs
+++ /dev/null
@@ -1,64 +0,0 @@
-// Copyright (c) 2017 Anatoly Ikorsky
-//
-// Licensed under the Apache License, Version 2.0
-// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
-// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
-// option. All files in the project carrying such notice may not be copied,
-// modified, or distributed except according to those terms.
-
-use futures::{try_ready, Async::Ready, Future, Poll};
-
-use crate::{
-    connection_like::ConnectionLike,
-    error::*,
-    queryable::{query_result::QueryResult, Protocol},
-    BoxFuture, Row,
-};
-
-pub struct Reduce<T, P, F, U> {
-    fut: BoxFuture<(QueryResult<T, P>, Option<Row>)>,
-    acc: Option<U>,
-    fun: F,
-}
-
-impl<T, P, F, U> Reduce<T, P, F, U>
-where
-    F: FnMut(U, Row) -> U,
-    P: Protocol + 'static,
-    T: ConnectionLike + Sized + 'static,
-{
-    pub fn new(query_result: QueryResult<T, P>, init: U, fun: F) -> Reduce<T, P, F, U> {
-        Reduce {
-            fut: Box::new(query_result.get_row()),
-            acc: Some(init),
-            fun,
-        }
-    }
-}
-
-impl<T, P, F, U> Future for Reduce<T, P, F, U>
-where
-    F: FnMut(U, Row) -> U,
-    P: Protocol + 'static,
-    T: ConnectionLike + Sized + 'static,
-{
-    type Item = (QueryResult<T, P>, U);
-    type Error = Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        loop {
-            let (query_result, row_opt) = try_ready!(self.fut.poll());
-            match row_opt {
-                Some(row) => {
-                    let prev_acc_val = self.acc.take().unwrap();
-                    let new_acc_value = (self.fun)(prev_acc_val, row);
-                    self.acc = Some(new_acc_value);
-                }
-                None => {
-                    return Ok(Ready((query_result, self.acc.take().unwrap())));
-                }
-            }
-            self.fut = Box::new(query_result.get_row());
-        }
-    }
-}
diff --git a/src/queryable/stmt.rs b/src/queryable/stmt.rs
index 2183d333..ebb50ec1 100644
--- a/src/queryable/stmt.rs
+++ b/src/queryable/stmt.rs
@@ -8,11 +8,7 @@
 
 use bit_vec::BitVec;
 use byteorder::{LittleEndian as LE, ReadBytesExt, WriteBytesExt};
-use futures::future::{
-    err, loop_fn, ok,
-    Either::{self, *},
-    Future, IntoFuture, Loop,
-};
+use futures_util::future::Either;
 use mysql_common::value::serialize_bin_many;
 
 use std::io::Write;
@@ -26,7 +22,7 @@ use crate::{
     io,
     prelude::FromRow,
     queryable::{query_result::QueryResult, BinaryProtocol},
-    Column, MyFuture, Params, Row,
+    Column, Params, Row,
     Value::{self, *},
 };
 
@@ -86,18 +82,19 @@ where
 {
     fn new(conn_like: T, inner: InnerStmt, cached: StmtCacheResult) -> Stmt<T> {
         Stmt {
-            conn_like: Some(A(conn_like)),
+            conn_like: Some(Either::Left(conn_like)),
             inner,
             cached: Some(cached),
         }
     }
 
-    fn send_long_data_for_index(
+    async fn send_long_data_for_index(
         self,
         params: Vec<Value>,
         index: usize,
-    ) -> impl MyFuture<(Self, Vec<Value>)> {
-        loop_fn((self, params, index, 0), |(this, params, index, chunk)| {
+    ) -> Result<(Self, Vec<Value>)> {
+        let mut this = self;
+        for chunk in 0.. {
             let data_cap = crate::consts::MAX_PAYLOAD_LEN - 10;
             let buf = match params[index] {
                 Bytes(ref x) => {
@@ -119,42 +116,40 @@ where
             match buf {
                 Some(buf) => {
                     let chunk_len = buf.len() - 6;
-                    let fut = this
+                    this = this
                         .write_command_data(Command::COM_STMT_SEND_LONG_DATA, buf)
-                        .map(move |this| {
-                            if chunk_len < data_cap {
-                                Loop::Break((this, params))
-                            } else {
-                                Loop::Continue((this, params, index, chunk + 1))
-                            }
-                        });
-                    A(fut)
+                        .await?;
+                    if chunk_len < data_cap {
+                        break;
+                    }
                 }
-                None => B(ok(Loop::Break((this, params)))),
+                None => break,
             }
-        })
+        }
+        Ok((this, params))
     }
 
-    fn send_long_data(
+    async fn send_long_data(
         self,
-        params: Vec<Value>,
+        mut params: Vec<Value>,
         large_bitmap: BitVec<u8>,
-    ) -> impl MyFuture<(Self, Vec<Value>)> {
-        let bits = large_bitmap.into_iter().enumerate();
-
-        loop_fn(
-            (self, params, bits),
-            |(this, params, mut bits)| match bits.next() {
-                Some((index, true)) => A(this
-                    .send_long_data_for_index(params, index)
-                    .map(|(this, params)| Loop::Continue((this, params, bits)))),
-                Some((_, false)) => B(ok(Loop::Continue((this, params, bits)))),
-                None => B(ok(Loop::Break((this, params)))),
-            },
-        )
+    ) -> Result<(Self, Vec<Value>)> {
+        let mut bits = large_bitmap.into_iter().enumerate();
+        let mut this = self;
+        loop {
+            match bits.next() {
+                Some((index, true)) => {
+                    let (this_, params_) = this.send_long_data_for_index(params, index).await?;
+                    this = this_;
+                    params = params_;
+                }
+                Some((_, false)) => {}
+                None => return Ok((this, params)),
+            }
+        }
     }
 
-    fn execute_positional<U>(self, params: U) -> impl MyFuture<QueryResult<Self, BinaryProtocol>>
+    async fn execute_positional<U>(self, params: U) -> Result<QueryResult<Self, BinaryProtocol>>
     where
         U: ::std::ops::Deref<Target = [Value]>,
         U: IntoIterator<Item = Value>,
@@ -166,64 +161,58 @@ where
                 supplied: params.len() as u16,
             }
             .into();
-            return A(err(error));
+            return Err(error);
         }
 
-        let fut = self
-            .inner
-            .params
-            .as_ref()
-            .ok_or_else(|| unreachable!())
-            .and_then(|params_def| serialize_bin_many(&*params_def, &*params).map_err(Error::from))
-            .into_future()
-            .and_then(|bin_payload| match bin_payload {
-                (row_data, null_bitmap, large_bitmap) => self
-                    .send_long_data(params.into_iter().collect(), large_bitmap.clone())
-                    .and_then(|(this, params)| {
-                        let mut data = Vec::new();
-                        write_data(
-                            &mut data,
-                            this.inner.statement_id,
-                            row_data,
-                            params,
-                            this.inner.params.as_ref().unwrap(),
-                            null_bitmap,
-                        );
-                        this.write_command_data(Command::COM_STMT_EXECUTE, data)
-                    }),
-            })
-            .and_then(|this| this.read_result_set(None));
-        B(fut)
+        let params_def = self.inner.params.as_ref().unwrap();
+        let bin_payload = serialize_bin_many(&*params_def, &*params)?;
+        let (row_data, null_bitmap, large_bitmap) = bin_payload;
+        let (this, params) = self
+            .send_long_data(params.into_iter().collect(), large_bitmap.clone())
+            .await?;
+
+        let mut data = Vec::new();
+        write_data(
+            &mut data,
+            this.inner.statement_id,
+            row_data,
+            params,
+            this.inner.params.as_ref().unwrap(),
+            null_bitmap,
+        );
+        let this = this
+            .write_command_data(Command::COM_STMT_EXECUTE, data)
+            .await?;
+        let this = this.read_result_set(None).await?;
+        Ok(this)
     }
 
-    fn execute_named(self, params: Params) -> impl MyFuture<QueryResult<Self, BinaryProtocol>> {
+    async fn execute_named(self, params: Params) -> Result<QueryResult<Self, BinaryProtocol>> {
         if self.inner.named_params.is_none() {
             let error = DriverError::NamedParamsForPositionalQuery.into();
-            return A(err(error));
+            return Err(error);
         }
 
         let positional_params =
             match params.into_positional(self.inner.named_params.as_ref().unwrap()) {
                 Ok(positional_params) => positional_params,
-                Err(error) => {
-                    return A(err(error.into()));
-                }
+                Err(error) => return Err(error.into()),
             };
 
         match positional_params {
-            Params::Positional(params) => B(self.execute_positional(params)),
+            Params::Positional(params) => self.execute_positional(params).await,
             _ => unreachable!(),
         }
     }
 
-    fn execute_empty(self) -> impl MyFuture<QueryResult<Self, BinaryProtocol>> {
+    async fn execute_empty(self) -> Result<QueryResult<Self, BinaryProtocol>> {
         if self.inner.num_params > 0 {
             let error = DriverError::StmtParamsMismatch {
                 required: self.inner.num_params,
                 supplied: 0,
             }
             .into();
-            return A(err(error));
+            return Err(error);
         }
 
         let mut data = Vec::with_capacity(4 + 1 + 4);
@@ -231,72 +220,69 @@ where
         data.write_u8(0u8).unwrap();
         data.write_u32::<LE>(1u32).unwrap();
 
-        B(self
+        let this = self
             .write_command_data(Command::COM_STMT_EXECUTE, data)
-            .and_then(|this| this.read_result_set(None)))
+            .await?;
+        this.read_result_set(None).await
     }
 
     /// See `Queryable::execute`
-    pub fn execute<P>(self, params: P) -> impl MyFuture<QueryResult<Self, BinaryProtocol>>
+    pub async fn execute<P>(self, params: P) -> Result<QueryResult<Self, BinaryProtocol>>
     where
         P: Into<Params>,
     {
         let params = params.into();
         match params {
-            Params::Positional(params) => A(self.execute_positional(params)),
-            Params::Named(_) => B(A(self.execute_named(params))),
-            Params::Empty => B(B(self.execute_empty())),
+            Params::Positional(params) => self.execute_positional(params).await,
+            Params::Named(_) => self.execute_named(params).await,
+            Params::Empty => self.execute_empty().await,
         }
     }
 
     /// See `Queryable::first`
-    pub fn first<P, R>(self, params: P) -> impl MyFuture<(Self, Option<R>)>
+    pub async fn first<P, R>(self, params: P) -> Result<(Self, Option<R>)>
     where
         P: Into<Params> + 'static,
         R: FromRow,
     {
-        self.execute(params)
-            .and_then(|result| result.collect_and_drop::<Row>())
-            .map(|(this, mut rows)| {
-                if rows.len() > 1 {
-                    (this, Some(FromRow::from_row(rows.swap_remove(0))))
-                } else {
-                    (this, rows.pop().map(FromRow::from_row))
-                }
-            })
+        let result = self.execute(params).await?;
+        let (this, mut rows) = result.collect_and_drop::<Row>().await?;
+        if rows.len() > 1 {
+            Ok((this, Some(FromRow::from_row(rows.swap_remove(0)))))
+        } else {
+            Ok((this, rows.pop().map(FromRow::from_row)))
+        }
     }
 
     /// See `Queryable::batch`
-    pub fn batch<I, P>(self, params_iter: I) -> impl MyFuture<Self>
+    pub async fn batch<I, P>(self, params_iter: I) -> Result<Self>
     where
         I: IntoIterator<Item = P>,
         I::IntoIter: Send + 'static,
         Params: From<P>,
         P: 'static,
     {
-        let params_iter = params_iter.into_iter().map(Params::from);
-
-        loop_fn(
-            (self, params_iter),
-            |(this, mut params_iter)| match params_iter.next() {
-                Some(params) => A(this
-                    .execute(params)
-                    .and_then(|result| result.drop_result())
-                    .map(|this| Loop::Continue((this, params_iter)))),
-                None => B(ok(Loop::Break(this))),
-            },
-        )
+        let mut params_iter = params_iter.into_iter().map(Params::from);
+        let mut this = self;
+        loop {
+            match params_iter.next() {
+                Some(params) => {
+                    this = this.execute(params).await?.drop_result().await?;
+                }
+                None => break Ok(this),
+            }
+        }
     }
 
     /// This will close statement (if it's not in the cache) and resolve to a wrapped queryable.
-    pub fn close(mut self) -> impl MyFuture<T> {
+    pub async fn close(mut self) -> Result<T> {
         let cached = self.cached.take();
         match self.conn_like {
-            Some(A(conn_like)) => {
+            Some(Either::Left(conn_like)) => {
                 if let Some(StmtCacheResult::NotCached(stmt_id)) = cached {
-                    A(conn_like.close_stmt(stmt_id))
+                    conn_like.close_stmt(stmt_id).await
                 } else {
-                    B(ok(conn_like))
+                    Ok(conn_like)
                 }
             }
             _ => unreachable!(),
@@ -305,7 +291,7 @@ where
 
     pub(crate) fn unwrap(mut self) -> (T, Option<StmtCacheResult>) {
         match self.conn_like {
-            Some(A(conn_like)) => (conn_like, self.cached.take()),
+            Some(Either::Left(conn_like)) => (conn_like, self.cached.take()),
             _ => unreachable!(),
         }
     }
@@ -324,10 +310,10 @@ impl<T: ConnectionLike + 'static> ConnectionLikeWrapper for Stmt<T> {
             cached,
         } = self;
         match conn_like {
-            Some(A(conn_like)) => {
+            Some(Either::Left(conn_like)) => {
                 let (streamless, stream) = conn_like.take_stream();
                 let this = Stmt {
-                    conn_like: Some(B(streamless)),
+                    conn_like: Some(Either::Right(streamless)),
                     inner,
                     cached,
                 };
@@ -340,8 +326,8 @@ impl<T: ConnectionLike + 'static> ConnectionLikeWrapper for Stmt<T> {
     fn return_stream(&mut self, stream: io::Stream) {
         let conn_like = self.conn_like.take().unwrap();
         match conn_like {
-            B(streamless) => {
-                self.conn_like = Some(A(streamless.return_stream(stream)));
+            Either::Right(streamless) => {
+                self.conn_like = Some(Either::Left(streamless.return_stream(stream)));
             }
             _ => unreachable!(),
         }
@@ -349,14 +335,14 @@ impl<T: ConnectionLike + 'static> ConnectionLikeWrapper for Stmt<T> {
 
     fn conn_like_ref(&self) -> &Self::ConnLike {
         match self.conn_like {
-            Some(A(ref conn_like)) => conn_like,
+            Some(Either::Left(ref conn_like)) => conn_like,
             _ => unreachable!(),
         }
     }
 
     fn conn_like_mut(&mut self) -> &mut Self::ConnLike {
         match self.conn_like {
-            Some(A(ref mut conn_like)) => conn_like,
+            Some(Either::Left(ref mut conn_like)) => conn_like,
             _ => unreachable!(),
         }
     }
diff --git a/src/queryable/transaction.rs b/src/queryable/transaction.rs
index 595ee952..d09c5806 100644
--- a/src/queryable/transaction.rs
+++ b/src/queryable/transaction.rs
@@ -6,11 +6,7 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::future::{
-    err, ok,
-    Either::{self, *},
-    Future, IntoFuture,
-};
+use futures_util::future::Either;
 
 use std::fmt;
 
@@ -19,7 +15,6 @@ use crate::{
     error::*,
     io,
     queryable::Queryable,
-    MyFuture,
 };
 
 /// Options for transaction
@@ -96,15 +91,15 @@ impl fmt::Display for IsolationLevel {
 /// `NestedTransaction` error if you call `transaction.start_transaction(_)`.
 pub struct Transaction<T>(Option<Either<T, Streamless<T>>>);
 
-pub fn new<T>(conn_like: T, options: TransactionOptions) -> impl MyFuture<Transaction<T>>
+pub async fn new<T>(conn_like: T, options: TransactionOptions) -> Result<Transaction<T>>
 where
     T: Queryable + ConnectionLike,
 {
-    Transaction::new(conn_like, options)
+    Transaction::new(conn_like, options).await
 }
 
 impl<T: Queryable + ConnectionLike> Transaction<T> {
-    fn new(conn_like: T, options: TransactionOptions) -> impl MyFuture<Transaction<T>> {
+    async fn new(mut conn_like: T, options: TransactionOptions) -> Result<Transaction<T>> {
         let TransactionOptions {
             consistent_snapshot,
             isolation_level,
@@ -112,71 +107,57 @@ impl<T: Queryable + ConnectionLike> Transaction<T> {
         } = options;
 
         if conn_like.get_in_transaction() {
-            return A(err(DriverError::NestedTransaction.into()));
+            return Err(DriverError::NestedTransaction.into());
         }
 
         if readonly.is_some() && conn_like.get_server_version() < (5, 6, 5) {
-            return A(err(DriverError::ReadOnlyTransNotSupported.into()));
+            return Err(DriverError::ReadOnlyTransNotSupported.into());
+        }
+
+        if let Some(isolation_level) = isolation_level {
+            let query = format!("SET TRANSACTION ISOLATION LEVEL {}", isolation_level);
+            conn_like = conn_like.drop_query(query).await?;
+        }
+
+        if let Some(readonly) = readonly {
+            if readonly {
+                conn_like = conn_like.drop_query("SET TRANSACTION READ ONLY").await?;
+            } else {
+                conn_like = conn_like.drop_query("SET TRANSACTION READ WRITE").await?;
+            }
         }
 
-        let fut = if let Some(isolation_level) = isolation_level {
-            A(conn_like.drop_query(format!(
-                "SET TRANSACTION ISOLATION LEVEL {}",
-                isolation_level
-            )))
+        conn_like = if consistent_snapshot {
+            conn_like
+                .drop_query("START TRANSACTION WITH CONSISTENT SNAPSHOT")
+                .await?
         } else {
-            B(ok(conn_like))
+            conn_like.drop_query("START TRANSACTION").await?
         };
 
-        let fut = fut
-            .into_future()
-            .and_then(move |conn_like| {
-                if let Some(readonly) = readonly {
-                    if readonly {
-                        A(conn_like.drop_query("SET TRANSACTION READ ONLY"))
-                    } else {
-                        A(conn_like.drop_query("SET TRANSACTION READ WRITE"))
-                    }
-                } else {
-                    B(ok(conn_like))
-                }
-            })
-            .and_then(move |conn_like| {
-                if consistent_snapshot {
-                    conn_like.drop_query("START TRANSACTION WITH CONSISTENT SNAPSHOT")
-                } else {
-                    conn_like.drop_query("START TRANSACTION")
-                }
-            })
-            .map(|mut conn_like| {
-                conn_like.set_in_transaction(true);
-                Transaction(Some(A(conn_like)))
-            });
-
-        B(fut)
+        conn_like.set_in_transaction(true);
+        Ok(Transaction(Some(Either::Left(conn_like))))
     }
 
     fn unwrap(self) -> T {
         match self {
-            Transaction(Some(A(conn_like))) => conn_like,
+            Transaction(Some(Either::Left(conn_like))) => conn_like,
             _ => unreachable!(),
         }
     }
 
     /// Returns future that will perform `COMMIT` query and resolve to a wrapped `Queryable`.
-    pub fn commit(self) -> impl MyFuture<T> {
-        self.drop_query("COMMIT").map(|mut this| {
-            this.set_in_transaction(false);
-            this.unwrap()
-        })
+    pub async fn commit(self) -> Result<T> {
+        let mut this = self.drop_query("COMMIT").await?;
+        this.set_in_transaction(false);
+        Ok(this.unwrap())
     }
 
     /// Returns future that will perform `ROLLBACK` query and resolve to a wrapped `Queryable`.
-    pub fn rollback(self) -> impl MyFuture<T> {
-        self.drop_query("ROLLBACK").map(|mut this| {
-            this.set_in_transaction(false);
-            this.unwrap()
-        })
+    pub async fn rollback(self) -> Result<T> {
+        let mut this = self.drop_query("ROLLBACK").await?;
+        this.set_in_transaction(false);
+        Ok(this.unwrap())
     }
 }
 
@@ -189,9 +170,9 @@ impl<T: ConnectionLike + 'static> ConnectionLikeWrapper for Transaction<T> {
     {
         let Transaction(conn_like) = self;
         match conn_like {
-            Some(A(conn_like)) => {
+            Some(Either::Left(conn_like)) => {
                 let (streamless, stream) = conn_like.take_stream();
-                let this = Transaction(Some(B(streamless)));
+                let this = Transaction(Some(Either::Right(streamless)));
                 (Streamless::new(this), stream)
             }
             _ => unreachable!(),
@@ -201,8 +182,8 @@ impl<T: ConnectionLike + 'static> ConnectionLikeWrapper for Transaction<T> {
     fn return_stream(&mut self, stream: io::Stream) {
         let conn_like = self.0.take().unwrap();
         match conn_like {
-            B(streamless) => {
-                self.0 = Some(A(streamless.return_stream(stream)));
+            Either::Right(streamless) => {
+                self.0 = Some(Either::Left(streamless.return_stream(stream)));
             }
             _ => unreachable!(),
         }
@@ -210,14 +191,14 @@ impl<T: ConnectionLike + 'static> ConnectionLikeWrapper for Transaction<T> {
 
     fn conn_like_ref(&self) -> &Self::ConnLike {
         match self.0 {
-            Some(A(ref conn_like)) => conn_like,
+            Some(Either::Left(ref conn_like)) => conn_like,
             _ => unreachable!(),
         }
     }
 
     fn conn_like_mut(&mut self) -> &mut Self::ConnLike {
         match self.0 {
-            Some(A(ref mut conn_like)) => conn_like,
+            Some(Either::Left(ref mut conn_like)) => conn_like,
             _ => unreachable!(),
         }
     }


Test patch:
diff --git a/tests/exports.rs b/tests/exports.rs
index 456928e1..d89899aa 100644
--- a/tests/exports.rs
+++ b/tests/exports.rs
@@ -1,16 +1,14 @@
 #[allow(unused_imports)]
 use mysql_async::{
     chrono, consts, error, from_row, from_row_opt, from_value, from_value_opt,
-    futures::{
-        DisconnectPool, ForEach, ForEachAndDrop, GetConn, Map, MapAndDrop, Reduce, ReduceAndDrop,
-    },
+    futures::{DisconnectPool, GetConn},
     params,
     prelude::{
         ConnectionLike, ConvIr, FromRow, FromValue, LocalInfileHandler, Protocol, Queryable,
         ToValue,
     },
     time, uuid, BinaryProtocol, BoxFuture, Column, Conn, Deserialized, FromRowError,
-    FromValueError, IsolationLevel, MyFuture, Opts, OptsBuilder, Params, Pool, PoolConstraints,
-    QueryResult, Row, Serialized, SslOpts, Stmt, TextProtocol, Transaction, TransactionOptions,
-    Value, WhiteListFsLocalInfileHandler,
+    FromValueError, IsolationLevel, Opts, OptsBuilder, Params, Pool, PoolConstraints, QueryResult,
+    Row, Serialized, SslOpts, Stmt, TextProtocol, Transaction, TransactionOptions, Value,
+    WhiteListFsLocalInfileHandler,
 };
diff --git a/tests/generic.rs b/tests/generic.rs
index 57e8cdf0..0ee875f3 100644
--- a/tests/generic.rs
+++ b/tests/generic.rs
@@ -6,28 +6,13 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use ::futures::Future;
-use mysql_async::error::Error;
+use mysql_async::error::*;
 use mysql_async::prelude::*;
-use mysql_async::{MyFuture, Opts, Pool, QueryResult};
+use mysql_async::{Opts, Pool, QueryResult};
 
 use std::env;
 use std::io;
 
-/// Same as `tokio::run`, but will panic if future panics and will return the result
-/// of future execution.
-fn run<F, T, U>(future: F) -> Result<T, U>
-where
-    F: Future<Item = T, Error = U> + Send + 'static,
-    T: Send + 'static,
-    U: Send + 'static,
-{
-    let mut runtime = tokio::runtime::Runtime::new().unwrap();
-    let result = runtime.block_on(future);
-    runtime.shutdown_on_idle().wait().unwrap();
-    result
-}
-
 fn get_url() -> String {
     if let Ok(url) = env::var("DATABASE_URL") {
         let opts = Opts::from_url(&url).expect("DATABASE_URL invalid");
@@ -44,39 +29,37 @@ fn get_url() -> String {
     }
 }
 
-pub fn get_all_results<TupleType, T, P>(result: QueryResult<T, P>) -> impl MyFuture<Vec<TupleType>>
+pub async fn get_all_results<TupleType, T, P>(result: QueryResult<T, P>) -> Result<Vec<TupleType>>
 where
     TupleType: FromRow + Send + 'static,
     P: Protocol + Send + 'static,
     T: ConnectionLike + Sized + Send + 'static,
 {
-    result.collect().map(|(_, data)| data)
+    Ok(result.collect().await?.1)
 }
 
-pub fn get_single_result<TupleType, T, P>(result: QueryResult<T, P>) -> impl MyFuture<TupleType>
+pub async fn get_single_result<TupleType, T, P>(result: QueryResult<T, P>) -> Result<TupleType>
 where
     TupleType: FromRow + Send + 'static,
     P: Protocol + Send + 'static,
     T: ConnectionLike + Sized + Send + 'static,
 {
-    get_all_results(result).and_then(|mut data| {
-        if data.len() != 1 {
-            Err(Error::from(io::Error::from(io::ErrorKind::InvalidData)))
-        } else {
-            Ok(data.remove(0))
-        }
-    })
+    let mut data = get_all_results(result).await?;
+    if data.len() != 1 {
+        Err(Error::from(io::Error::from(io::ErrorKind::InvalidData)))
+    } else {
+        Ok(data.remove(0))
+    }
 }
 
-#[test]
-fn use_generic_code() {
+#[tokio::test]
+async fn use_generic_code() {
     let pool = Pool::new(Opts::from_url(&*get_url()).unwrap());
-    let fut = pool
-        .get_conn()
-        .and_then(move |conn| conn.query("SELECT 1, 2, 3"))
-        .and_then(get_single_result::<(u8, u8, u8), _, _>)
-        .and_then(|out| pool.disconnect().map(move |_| out));
-
-    let result = run(fut).unwrap();
+    let conn = pool.get_conn().await.unwrap();
+    let result = conn.query("SELECT 1, 2, 3").await.unwrap();
+    let result = get_single_result::<(u8, u8, u8), _, _>(result)
+        .await
+        .unwrap();
+    pool.disconnect().await.unwrap();
     assert_eq!(result, (1, 2, 3));
 }