diff --git a/postgres/Cargo.toml b/postgres/Cargo.toml
index 5200f788..3e24504f 100644
--- a/postgres/Cargo.toml
+++ b/postgres/Cargo.toml
@@ -30,9 +30,14 @@ runtime = ["tokio-postgres/runtime", "tokio", "lazy_static", "log"]
 [dependencies]
 bytes = "0.4"
 fallible-iterator = "0.2"
-futures = "0.1"
+futures-preview = "0.3.0-alpha.17"
+pin-utils = "0.1.0-alpha.4"
 tokio-postgres = { version = "0.4.0-rc.2", path = "../tokio-postgres", default-features = false }
+tokio-executor = { git = "https://github.com/tokio-rs/tokio" }
 
-tokio = { version = "0.1", optional = true }
+tokio = { git = "https://github.com/tokio-rs/tokio", optional = true }
 lazy_static = { version = "1.0", optional = true }
 log = { version = "0.4", optional = true }
+
+[dev-dependencies]
+tokio = { git = "https://github.com/tokio-rs/tokio" }
diff --git a/postgres/src/client.rs b/postgres/src/client.rs
index c811e74d..e4a1e382 100644
--- a/postgres/src/client.rs
+++ b/postgres/src/client.rs
@@ -1,15 +1,18 @@
 use fallible_iterator::FallibleIterator;
-use futures::{Async, Future, Poll, Stream};
-use std::io::{self, Read};
+use futures::executor;
+use std::io::{BufRead, Read};
 use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
 use tokio_postgres::types::{ToSql, Type};
 #[cfg(feature = "runtime")]
 use tokio_postgres::Socket;
 use tokio_postgres::{Error, Row, SimpleQueryMessage};
 
+use crate::copy_in_stream::CopyInStream;
+use crate::copy_out_reader::CopyOutReader;
+use crate::iter::Iter;
 #[cfg(feature = "runtime")]
 use crate::Config;
-use crate::{CopyOutReader, QueryIter, SimpleQueryIter, Statement, ToStatement, Transaction};
+use crate::{Statement, ToStatement, Transaction};
 
 /// A synchronous PostgreSQL client.
 ///
@@ -82,7 +85,7 @@ impl Client {
         T: ?Sized + ToStatement,
     {
         let statement = query.__statement(self)?;
-        self.0.execute(&statement, params).wait()
+        executor::block_on(self.0.execute(&statement, params))
     }
 
     /// Executes a statement, returning the resulting rows.
@@ -149,16 +152,16 @@ impl Client {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn query_iter<T>(
-        &mut self,
+    pub fn query_iter<'a, T>(
+        &'a mut self,
         query: &T,
         params: &[&dyn ToSql],
-    ) -> Result<QueryIter<'_>, Error>
+    ) -> Result<impl FallibleIterator<Item = Row, Error = Error> + 'a, Error>
     where
         T: ?Sized + ToStatement,
     {
         let statement = query.__statement(self)?;
-        Ok(QueryIter::new(self.0.query(&statement, params)))
+        Ok(Iter::new(self.0.query(&statement, params)))
     }
 
     /// Creates a new prepared statement.
@@ -185,7 +188,7 @@ impl Client {
     /// # }
     /// ```
     pub fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
-        self.0.prepare(query).wait()
+        executor::block_on(self.0.prepare(query))
     }
 
     /// Like `prepare`, but allows the types of query parameters to be explicitly specified.
@@ -216,7 +219,7 @@ impl Client {
     /// # }
     /// ```
     pub fn prepare_typed(&mut self, query: &str, types: &[Type]) -> Result<Statement, Error> {
-        self.0.prepare_typed(query, types).wait()
+        executor::block_on(self.0.prepare_typed(query, types))
     }
 
     /// Executes a `COPY FROM STDIN` statement, returning the number of rows created.
@@ -244,12 +247,10 @@ impl Client {
     ) -> Result<u64, Error>
     where
         T: ?Sized + ToStatement,
-        R: Read,
+        R: Read + Unpin,
     {
         let statement = query.__statement(self)?;
-        self.0
-            .copy_in(&statement, params, CopyInStream(reader))
-            .wait()
+        executor::block_on(self.0.copy_in(&statement, params, CopyInStream(reader)))
     }
 
     /// Executes a `COPY TO STDOUT` statement, returning a reader of the resulting data.
@@ -271,11 +272,11 @@ impl Client {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn copy_out<T>(
-        &mut self,
+    pub fn copy_out<'a, T>(
+        &'a mut self,
         query: &T,
         params: &[&dyn ToSql],
-    ) -> Result<CopyOutReader<'_>, Error>
+    ) -> Result<impl BufRead + 'a, Error>
     where
         T: ?Sized + ToStatement,
     {
@@ -311,8 +312,11 @@ impl Client {
     /// Prepared statements should be use for any query which contains user-specified data, as they provided the
     /// functionality to safely imbed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
-    pub fn simple_query_iter(&mut self, query: &str) -> Result<SimpleQueryIter<'_>, Error> {
-        Ok(SimpleQueryIter::new(self.0.simple_query(query)))
+    pub fn simple_query_iter<'a>(
+        &'a mut self,
+        query: &str,
+    ) -> Result<impl FallibleIterator<Item = SimpleQueryMessage, Error = Error> + 'a, Error> {
+        Ok(Iter::new(self.0.simple_query(query)))
     }
 
     /// Begins a new database transaction.
@@ -336,8 +340,8 @@ impl Client {
     /// # }
     /// ```
     pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
-        self.simple_query("BEGIN")?;
-        Ok(Transaction::new(self))
+        let transaction = executor::block_on(self.0.transaction())?;
+        Ok(Transaction::new(transaction))
     }
 
     /// Determines if the client's connection has already closed.
@@ -368,21 +372,3 @@ impl From<tokio_postgres::Client> for Client {
         Client(c)
     }
 }
-
-struct CopyInStream<R>(R);
-
-impl<R> Stream for CopyInStream<R>
-where
-    R: Read,
-{
-    type Item = Vec<u8>;
-    type Error = io::Error;
-
-    fn poll(&mut self) -> Poll<Option<Vec<u8>>, io::Error> {
-        let mut buf = vec![];
-        match self.0.by_ref().take(4096).read_to_end(&mut buf)? {
-            0 => Ok(Async::Ready(None)),
-            _ => Ok(Async::Ready(Some(buf))),
-        }
-    }
-}
diff --git a/postgres/src/config.rs b/postgres/src/config.rs
index b04c03f8..2c2fa655 100644
--- a/postgres/src/config.rs
+++ b/postgres/src/config.rs
@@ -2,25 +2,22 @@
 //!
 //! Requires the `runtime` Cargo feature (enabled by default).
 
-use futures::future::Executor;
-use futures::sync::oneshot;
-use futures::Future;
+use futures::FutureExt;
 use log::error;
 use std::fmt;
 use std::path::Path;
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{mpsc, Arc, Mutex};
 use std::time::Duration;
+use tokio_executor::Executor;
 use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
 use tokio_postgres::{Error, Socket};
 
 #[doc(inline)]
-use tokio_postgres::config::{SslMode, TargetSessionAttrs};
+pub use tokio_postgres::config::{SslMode, TargetSessionAttrs};
 
 use crate::{Client, RUNTIME};
 
-type DynExecutor = dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Sync + Send;
-
 /// Connection configuration.
 ///
 /// Configuration can be parsed from libpq-style connection strings. These strings come in two formats:
@@ -98,7 +95,7 @@ type DynExecutor = dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>> +
 pub struct Config {
     config: tokio_postgres::Config,
     // this is an option since we don't want to boot up our default runtime unless we're actually going to use it.
-    executor: Option<Arc<DynExecutor>>,
+    executor: Option<Arc<Mutex<dyn Executor + Sync + Send>>>,
 }
 
 impl fmt::Debug for Config {
@@ -242,45 +239,53 @@ impl Config {
     /// Defaults to a postgres-specific tokio `Runtime`.
     pub fn executor<E>(&mut self, executor: E) -> &mut Config
     where
-        E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + 'static + Sync + Send,
+        E: Executor + 'static + Sync + Send,
     {
-        self.executor = Some(Arc::new(executor));
+        self.executor = Some(Arc::new(Mutex::new(executor)));
         self
     }
 
     /// Opens a connection to a PostgreSQL database.
-    pub fn connect<T>(&self, tls_mode: T) -> Result<Client, Error>
+    pub fn connect<T>(&self, tls: T) -> Result<Client, Error>
     where
         T: MakeTlsConnect<Socket> + 'static + Send,
         T::TlsConnect: Send,
         T::Stream: Send,
         <T::TlsConnect as TlsConnect<Socket>>::Future: Send,
     {
-        let (tx, rx) = oneshot::channel();
-        let connect = self
-            .config
-            .connect(tls_mode)
-            .then(|r| tx.send(r).map_err(|_| ()));
-        self.with_executor(|e| e.execute(Box::new(connect)))
-            .unwrap();
-        let (client, connection) = rx.wait().unwrap()?;
+        let (client, connection) = match &self.executor {
+            Some(executor) => {
+                let (tx, rx) = mpsc::channel();
+                let config = self.config.clone();
+                let connect = async move {
+                    let r = config.connect(tls).await;
+                    let _ = tx.send(r);
+                };
+                executor.lock().unwrap().spawn(Box::pin(connect)).unwrap();
+                rx.recv().unwrap()?
+            }
+            None => {
+                let connect = self.config.connect(tls);
+                RUNTIME.block_on(connect)?
+            }
+        };
 
-        let connection = connection.map_err(|e| error!("postgres connection error: {}", e));
-        self.with_executor(|e| e.execute(Box::new(connection)))
-            .unwrap();
+        let connection = connection.map(|r| {
+            if let Err(e) = r {
+                error!("postgres connection error: {}", e)
+            }
+        });
+        match &self.executor {
+            Some(executor) => {
+                executor.lock().unwrap().spawn(Box::pin(connection)).unwrap();
+            },
+            None => {
+                RUNTIME.spawn(connection);
+            }
+        }
 
         Ok(Client::from(client))
     }
-
-    fn with_executor<F, T>(&self, f: F) -> T
-    where
-        F: FnOnce(&dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>>) -> T,
-    {
-        match &self.executor {
-            Some(e) => f(&**e),
-            None => f(&RUNTIME.executor()),
-        }
-    }
 }
 
 impl FromStr for Config {
diff --git a/postgres/src/copy_in_stream.rs b/postgres/src/copy_in_stream.rs
new file mode 100644
index 00000000..0dc2f0bb
--- /dev/null
+++ b/postgres/src/copy_in_stream.rs
@@ -0,0 +1,25 @@
+use futures::Stream;
+use std::io;
+use std::io::Read;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pub struct CopyInStream<R>(pub R);
+
+impl<R> Stream for CopyInStream<R>
+where
+    R: Read + Unpin,
+{
+    type Item = io::Result<Vec<u8>>;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        _: &mut Context<'_>,
+    ) -> Poll<Option<io::Result<Vec<u8>>>> {
+        let mut buf = vec![];
+        match self.0.by_ref().take(4096).read_to_end(&mut buf)? {
+            0 => Poll::Ready(None),
+            _ => Poll::Ready(Some(Ok(buf))),
+        }
+    }
+}
diff --git a/postgres/src/copy_out_reader.rs b/postgres/src/copy_out_reader.rs
index ff0c30a2..680d4d31 100644
--- a/postgres/src/copy_out_reader.rs
+++ b/postgres/src/copy_out_reader.rs
@@ -1,26 +1,34 @@
 use bytes::{Buf, Bytes};
-use futures::stream::{self, Stream};
+use futures::{executor, Stream};
 use std::io::{self, BufRead, Cursor, Read};
 use std::marker::PhantomData;
-use tokio_postgres::impls;
+use std::pin::Pin;
 use tokio_postgres::Error;
 
 /// The reader returned by the `copy_out` method.
-pub struct CopyOutReader<'a> {
-    it: stream::Wait<impls::CopyOut>,
+pub struct CopyOutReader<'a, S>
+where
+    S: Stream,
+{
+    it: executor::BlockingStream<Pin<Box<S>>>,
     cur: Cursor<Bytes>,
     _p: PhantomData<&'a mut ()>,
 }
 
 // no-op impl to extend borrow until drop
-impl<'a> Drop for CopyOutReader<'a> {
+impl<'a, S> Drop for CopyOutReader<'a, S>
+where
+    S: Stream,
+{
     fn drop(&mut self) {}
 }
 
-impl<'a> CopyOutReader<'a> {
-    #[allow(clippy::new_ret_no_self)]
-    pub(crate) fn new(stream: impls::CopyOut) -> Result<CopyOutReader<'a>, Error> {
-        let mut it = stream.wait();
+impl<'a, S> CopyOutReader<'a, S>
+where
+    S: Stream<Item = Result<Bytes, Error>>,
+{
+    pub(crate) fn new(stream: S) -> Result<CopyOutReader<'a, S>, Error> {
+        let mut it = executor::block_on_stream(Box::pin(stream));
         let cur = match it.next() {
             Some(Ok(cur)) => cur,
             Some(Err(e)) => return Err(e),
@@ -35,7 +43,10 @@ impl<'a> CopyOutReader<'a> {
     }
 }
 
-impl<'a> Read for CopyOutReader<'a> {
+impl<'a, S> Read for CopyOutReader<'a, S>
+where
+    S: Stream<Item = Result<Bytes, Error>>,
+{
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let b = self.fill_buf()?;
         let len = usize::min(buf.len(), b.len());
@@ -45,7 +56,10 @@ impl<'a> Read for CopyOutReader<'a> {
     }
 }
 
-impl<'a> BufRead for CopyOutReader<'a> {
+impl<'a, S> BufRead for CopyOutReader<'a, S>
+where
+    S: Stream<Item = Result<Bytes, Error>>,
+{
     fn fill_buf(&mut self) -> io::Result<&[u8]> {
         if self.cur.remaining() == 0 {
             match self.it.next() {
diff --git a/postgres/src/iter.rs b/postgres/src/iter.rs
new file mode 100644
index 00000000..1f3ffc96
--- /dev/null
+++ b/postgres/src/iter.rs
@@ -0,0 +1,45 @@
+use fallible_iterator::FallibleIterator;
+use futures::executor::{self, BlockingStream};
+use futures::Stream;
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+pub struct Iter<'a, S>
+where
+    S: Stream,
+{
+    it: BlockingStream<Pin<Box<S>>>,
+    _p: PhantomData<&'a mut ()>,
+}
+
+// no-op impl to extend the borrow until drop
+impl<'a, S> Drop for Iter<'a, S>
+where
+    S: Stream,
+{
+    fn drop(&mut self) {}
+}
+
+impl<'a, S> Iter<'a, S>
+where
+    S: Stream,
+{
+    pub fn new(stream: S) -> Iter<'a, S> {
+        Iter {
+            it: executor::block_on_stream(Box::pin(stream)),
+            _p: PhantomData,
+        }
+    }
+}
+
+impl<'a, S, T, E> FallibleIterator for Iter<'a, S>
+where
+    S: Stream<Item = Result<T, E>>,
+{
+    type Item = T;
+    type Error = E;
+
+    fn next(&mut self) -> Result<Option<T>, E> {
+        self.it.next().transpose()
+    }
+}
diff --git a/postgres/src/lib.rs b/postgres/src/lib.rs
index a96eb168..3b5ee03e 100644
--- a/postgres/src/lib.rs
+++ b/postgres/src/lib.rs
@@ -54,6 +54,7 @@
 //! crates, respectively.
 #![doc(html_root_url = "https://docs.rs/postgres/0.16.0-rc.2")]
 #![warn(clippy::all, rust_2018_idioms, missing_docs)]
+#![feature(async_await)]
 
 #[cfg(feature = "runtime")]
 use lazy_static::lazy_static;
@@ -69,14 +70,10 @@ pub use tokio_postgres::{
 pub use crate::client::*;
 #[cfg(feature = "runtime")]
 pub use crate::config::Config;
-pub use crate::copy_out_reader::*;
 #[doc(no_inline)]
 pub use crate::error::Error;
-pub use crate::query_iter::*;
-pub use crate::query_portal_iter::*;
 #[doc(no_inline)]
 pub use crate::row::{Row, SimpleQueryRow};
-pub use crate::simple_query_iter::*;
 #[doc(no_inline)]
 pub use crate::tls::NoTls;
 pub use crate::to_statement::*;
@@ -85,10 +82,9 @@ pub use crate::transaction::*;
 mod client;
 #[cfg(feature = "runtime")]
 pub mod config;
+mod copy_in_stream;
 mod copy_out_reader;
-mod query_iter;
-mod query_portal_iter;
-mod simple_query_iter;
+mod iter;
 mod to_statement;
 mod transaction;
 
diff --git a/postgres/src/query_iter.rs b/postgres/src/query_iter.rs
deleted file mode 100644
index 8f9a5059..00000000
--- a/postgres/src/query_iter.rs
+++ /dev/null
@@ -1,38 +0,0 @@
-use fallible_iterator::FallibleIterator;
-use futures::stream::{self, Stream};
-use std::marker::PhantomData;
-use tokio_postgres::impls;
-use tokio_postgres::{Error, Row};
-
-/// The iterator returned by the `query_iter` method.
-pub struct QueryIter<'a> {
-    it: stream::Wait<impls::Query>,
-    _p: PhantomData<&'a mut ()>,
-}
-
-// no-op impl to extend the borrow until drop
-impl<'a> Drop for QueryIter<'a> {
-    fn drop(&mut self) {}
-}
-
-impl<'a> QueryIter<'a> {
-    pub(crate) fn new(stream: impls::Query) -> QueryIter<'a> {
-        QueryIter {
-            it: stream.wait(),
-            _p: PhantomData,
-        }
-    }
-}
-
-impl<'a> FallibleIterator for QueryIter<'a> {
-    type Item = Row;
-    type Error = Error;
-
-    fn next(&mut self) -> Result<Option<Row>, Error> {
-        match self.it.next() {
-            Some(Ok(row)) => Ok(Some(row)),
-            Some(Err(e)) => Err(e),
-            None => Ok(None),
-        }
-    }
-}
diff --git a/postgres/src/query_portal_iter.rs b/postgres/src/query_portal_iter.rs
deleted file mode 100644
index 8fab3486..00000000
--- a/postgres/src/query_portal_iter.rs
+++ /dev/null
@@ -1,38 +0,0 @@
-use fallible_iterator::FallibleIterator;
-use futures::stream::{self, Stream};
-use std::marker::PhantomData;
-use tokio_postgres::impls;
-use tokio_postgres::{Error, Row};
-
-/// The iterator returned by the `query_portal_iter` method.
-pub struct QueryPortalIter<'a> {
-    it: stream::Wait<impls::QueryPortal>,
-    _p: PhantomData<&'a mut ()>,
-}
-
-// no-op impl to extend the borrow until drop
-impl<'a> Drop for QueryPortalIter<'a> {
-    fn drop(&mut self) {}
-}
-
-impl<'a> QueryPortalIter<'a> {
-    pub(crate) fn new(stream: impls::QueryPortal) -> QueryPortalIter<'a> {
-        QueryPortalIter {
-            it: stream.wait(),
-            _p: PhantomData,
-        }
-    }
-}
-
-impl<'a> FallibleIterator for QueryPortalIter<'a> {
-    type Item = Row;
-    type Error = Error;
-
-    fn next(&mut self) -> Result<Option<Row>, Error> {
-        match self.it.next() {
-            Some(Ok(row)) => Ok(Some(row)),
-            Some(Err(e)) => Err(e),
-            None => Ok(None),
-        }
-    }
-}
diff --git a/postgres/src/simple_query_iter.rs b/postgres/src/simple_query_iter.rs
deleted file mode 100644
index 3053cd30..00000000
--- a/postgres/src/simple_query_iter.rs
+++ /dev/null
@@ -1,38 +0,0 @@
-use fallible_iterator::FallibleIterator;
-use futures::stream::{self, Stream};
-use std::marker::PhantomData;
-use tokio_postgres::impls;
-use tokio_postgres::{Error, SimpleQueryMessage};
-
-/// The iterator returned by the `simple_query_iter` method.
-pub struct SimpleQueryIter<'a> {
-    it: stream::Wait<impls::SimpleQuery>,
-    _p: PhantomData<&'a mut ()>,
-}
-
-// no-op impl to extend borrow until drop
-impl<'a> Drop for SimpleQueryIter<'a> {
-    fn drop(&mut self) {}
-}
-
-impl<'a> SimpleQueryIter<'a> {
-    pub(crate) fn new(stream: impls::SimpleQuery) -> SimpleQueryIter<'a> {
-        SimpleQueryIter {
-            it: stream.wait(),
-            _p: PhantomData,
-        }
-    }
-}
-
-impl<'a> FallibleIterator for SimpleQueryIter<'a> {
-    type Item = SimpleQueryMessage;
-    type Error = Error;
-
-    fn next(&mut self) -> Result<Option<SimpleQueryMessage>, Error> {
-        match self.it.next() {
-            Some(Ok(row)) => Ok(Some(row)),
-            Some(Err(e)) => Err(e),
-            None => Ok(None),
-        }
-    }
-}
diff --git a/postgres/src/test.rs b/postgres/src/test.rs
index 0ff766cb..06399a19 100644
--- a/postgres/src/test.rs
+++ b/postgres/src/test.rs
@@ -1,4 +1,5 @@
 use std::io::Read;
+use tokio::runtime::Runtime;
 use tokio_postgres::types::Type;
 use tokio_postgres::NoTls;
 
@@ -95,6 +96,7 @@ fn transaction_drop() {
     assert_eq!(rows.len(), 0);
 }
 
+/*
 #[test]
 fn nested_transactions() {
     let mut client = Client::connect("host=localhost port=5433 user=postgres", NoTls).unwrap();
@@ -145,6 +147,7 @@ fn nested_transactions() {
     assert_eq!(rows[1].get::<_, i32>(0), 3);
     assert_eq!(rows[2].get::<_, i32>(0), 4);
 }
+*/
 
 #[test]
 fn copy_in() {
@@ -222,3 +225,21 @@ fn portal() {
     assert_eq!(rows.len(), 1);
     assert_eq!(rows[0].get::<_, i32>(0), 3);
 }
+
+#[test]
+fn custom_executor() {
+    let runtime = Runtime::new().unwrap();
+    let mut config = "host=localhost port=5433 user=postgres"
+        .parse::<crate::Config>()
+        .unwrap();
+    config.executor(runtime.executor());
+
+    let mut client = config.connect(NoTls).unwrap();
+
+    let rows = client.query("SELECT $1::TEXT", &[&"hello"]).unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].get::<_, &str>(0), "hello");
+
+    drop(runtime);
+    assert!(client.is_closed());
+}
diff --git a/postgres/src/to_statement.rs b/postgres/src/to_statement.rs
index 5cbe56a3..a77ad28a 100644
--- a/postgres/src/to_statement.rs
+++ b/postgres/src/to_statement.rs
@@ -1,11 +1,28 @@
 use tokio_postgres::Error;
 
-use crate::{Client, Statement};
+use crate::{Client, Statement, Transaction};
 
 mod sealed {
     pub trait Sealed {}
 }
 
+#[doc(hidden)]
+pub trait Prepare {
+    fn prepare(&mut self, query: &str) -> Result<Statement, Error>;
+}
+
+impl Prepare for Client {
+    fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
+        self.prepare(query)
+    }
+}
+
+impl<'a> Prepare for Transaction<'a> {
+    fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
+        self.prepare(query)
+    }
+}
+
 /// A trait abstracting over prepared and unprepared statements.
 ///
 /// Many methods are generic over this bound, so that they support both a raw query string as well as a statement which
@@ -14,13 +31,18 @@ mod sealed {
 /// This trait is "sealed" and cannot be implemented by anything outside this crate.
 pub trait ToStatement: sealed::Sealed {
     #[doc(hidden)]
-    fn __statement(&self, client: &mut Client) -> Result<Statement, Error>;
+    fn __statement<T>(&self, client: &mut T) -> Result<Statement, Error>
+    where
+        T: Prepare;
 }
 
 impl sealed::Sealed for str {}
 
 impl ToStatement for str {
-    fn __statement(&self, client: &mut Client) -> Result<Statement, Error> {
+    fn __statement<T>(&self, client: &mut T) -> Result<Statement, Error>
+    where
+        T: Prepare,
+    {
         client.prepare(self)
     }
 }
@@ -28,7 +50,10 @@ impl ToStatement for str {
 impl sealed::Sealed for Statement {}
 
 impl ToStatement for Statement {
-    fn __statement(&self, _: &mut Client) -> Result<Statement, Error> {
+    fn __statement<T>(&self, _: &mut T) -> Result<Statement, Error>
+    where
+        T: Prepare,
+    {
         Ok(self.clone())
     }
 }
diff --git a/postgres/src/transaction.rs b/postgres/src/transaction.rs
index 8850baee..fcf7ce78 100644
--- a/postgres/src/transaction.rs
+++ b/postgres/src/transaction.rs
@@ -1,79 +1,45 @@
 use fallible_iterator::FallibleIterator;
-use futures::Future;
-use std::io::Read;
+use futures::executor;
+use std::io::{BufRead, Read};
 use tokio_postgres::types::{ToSql, Type};
 use tokio_postgres::{Error, Row, SimpleQueryMessage};
 
-use crate::{
-    Client, CopyOutReader, Portal, QueryIter, QueryPortalIter, SimpleQueryIter, Statement,
-    ToStatement,
-};
+use crate::copy_in_stream::CopyInStream;
+use crate::copy_out_reader::CopyOutReader;
+use crate::iter::Iter;
+use crate::{Portal, Statement, ToStatement};
 
 /// A representation of a PostgreSQL database transaction.
 ///
 /// Transactions will implicitly roll back by default when dropped. Use the `commit` method to commit the changes made
 /// in the transaction. Transactions can be nested, with inner transactions implemented via safepoints.
-pub struct Transaction<'a> {
-    client: &'a mut Client,
-    depth: u32,
-    done: bool,
-}
-
-impl<'a> Drop for Transaction<'a> {
-    fn drop(&mut self) {
-        if !self.done {
-            let _ = self.rollback_inner();
-        }
-    }
-}
+pub struct Transaction<'a>(tokio_postgres::Transaction<'a>);
 
 impl<'a> Transaction<'a> {
-    pub(crate) fn new(client: &'a mut Client) -> Transaction<'a> {
-        Transaction {
-            client,
-            depth: 0,
-            done: false,
-        }
+    pub(crate) fn new(transaction: tokio_postgres::Transaction<'a>) -> Transaction<'a> {
+        Transaction(transaction)
     }
 
     /// Consumes the transaction, committing all changes made within it.
-    pub fn commit(mut self) -> Result<(), Error> {
-        self.done = true;
-        if self.depth == 0 {
-            self.client.simple_query("COMMIT")?;
-        } else {
-            self.client
-                .simple_query(&format!("RELEASE sp{}", self.depth))?;
-        }
-        Ok(())
+    pub fn commit(self) -> Result<(), Error> {
+        executor::block_on(self.0.commit())
     }
 
     /// Rolls the transaction back, discarding all changes made within it.
     ///
     /// This is equivalent to `Transaction`'s `Drop` implementation, but provides any error encountered to the caller.
-    pub fn rollback(mut self) -> Result<(), Error> {
-        self.done = true;
-        self.rollback_inner()
-    }
-
-    fn rollback_inner(&mut self) -> Result<(), Error> {
-        if self.depth == 0 {
-            self.client.simple_query("ROLLBACK")?;
-        } else {
-            self.client
-                .simple_query(&format!("ROLLBACK TO sp{}", self.depth))?;
-        }
-        Ok(())
+    pub fn rollback(self) -> Result<(), Error> {
+        executor::block_on(self.0.rollback())
     }
 
     /// Like `Client::prepare`.
     pub fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
-        self.client.prepare(query)
+        executor::block_on(self.0.prepare(query))
     }
 
     /// Like `Client::prepare_typed`.
     pub fn prepare_typed(&mut self, query: &str, types: &[Type]) -> Result<Statement, Error> {
-        self.client.prepare_typed(query, types)
+        executor::block_on(self.0.prepare_typed(query, types))
     }
 
     /// Like `Client::execute`.
@@ -81,7 +47,8 @@ impl<'a> Transaction<'a> {
     where
         T: ?Sized + ToStatement,
     {
-        self.client.execute(query, params)
+        let statement = query.__statement(self)?;
+        executor::block_on(self.0.execute(&statement, params))
     }
 
     /// Like `Client::query`.
@@ -89,7 +56,7 @@ impl<'a> Transaction<'a> {
     where
         T: ?Sized + ToStatement,
     {
-        self.client.query(query, params)
+        self.query_iter(query, params)?.collect()
     }
 
     /// Like `Client::query_iter`.
@@ -97,11 +64,12 @@ impl<'a> Transaction<'a> {
         &mut self,
         query: &T,
         params: &[&dyn ToSql],
-    ) -> Result<QueryIter<'_>, Error>
+    ) -> Result<impl FallibleIterator<Item = Row, Error = Error>, Error>
     where
         T: ?Sized + ToStatement,
     {
-        self.client.query_iter(query, params)
+        let statement = query.__statement(self)?;
+        Ok(Iter::new(self.0.query(&statement, params)))
     }
 
     /// Binds parameters to a statement, creating a "portal".
@@ -118,8 +86,8 @@ impl<'a> Transaction<'a> {
     where
         T: ?Sized + ToStatement,
     {
-        let statement = query.__statement(&mut self.client)?;
-        self.client.get_mut().bind(&statement, params).wait()
+        let statement = query.__statement(self)?;
+        executor::block_on(self.0.bind(&statement, params))
     }
 
     /// Continues execution of a portal, returning the next set of rows.
@@ -136,10 +104,8 @@ impl<'a> Transaction<'a> {
         &mut self,
         portal: &Portal,
         max_rows: i32,
-    ) -> Result<QueryPortalIter<'_>, Error> {
-        Ok(QueryPortalIter::new(
-            self.client.get_mut().query_portal(&portal, max_rows),
-        ))
+    ) -> Result<impl FallibleIterator<Item = Row, Error = Error>, Error> {
+        Ok(Iter::new(self.0.query_portal(&portal, max_rows)))
     }
 
     /// Like `Client::copy_in`.
@@ -151,42 +117,48 @@ impl<'a> Transaction<'a> {
     ) -> Result<u64, Error>
     where
         T: ?Sized + ToStatement,
-        R: Read,
+        R: Read + Unpin,
     {
-        self.client.copy_in(query, params, reader)
+        let statement = query.__statement(self)?;
+        executor::block_on(self.0.copy_in(&statement, params, CopyInStream(reader)))
     }
 
     /// Like `Client::copy_out`.
-    pub fn copy_out<T>(
-        &mut self,
+    pub fn copy_out<'b, T>(
+        &'a mut self,
         query: &T,
         params: &[&dyn ToSql],
-    ) -> Result<CopyOutReader<'_>, Error>
+    ) -> Result<impl BufRead + 'b, Error>
     where
         T: ?Sized + ToStatement,
     {
-        self.client.copy_out(query, params)
+        let statement = query.__statement(self)?;
+        let stream = self.0.copy_out(&statement, params);
+        CopyOutReader::new(stream)
     }
 
     /// Like `Client::simple_query`.
     pub fn simple_query(&mut self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
-        self.client.simple_query(query)
+        self.simple_query_iter(query)?.collect()
     }
 
     /// Like `Client::simple_query_iter`.
-    pub fn simple_query_iter(&mut self, query: &str) -> Result<SimpleQueryIter<'_>, Error> {
-        self.client.simple_query_iter(query)
+    pub fn simple_query_iter<'b>(
+        &'b mut self,
+        query: &str,
+    ) -> Result<impl FallibleIterator<Item = SimpleQueryMessage, Error = Error> + 'b, Error> {
+        Ok(Iter::new(self.0.simple_query(query)))
     }
 
-    /// Like `Client::transaction`.
-    pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
-        let depth = self.depth + 1;
-        self.client
-            .simple_query(&format!("SAVEPOINT sp{}", depth))?;
-        Ok(Transaction {
-            client: self.client,
-            depth,
-            done: false,
-        })
-    }
+    //    /// Like `Client::transaction`.
+    //    pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
+    //        let depth = self.depth + 1;
+    //        self.client
+    //            .simple_query(&format!("SAVEPOINT sp{}", depth))?;
+    //        Ok(Transaction {
+    //            client: self.client,
+    //            depth,
+    //            done: false,
+    //        })
+    //    }
 }
diff --git a/tokio-postgres/src/client.rs b/tokio-postgres/src/client.rs
index 03f0d031..ba43be6d 100644
--- a/tokio-postgres/src/client.rs
+++ b/tokio-postgres/src/client.rs
@@ -370,4 +370,11 @@ impl Client {
             self.secret_key,
         )
     }
+
+    /// Determines if the connection to the server has already closed.
+    ///
+    /// In that case, all future queries will fail.
+    pub fn is_closed(&self) -> bool {
+        self.inner.sender.is_closed()
+    }
 }
diff --git a/tokio-postgres/src/config.rs b/tokio-postgres/src/config.rs
index 68fe4459..19df1a35 100644
--- a/tokio-postgres/src/config.rs
+++ b/tokio-postgres/src/config.rs
@@ -126,7 +126,7 @@ pub(crate) enum Host {
 /// ```not_rust
 /// postgresql:///mydb?user=user&host=/var/lib/postgresql
 /// ```
-#[derive(PartialEq)]
+#[derive(PartialEq, Clone)]
 pub struct Config {
     pub(crate) user: Option<String>,
     pub(crate) password: Option<Vec<u8>>,
diff --git a/tokio-postgres/src/connection.rs b/tokio-postgres/src/connection.rs
index c8d55a34..c3f90ef8 100644
--- a/tokio-postgres/src/connection.rs
+++ b/tokio-postgres/src/connection.rs
@@ -5,6 +5,7 @@ use crate::maybe_tls_stream::MaybeTlsStream;
 use crate::{AsyncMessage, Error, Notification};
 use fallible_iterator::FallibleIterator;
 use futures::channel::mpsc;
+use futures::stream::FusedStream;
 use futures::{ready, Sink, Stream, StreamExt};
 use log::trace;
 use postgres_protocol::message::backend::Message;
@@ -173,6 +174,10 @@ where
             return Poll::Ready(Some(messages));
         }
 
+        if self.receiver.is_terminated() {
+            return Poll::Ready(None);
+        }
+
         match self.receiver.poll_next_unpin(cx) {
             Poll::Ready(Some(request)) => {
                 trace!("polled new request");