Fully working prepare

tokio-postgres/src/client.rs

@@ -1,7 +1,7 @@
 use crate::codec::BackendMessages;
 use crate::connection::{Request, RequestMessages};
 use crate::prepare::prepare;
-use crate::query::{execute, query, Query};
+use crate::query::{self, Query};
 use crate::types::{Oid, ToSql, Type};
 use crate::{Error, Statement};
 use fallible_iterator::FallibleIterator;
@@ -42,9 +42,9 @@ impl Responses {
 }
 
 struct State {
-    has_typeinfo: bool,
+    typeinfo: Option<Statement>,
-    has_typeinfo_composite: bool,
+    typeinfo_composite: Option<Statement>,
-    has_typeinfo_enum: bool,
+    typeinfo_enum: Option<Statement>,
     types: HashMap<Oid, Type>,
 }
 
@@ -67,36 +67,36 @@ impl InnerClient {
         })
     }
 
-    pub fn has_typeinfo(&self) -> bool {
+    pub fn typeinfo(&self) -> Option<Statement> {
-        self.state.lock().has_typeinfo
+        self.state.lock().typeinfo.clone()
     }
 
-    pub fn set_has_typeinfo(&self) {
+    pub fn set_typeinfo(&self, statement: &Statement) {
-        self.state.lock().has_typeinfo = true;
+        self.state.lock().typeinfo = Some(statement.clone());
     }
 
-    pub fn has_typeinfo_composite(&self) -> bool {
+    pub fn typeinfo_composite(&self) -> Option<Statement> {
-        self.state.lock().has_typeinfo_composite
+        self.state.lock().typeinfo_composite.clone()
     }
 
-    pub fn set_has_typeinfo_composite(&self) {
+    pub fn set_typeinfo_composite(&self, statement: &Statement) {
-        self.state.lock().has_typeinfo_composite = true;
+        self.state.lock().typeinfo_composite = Some(statement.clone());
     }
 
-    pub fn has_typeinfo_enum(&self) -> bool {
+    pub fn typeinfo_enum(&self) -> Option<Statement> {
-        self.state.lock().has_typeinfo_enum
+        self.state.lock().typeinfo_enum.clone()
     }
 
-    pub fn set_has_typeinfo_enum(&self) {
+    pub fn set_typeinfo_enum(&self, statement: &Statement) {
-        self.state.lock().has_typeinfo_enum = true;
+        self.state.lock().typeinfo_enum = Some(statement.clone());
     }
 
     pub fn type_(&self, oid: Oid) -> Option<Type> {
         self.state.lock().types.get(&oid).cloned()
     }
 
-    pub fn set_type(&self, oid: Oid, type_: Type) {
+    pub fn set_type(&self, oid: Oid, type_: &Type) {
-        self.state.lock().types.insert(oid, type_);
+        self.state.lock().types.insert(oid, type_.clone());
     }
 }
 
@@ -116,9 +116,9 @@ impl Client {
             inner: Arc::new(InnerClient {
                 sender,
                 state: Mutex::new(State {
-                    has_typeinfo: false,
+                    typeinfo: None,
-                    has_typeinfo_composite: false,
+                    typeinfo_composite: None,
-                    has_typeinfo_enum: false,
+                    typeinfo_enum: None,
                     types: HashMap::new(),
                 }),
             }),
@@ -131,58 +131,59 @@ impl Client {
         self.inner.clone()
     }
 
-    pub fn prepare<'a>(
+    pub fn prepare(&mut self, query: &str) -> impl Future<Output = Result<Statement, Error>> {
-        &mut self,
-        query: &'a str,
-    ) -> impl Future<Output = Result<Statement, Error>> + 'a {
         self.prepare_typed(query, &[])
     }
 
-    pub fn prepare_typed<'a>(
+    pub fn prepare_typed(
         &mut self,
-        query: &'a str,
+        query: &str,
-        parameter_types: &'a [Type],
+        parameter_types: &[Type],
-    ) -> impl Future<Output = Result<Statement, Error>> + 'a {
+    ) -> impl Future<Output = Result<Statement, Error>> {
         prepare(self.inner(), query, parameter_types)
     }
 
-    pub fn query<'a>(
+    pub fn query(
         &mut self,
-        statement: &'a Statement,
+        statement: &Statement,
-        params: &'a [&dyn ToSql],
+        params: &[&dyn ToSql],
-    ) -> impl Future<Output = Result<Query, Error>> + 'a {
+    ) -> impl Future<Output = Result<Query, Error>> {
-        self.query_iter(statement, params.iter().cloned())
+        let buf = query::encode(statement, params.iter().cloned());
+        query::query(self.inner(), statement.clone(), buf)
     }
 
     pub fn query_iter<'a, I>(
         &mut self,
-        statement: &'a Statement,
+        statement: &Statement,
         params: I,
-    ) -> impl Future<Output = Result<Query, Error>> + 'a
+    ) -> impl Future<Output = Result<Query, Error>>
     where
-        I: IntoIterator<Item = &'a dyn ToSql> + 'a,
+        I: IntoIterator<Item = &'a dyn ToSql>,
         I::IntoIter: ExactSizeIterator,
     {
-        query(self.inner(), statement, params)
+        let buf = query::encode(statement, params);
+        query::query(self.inner(), statement.clone(), buf)
     }
 
-    pub fn execute<'a>(
+    pub fn execute(
         &mut self,
-        statement: &'a Statement,
+        statement: &Statement,
-        params: &'a [&dyn ToSql],
+        params: &[&dyn ToSql],
-    ) -> impl Future<Output = Result<u64, Error>> + 'a {
+    ) -> impl Future<Output = Result<u64, Error>> {
-        self.execute_iter(statement, params.iter().cloned())
+        let buf = query::encode(statement, params.iter().cloned());
+        query::execute(self.inner(), buf)
     }
 
     pub fn execute_iter<'a, I>(
         &mut self,
-        statement: &'a Statement,
+        statement: &Statement,
         params: I,
-    ) -> impl Future<Output = Result<u64, Error>> + 'a
+    ) -> impl Future<Output = Result<u64, Error>>
     where
-        I: IntoIterator<Item = &'a dyn ToSql> + 'a,
+        I: IntoIterator<Item = &'a dyn ToSql>,
         I::IntoIter: ExactSizeIterator,
     {
-        execute(self.inner(), statement, params)
+        let buf = query::encode(statement, params);
+        query::execute(self.inner(), buf)
     }
 }

tokio-postgres/src/prepare.rs

@@ -1,64 +1,119 @@
 use crate::client::InnerClient;
 use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
-use crate::types::{Oid, Type};
+use crate::error::SqlState;
+use crate::query;
+use crate::types::{Field, Kind, Oid, ToSql, Type};
 use crate::{Column, Error, Statement};
 use fallible_iterator::FallibleIterator;
+use futures::{future, StreamExt, TryStreamExt};
 use postgres_protocol::message::backend::Message;
 use postgres_protocol::message::frontend;
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
+const TYPEINFO_STMT: &str = "typeinfo";
+
+const TYPEINFO_QUERY: &str = "\
+SELECT t.typname, t.typtype, t.typelem, r.rngsubtype, t.typbasetype, n.nspname, t.typrelid
+FROM pg_catalog.pg_type t
+LEFT OUTER JOIN pg_catalog.pg_range r ON r.rngtypid = t.oid
+INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid
+WHERE t.oid = $1
+";
+
+// Range types weren't added until Postgres 9.2, so pg_range may not exist
+const TYPEINFO_FALLBACK_QUERY: &str = "\
+SELECT t.typname, t.typtype, t.typelem, NULL::OID, t.typbasetype, n.nspname, t.typrelid
+FROM pg_catalog.pg_type t
+INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid
+WHERE t.oid = $1
+";
+
+const TYPEINFO_ENUM_QUERY: &str = "\
+SELECT enumlabel
+FROM pg_catalog.pg_enum
+WHERE enumtypid = $1
+ORDER BY enumsortorder
+";
+
+// Postgres 9.0 didn't have enumsortorder
+const TYPEINFO_ENUM_FALLBACK_QUERY: &str = "\
+SELECT enumlabel
+FROM pg_catalog.pg_enum
+WHERE enumtypid = $1
+ORDER BY oid
+";
+
+const TYPEINFO_COMPOSITE_QUERY: &str = "\
+SELECT attname, atttypid
+FROM pg_catalog.pg_attribute
+WHERE attrelid = $1
+AND NOT attisdropped
+AND attnum > 0
+ORDER BY attnum
+";
+
 static NEXT_ID: AtomicUsize = AtomicUsize::new(0);
 
-pub async fn prepare(
+pub fn prepare(
     client: Arc<InnerClient>,
     query: &str,
     types: &[Type],
-) -> Result<Statement, Error> {
+) -> impl Future<Output = Result<Statement, Error>> {
     let name = format!("s{}", NEXT_ID.fetch_add(1, Ordering::SeqCst));
+    let buf = encode(&name, query, types);
 
+    async move {
+        let buf = buf?;
+        let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+
+        match responses.next().await? {
+            Message::ParseComplete => {}
+            _ => return Err(Error::unexpected_message()),
+        }
+
+        let parameter_description = match responses.next().await? {
+            Message::ParameterDescription(body) => body,
+            _ => return Err(Error::unexpected_message()),
+        };
+
+        let row_description = match responses.next().await? {
+            Message::RowDescription(body) => Some(body),
+            Message::NoData => None,
+            _ => return Err(Error::unexpected_message()),
+        };
+
+        let mut parameters = vec![];
+        let mut it = parameter_description.parameters();
+        while let Some(oid) = it.next().map_err(Error::parse)? {
+            let type_ = get_type(&client, oid).await?;
+            parameters.push(type_);
+        }
+
+        let mut columns = vec![];
+        if let Some(row_description) = row_description {
+            let mut it = row_description.fields();
+            while let Some(field) = it.next().map_err(Error::parse)? {
+                let type_ = get_type(&client, field.type_oid()).await?;
+                let column = Column::new(field.name().to_string(), type_);
+                columns.push(column);
+            }
+        }
+
+        Ok(Statement::new(&client, name, parameters, columns))
+    }
+}
+
+fn encode(name: &str, query: &str, types: &[Type]) -> Result<Vec<u8>, Error> {
     let mut buf = vec![];
-    frontend::parse(&name, query, types.iter().map(Type::oid), &mut buf).map_err(Error::encode)?;
+    frontend::parse(name, query, types.iter().map(Type::oid), &mut buf).map_err(Error::encode)?;
     frontend::describe(b'S', &name, &mut buf).map_err(Error::encode)?;
     frontend::sync(&mut buf);
 
-    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+    Ok(buf)
-
-    match responses.next().await? {
-        Message::ParseComplete => {}
-        _ => return Err(Error::unexpected_message()),
-    }
-
-    let parameter_description = match responses.next().await? {
-        Message::ParameterDescription(body) => body,
-        _ => return Err(Error::unexpected_message()),
-    };
-
-    let row_description = match responses.next().await? {
-        Message::RowDescription(body) => Some(body),
-        Message::NoData => None,
-        _ => return Err(Error::unexpected_message()),
-    };
-
-    let mut parameters = vec![];
-    let mut it = parameter_description.parameters();
-    while let Some(oid) = it.next().map_err(Error::parse)? {
-        let type_ = get_type(&client, oid).await?;
-        parameters.push(type_);
-    }
-
-    let mut columns = vec![];
-    if let Some(row_description) = row_description {
-        let mut it = row_description.fields();
-        while let Some(field) = it.next().map_err(Error::parse)? {
-            let type_ = get_type(&client, field.type_oid()).await?;
-            let column = Column::new(field.name().to_string(), type_);
-            columns.push(column);
-        }
-    }
-
-    Ok(Statement::new(&client, name, parameters, columns))
 }
 
 async fn get_type(client: &Arc<InnerClient>, oid: Oid) -> Result<Type, Error> {
@@ -70,5 +125,133 @@ async fn get_type(client: &Arc<InnerClient>, oid: Oid) -> Result<Type, Error> {
         return Ok(type_);
     }
 
-    unimplemented!()
+    let stmt = typeinfo_statement(client).await?;
+
+    let params: &[&dyn ToSql] = &[&oid];
+    let buf = query::encode(&stmt, params.iter().cloned());
+    let mut rows = query::query(client.clone(), stmt, buf).await?;
+
+    let row = match rows.try_next().await? {
+        Some(row) => row,
+        None => return Err(Error::unexpected_message()),
+    };
+
+    let name: String = row.try_get(0)?;
+    let type_: i8 = row.try_get(1)?;
+    let elem_oid: Oid = row.try_get(2)?;
+    let rngsubtype: Option<Oid> = row.try_get(3)?;
+    let basetype: Oid = row.try_get(4)?;
+    let schema: String = row.try_get(5)?;
+    let relid: Oid = row.try_get(6)?;
+
+    let kind = if type_ == b'e' as i8 {
+        let variants = get_enum_variants(client, oid).await?;
+        Kind::Enum(variants)
+    } else if type_ == b'p' as i8 {
+        Kind::Pseudo
+    } else if basetype != 0 {
+        let type_ = get_type_rec(client, basetype).await?;
+        Kind::Domain(type_)
+    } else if elem_oid != 0 {
+        let type_ = get_type_rec(client, elem_oid).await?;
+        Kind::Array(type_)
+    } else if relid != 0 {
+        let fields = get_composite_fields(client, relid).await?;
+        Kind::Composite(fields)
+    } else if let Some(rngsubtype) = rngsubtype {
+        let type_ = get_type_rec(client, rngsubtype).await?;
+        Kind::Range(type_)
+    } else {
+        Kind::Simple
+    };
+
+    let type_ = Type::new(name, oid, kind, schema);
+    client.set_type(oid, &type_);
+
+    Ok(type_)
+}
+
+fn get_type_rec<'a>(
+    client: &'a Arc<InnerClient>,
+    oid: Oid,
+) -> Pin<Box<dyn Future<Output = Result<Type, Error>> + 'a>> {
+    Box::pin(get_type(client, oid))
+}
+
+async fn typeinfo_statement(client: &Arc<InnerClient>) -> Result<Statement, Error> {
+    if let Some(stmt) = client.typeinfo() {
+        return Ok(stmt);
+    }
+
+    let stmt = match Box::pin(prepare(client.clone(), TYPEINFO_QUERY, &[])).await {
+        Ok(stmt) => stmt,
+        Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_TABLE) => {
+            Box::pin(prepare(client.clone(), TYPEINFO_FALLBACK_QUERY, &[])).await?
+        }
+        Err(e) => return Err(e),
+    };
+
+    client.set_typeinfo(&stmt);
+    Ok(stmt)
+}
+
+async fn get_enum_variants(client: &Arc<InnerClient>, oid: Oid) -> Result<Vec<String>, Error> {
+    let stmt = typeinfo_enum_statement(client).await?;
+
+    let params: &[&dyn ToSql] = &[&oid];
+    let buf = query::encode(&stmt, params.iter().cloned());
+    query::query(client.clone(), stmt, buf)
+        .await?
+        .and_then(|row| future::ready(row.try_get(0)))
+        .try_collect()
+        .await
+}
+
+async fn typeinfo_enum_statement(client: &Arc<InnerClient>) -> Result<Statement, Error> {
+    if let Some(stmt) = client.typeinfo_enum() {
+        return Ok(stmt);
+    }
+
+    let stmt = match Box::pin(prepare(client.clone(), TYPEINFO_ENUM_QUERY, &[])).await {
+        Ok(stmt) => stmt,
+        Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_COLUMN) => {
+            Box::pin(prepare(client.clone(), TYPEINFO_ENUM_FALLBACK_QUERY, &[])).await?
+        }
+        Err(e) => return Err(e),
+    };
+
+    client.set_typeinfo_enum(&stmt);
+    Ok(stmt)
+}
+
+async fn get_composite_fields(client: &Arc<InnerClient>, oid: Oid) -> Result<Vec<Field>, Error> {
+    let stmt = typeinfo_composite_statement(client).await?;
+
+    let params: &[&dyn ToSql] = &[&oid];
+    let buf = query::encode(&stmt, params.iter().cloned());
+    let rows = query::query(client.clone(), stmt, buf)
+        .await?
+        .try_collect::<Vec<_>>()
+        .await?;
+
+    let mut fields = vec![];
+    for row in rows {
+        let name = row.try_get(0)?;
+        let oid = row.try_get(1)?;
+        let type_ = Box::pin(get_type(client, oid)).await?;
+        fields.push(Field::new(name, type_));
+    }
+
+    Ok(fields)
+}
+
+async fn typeinfo_composite_statement(client: &Arc<InnerClient>) -> Result<Statement, Error> {
+    if let Some(stmt) = client.typeinfo_composite() {
+        return Ok(stmt);
+    }
+
+    let stmt = Box::pin(prepare(client.clone(), TYPEINFO_COMPOSITE_QUERY, &[])).await?;
+
+    client.set_typeinfo_composite(&stmt);
+    Ok(stmt)
 }

tokio-postgres/src/query.rs

@@ -11,33 +11,21 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-pub async fn query<'a, I>(
+pub async fn query(
     client: Arc<InnerClient>,
-    statement: &Statement,
+    statement: Statement,
-    params: I,
+    buf: Result<Vec<u8>, Error>,
-) -> Result<Query, Error>
+) -> Result<Query, Error> {
-where
+    let responses = start(client, buf).await?;
-    I: IntoIterator<Item = &'a dyn ToSql>,
-    I::IntoIter: ExactSizeIterator,
-{
-    let responses = start(&client, &statement, params).await?;
 
     Ok(Query {
-        statement: statement.clone(),
+        statement,
         responses,
     })
 }
 
-pub async fn execute<'a, I>(
+pub async fn execute(client: Arc<InnerClient>, buf: Result<Vec<u8>, Error>) -> Result<u64, Error> {
-    client: Arc<InnerClient>,
+    let mut responses = start(client, buf).await?;
-    statement: &Statement,
-    params: I,
-) -> Result<u64, Error>
-where
-    I: IntoIterator<Item = &'a dyn ToSql>,
-    I::IntoIter: ExactSizeIterator,
-{
-    let mut responses = start(&client, &statement, params).await?;
 
     loop {
         match responses.next().await? {
@@ -59,11 +47,19 @@ where
     }
 }
 
-async fn start<'a, I>(
+async fn start(client: Arc<InnerClient>, buf: Result<Vec<u8>, Error>) -> Result<Responses, Error> {
-    client: &Arc<InnerClient>,
+    let buf = buf?;
-    statement: &Statement,
+    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
-    params: I,
+
-) -> Result<Responses, Error>
+    match responses.next().await? {
+        Message::BindComplete => {}
+        _ => return Err(Error::unexpected_message()),
+    }
+
+    Ok(responses)
+}
+
+pub fn encode<'a, I>(statement: &Statement, params: I) -> Result<Vec<u8>, Error>
 where
     I: IntoIterator<Item = &'a dyn ToSql>,
     I::IntoIter: ExactSizeIterator,
@@ -105,14 +101,7 @@ where
     frontend::execute("", 0, &mut buf).map_err(Error::encode)?;
     frontend::sync(&mut buf);
 
-    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+    Ok(buf)
-
-    match responses.next().await? {
-        Message::BindComplete => {}
-        _ => return Err(Error::unexpected_message()),
-    }
-
-    Ok(responses)
 }
 
 pub struct Query {

tokio-postgres/src/types/mod.rs

@@ -107,7 +107,7 @@ impl fmt::Display for Type {
 }
 
 impl Type {
-    pub(crate) fn _new(name: String, oid: Oid, kind: Kind, schema: String) -> Type {
+    pub(crate) fn new(name: String, oid: Oid, kind: Kind, schema: String) -> Type {
         Type(Inner::Other(Arc::new(Other {
             name,
             oid,
@@ -175,6 +175,10 @@ pub struct Field {
 }
 
 impl Field {
+    pub(crate) fn new(name: String, type_: Type) -> Field {
+        Field { name, type_ }
+    }
+
     /// Returns the name of the field.
     pub fn name(&self) -> &str {
         &self.name
@@ -186,12 +190,6 @@ impl Field {
     }
 }
 
-impl Field {
-    pub(crate) fn new(name: String, type_: Type) -> Field {
-        Field { name, type_ }
-    }
-}
-
 /// An error indicating that a `NULL` Postgres value was passed to a `FromSql`
 /// implementation that does not support `NULL` values.
 #[derive(Debug, Clone, Copy)]
@@ -543,7 +541,7 @@ pub enum IsNull {
 ///
 /// `ToSql` is implemented for `Vec<T>` and `&[T]` where `T` implements `ToSql`,
 /// and corresponds to one-dimensional Postgres arrays with an index offset of 1.
-pub trait ToSql: fmt::Debug + Sync + Send {
+pub trait ToSql: fmt::Debug {
     /// Converts the value of `self` into the binary format of the specified
     /// Postgres `Type`, appending it to `out`.
     ///
@@ -744,7 +742,7 @@ simple_to!(f64, float8_to_sql, FLOAT8);
 
 impl<H> ToSql for HashMap<String, Option<String>, H>
 where
-    H: BuildHasher + Sync + Send,
+    H: BuildHasher,
 {
     fn to_sql(&self, _: &Type, w: &mut Vec<u8>) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
         types::hstore_to_sql(

tokio-postgres/src/types/serde_json_1.rs

@@ -33,7 +33,7 @@ where
 
 impl<T> ToSql for Json<T>
 where
-    T: Serialize + Debug + Sync + Send,
+    T: Serialize + Debug,
 {
     fn to_sql(&self, ty: &Type, out: &mut Vec<u8>) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
         if *ty == Type::JSONB {

tokio-postgres/tests/test/main.rs

@@ -1,19 +1,17 @@
 #![warn(rust_2018_idioms)]
 #![feature(async_await)]
 
-use futures::{try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
+use futures::{try_join, FutureExt, TryFutureExt, TryStreamExt};
 use tokio::net::TcpStream;
 use tokio_postgres::error::SqlState;
 use tokio_postgres::tls::{NoTls, NoTlsStream};
-use tokio_postgres::types::Type;
+use tokio_postgres::types::{Kind, Type};
 use tokio_postgres::{Client, Config, Connection, Error};
 
 mod parse;
 #[cfg(feature = "runtime")]
 mod runtime;
-/*
 mod types;
-*/
 
 async fn connect_raw(s: &str) -> Result<(Client, Connection<TcpStream, NoTlsStream>), Error> {
     let socket = TcpStream::connect(&"127.0.0.1:5433".parse().unwrap())
@@ -100,13 +98,13 @@ async fn scram_password_ok() {
 async fn pipelined_prepare() {
     let mut client = connect("user=postgres").await;
 
-    let prepare1 = client.prepare("SELECT $1::TEXT");
+    let prepare1 = client.prepare("SELECT $1::HSTORE[]");
     let prepare2 = client.prepare("SELECT $1::BIGINT");
 
     let (statement1, statement2) = try_join!(prepare1, prepare2).unwrap();
 
-    assert_eq!(statement1.params()[0], Type::TEXT);
+    assert_eq!(statement1.params()[0].name(), "_hstore");
-    assert_eq!(statement1.columns()[0].type_(), &Type::TEXT);
+    assert_eq!(statement1.columns()[0].type_().name(), "_hstore");
 
     assert_eq!(statement2.params()[0], Type::INT8);
     assert_eq!(statement2.columns()[0].type_(), &Type::INT8);
@@ -140,6 +138,125 @@ async fn insert_select() {
     assert_eq!(rows[1].get::<_, &str>(1), "bob");
 }
 
+#[tokio::test]
+async fn custom_enum() {
+    let mut client = connect("user=postgres").await;
+
+    let create = client
+        .prepare(
+            "CREATE TYPE pg_temp.mood AS ENUM (
+                'sad',
+                'ok',
+                'happy'
+            )",
+        )
+        .await
+        .unwrap();
+    client.execute(&create, &[]).await.unwrap();
+
+    let select = client.prepare("SELECT $1::mood").await.unwrap();
+
+    let ty = &select.params()[0];
+    assert_eq!("mood", ty.name());
+    assert_eq!(
+        &Kind::Enum(vec![
+            "sad".to_string(),
+            "ok".to_string(),
+            "happy".to_string(),
+        ]),
+        ty.kind(),
+    );
+}
+
+#[tokio::test]
+async fn custom_domain() {
+    let mut client = connect("user=postgres").await;
+
+    let create = client
+        .prepare("CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16)")
+        .await
+        .unwrap();
+    client.execute(&create, &[]).await.unwrap();
+
+    let select = client.prepare("SELECT $1::session_id").await.unwrap();
+
+    let ty = &select.params()[0];
+    assert_eq!("session_id", ty.name());
+    assert_eq!(&Kind::Domain(Type::BYTEA), ty.kind());
+}
+
+#[tokio::test]
+async fn custom_array() {
+    let mut client = connect("user=postgres").await;
+
+    let select = client.prepare("SELECT $1::HSTORE[]").await.unwrap();
+
+    let ty = &select.params()[0];
+    assert_eq!("_hstore", ty.name());
+    match ty.kind() {
+        Kind::Array(ty) => {
+            assert_eq!("hstore", ty.name());
+            assert_eq!(&Kind::Simple, ty.kind());
+        }
+        _ => panic!("unexpected kind"),
+    }
+}
+
+#[tokio::test]
+async fn custom_composite() {
+    let mut client = connect("user=postgres").await;
+
+    let create = client
+        .prepare(
+            "CREATE TYPE pg_temp.inventory_item AS (
+            name TEXT,
+            supplier INTEGER,
+            price NUMERIC
+        )",
+        )
+        .await
+        .unwrap();
+    client.execute(&create, &[]).await.unwrap();
+
+    let select = client.prepare("SELECT $1::inventory_item").await.unwrap();
+
+    let ty = &select.params()[0];
+    assert_eq!(ty.name(), "inventory_item");
+    match ty.kind() {
+        Kind::Composite(fields) => {
+            assert_eq!(fields[0].name(), "name");
+            assert_eq!(fields[0].type_(), &Type::TEXT);
+            assert_eq!(fields[1].name(), "supplier");
+            assert_eq!(fields[1].type_(), &Type::INT4);
+            assert_eq!(fields[2].name(), "price");
+            assert_eq!(fields[2].type_(), &Type::NUMERIC);
+        }
+        _ => panic!("unexpected kind"),
+    }
+}
+
+#[tokio::test]
+async fn custom_range() {
+    let mut client = connect("user=postgres").await;
+
+    let create = client
+        .prepare(
+            "CREATE TYPE pg_temp.floatrange AS RANGE (
+            subtype = float8,
+            subtype_diff = float8mi
+        )",
+        )
+        .await
+        .unwrap();
+    client.execute(&create, &[]).await.unwrap();
+
+    let select = client.prepare("SELECT $1::floatrange").await.unwrap();
+
+    let ty = &select.params()[0];
+    assert_eq!("floatrange", ty.name());
+    assert_eq!(&Kind::Range(Type::FLOAT8), ty.kind());
+}
+
 /*
 #[test]
 fn query_portal() {
@@ -219,182 +336,6 @@ fn cancel_query_raw() {
     let ((), ()) = runtime.block_on(sleep.join(cancel)).unwrap();
 }
 
-#[test]
-fn custom_enum() {
-    let _ = env_logger::try_init();
-    let mut runtime = Runtime::new().unwrap();
-
-    let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.handle().spawn(connection).unwrap();
-
-    runtime
-        .block_on(
-            client
-                .simple_query(
-                    "CREATE TYPE pg_temp.mood AS ENUM (
-                        'sad',
-                        'ok',
-                        'happy'
-                    )",
-                )
-                .for_each(|_| Ok(())),
-        )
-        .unwrap();
-
-    let select = client.prepare("SELECT $1::mood");
-    let select = runtime.block_on(select).unwrap();
-
-    let ty = &select.params()[0];
-    assert_eq!("mood", ty.name());
-    assert_eq!(
-        &Kind::Enum(vec![
-            "sad".to_string(),
-            "ok".to_string(),
-            "happy".to_string(),
-        ]),
-        ty.kind()
-    );
-}
-
-#[test]
-fn custom_domain() {
-    let _ = env_logger::try_init();
-    let mut runtime = Runtime::new().unwrap();
-
-    let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.handle().spawn(connection).unwrap();
-
-    runtime
-        .block_on(
-            client
-                .simple_query(
-                    "CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16)",
-                )
-                .for_each(|_| Ok(())),
-        )
-        .unwrap();
-
-    let select = client.prepare("SELECT $1::session_id");
-    let select = runtime.block_on(select).unwrap();
-
-    let ty = &select.params()[0];
-    assert_eq!("session_id", ty.name());
-    assert_eq!(&Kind::Domain(Type::BYTEA), ty.kind());
-}
-
-#[test]
-fn custom_array() {
-    let _ = env_logger::try_init();
-    let mut runtime = Runtime::new().unwrap();
-
-    let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.handle().spawn(connection).unwrap();
-
-    let select = client.prepare("SELECT $1::HSTORE[]");
-    let select = runtime.block_on(select).unwrap();
-
-    let ty = &select.params()[0];
-    assert_eq!("_hstore", ty.name());
-    match *ty.kind() {
-        Kind::Array(ref ty) => {
-            assert_eq!("hstore", ty.name());
-            assert_eq!(&Kind::Simple, ty.kind());
-        }
-        _ => panic!("unexpected kind"),
-    }
-}
-
-#[test]
-fn custom_composite() {
-    let _ = env_logger::try_init();
-    let mut runtime = Runtime::new().unwrap();
-
-    let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.handle().spawn(connection).unwrap();
-
-    runtime
-        .block_on(
-            client
-                .simple_query(
-                    "CREATE TYPE pg_temp.inventory_item AS (
-                        name TEXT,
-                        supplier INTEGER,
-                        price NUMERIC
-                    )",
-                )
-                .for_each(|_| Ok(())),
-        )
-        .unwrap();
-
-    let select = client.prepare("SELECT $1::inventory_item");
-    let select = runtime.block_on(select).unwrap();
-
-    let ty = &select.params()[0];
-    assert_eq!(ty.name(), "inventory_item");
-    match *ty.kind() {
-        Kind::Composite(ref fields) => {
-            assert_eq!(fields[0].name(), "name");
-            assert_eq!(fields[0].type_(), &Type::TEXT);
-            assert_eq!(fields[1].name(), "supplier");
-            assert_eq!(fields[1].type_(), &Type::INT4);
-            assert_eq!(fields[2].name(), "price");
-            assert_eq!(fields[2].type_(), &Type::NUMERIC);
-        }
-        ref t => panic!("bad type {:?}", t),
-    }
-}
-
-#[test]
-fn custom_range() {
-    let _ = env_logger::try_init();
-    let mut runtime = Runtime::new().unwrap();
-
-    let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.handle().spawn(connection).unwrap();
-
-    runtime
-        .block_on(
-            client
-                .simple_query(
-                    "CREATE TYPE pg_temp.floatrange AS RANGE (
-                        subtype = float8,
-                        subtype_diff = float8mi
-                    )",
-                )
-                .for_each(|_| Ok(())),
-        )
-        .unwrap();
-
-    let select = client.prepare("SELECT $1::floatrange");
-    let select = runtime.block_on(select).unwrap();
-
-    let ty = &select.params()[0];
-    assert_eq!("floatrange", ty.name());
-    assert_eq!(&Kind::Range(Type::FLOAT8), ty.kind());
-}
-
-#[test]
-fn custom_simple() {
-    let _ = env_logger::try_init();
-    let mut runtime = Runtime::new().unwrap();
-
-    let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.handle().spawn(connection).unwrap();
-
-    let select = client.prepare("SELECT $1::HSTORE");
-    let select = runtime.block_on(select).unwrap();
-
-    let ty = &select.params()[0];
-    assert_eq!("hstore", ty.name());
-    assert_eq!(&Kind::Simple, ty.kind());
-}
-
 #[test]
 fn notifications() {
     let _ = env_logger::try_init();

tokio-postgres/tests/test/runtime.rs

@@ -1,6 +1,5 @@
-use futures::{Future, FutureExt, Stream};
+use futures::{FutureExt, TryStreamExt};
 use std::time::{Duration, Instant};
-use tokio::runtime::current_thread::Runtime;
 use tokio::timer::Delay;
 use tokio_postgres::error::SqlState;
 use tokio_postgres::NoTls;
@@ -10,10 +9,9 @@ async fn smoke_test(s: &str) {
     let connection = connection.map(|e| e.unwrap());
     tokio::spawn(connection);
 
-    /*
+    let stmt = client.prepare("SELECT $1::INT").await.unwrap();
-    let execute = client.simple_query("SELECT 1").for_each(|_| Ok(()));
+    let rows = client.query(&stmt, &[&1i32]).await.unwrap().try_collect::<Vec<_>>().await.unwrap();
-    runtime.block_on(execute).unwrap();
+    assert_eq!(rows[0].get::<_, i32>(0), 1i32);
-    */
 }
 
 #[tokio::test]

tokio-postgres/tests/test/types/bit_vec_07.rs

@@ -2,28 +2,30 @@ use bit_vec_07::BitVec;
 
 use crate::types::test_type;
 
-#[test]
+#[tokio::test]
-fn test_bit_params() {
+async fn test_bit_params() {
     let mut bv = BitVec::from_bytes(&[0b0110_1001, 0b0000_0111]);
     bv.pop();
     bv.pop();
     test_type(
         "BIT(14)",
-        &[(Some(bv), "B'01101001000001'"), (None, "NULL")],
+        vec![(Some(bv), "B'01101001000001'"), (None, "NULL")],
     )
+    .await
 }
 
-#[test]
+#[tokio::test]
-fn test_varbit_params() {
+async fn test_varbit_params() {
     let mut bv = BitVec::from_bytes(&[0b0110_1001, 0b0000_0111]);
     bv.pop();
     bv.pop();
     test_type(
         "VARBIT",
-        &[
+        vec![
             (Some(bv), "B'01101001000001'"),
             (Some(BitVec::from_bytes(&[])), "B''"),
             (None, "NULL"),
         ],
     )
+    .await
 }

tokio-postgres/tests/test/types/chrono_04.rs

@@ -3,8 +3,8 @@ use tokio_postgres::types::{Date, Timestamp};
 
 use crate::types::test_type;
 
-#[test]
+#[tokio::test]
-fn test_naive_date_time_params() {
+async fn test_naive_date_time_params() {
     fn make_check(time: &str) -> (Option<NaiveDateTime>, &str) {
         (
             Some(NaiveDateTime::parse_from_str(time, "'%Y-%m-%d %H:%M:%S.%f'").unwrap()),
@@ -13,17 +13,17 @@ fn test_naive_date_time_params() {
     }
     test_type(
         "TIMESTAMP",
-        &[
+        vec![
             make_check("'1970-01-01 00:00:00.010000000'"),
             make_check("'1965-09-25 11:19:33.100314000'"),
             make_check("'2010-02-09 23:11:45.120200000'"),
             (None, "NULL"),
         ],
-    );
+    ).await;
 }
 
-#[test]
+#[tokio::test]
-fn test_with_special_naive_date_time_params() {
+async fn test_with_special_naive_date_time_params() {
     fn make_check(time: &str) -> (Timestamp<NaiveDateTime>, &str) {
         (
             Timestamp::Value(
@@ -34,18 +34,19 @@ fn test_with_special_naive_date_time_params() {
     }
     test_type(
         "TIMESTAMP",
-        &[
+        vec![
             make_check("'1970-01-01 00:00:00.010000000'"),
             make_check("'1965-09-25 11:19:33.100314000'"),
             make_check("'2010-02-09 23:11:45.120200000'"),
             (Timestamp::PosInfinity, "'infinity'"),
             (Timestamp::NegInfinity, "'-infinity'"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_date_time_params() {
+async fn test_date_time_params() {
     fn make_check(time: &str) -> (Option<DateTime<Utc>>, &str) {
         (
             Some(
@@ -57,17 +58,18 @@ fn test_date_time_params() {
     }
     test_type(
         "TIMESTAMP WITH TIME ZONE",
-        &[
+        vec![
             make_check("'1970-01-01 00:00:00.010000000'"),
             make_check("'1965-09-25 11:19:33.100314000'"),
             make_check("'2010-02-09 23:11:45.120200000'"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_with_special_date_time_params() {
+async fn test_with_special_date_time_params() {
     fn make_check(time: &str) -> (Timestamp<DateTime<Utc>>, &str) {
         (
             Timestamp::Value(
@@ -79,18 +81,19 @@ fn test_with_special_date_time_params() {
     }
     test_type(
         "TIMESTAMP WITH TIME ZONE",
-        &[
+        vec![
             make_check("'1970-01-01 00:00:00.010000000'"),
             make_check("'1965-09-25 11:19:33.100314000'"),
             make_check("'2010-02-09 23:11:45.120200000'"),
             (Timestamp::PosInfinity, "'infinity'"),
             (Timestamp::NegInfinity, "'-infinity'"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_date_params() {
+async fn test_date_params() {
     fn make_check(time: &str) -> (Option<NaiveDate>, &str) {
         (
             Some(NaiveDate::parse_from_str(time, "'%Y-%m-%d'").unwrap()),
@@ -99,17 +102,18 @@ fn test_date_params() {
     }
     test_type(
         "DATE",
-        &[
+        vec![
             make_check("'1970-01-01'"),
             make_check("'1965-09-25'"),
             make_check("'2010-02-09'"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_with_special_date_params() {
+async fn test_with_special_date_params() {
     fn make_check(date: &str) -> (Date<NaiveDate>, &str) {
         (
             Date::Value(NaiveDate::parse_from_str(date, "'%Y-%m-%d'").unwrap()),
@@ -118,18 +122,19 @@ fn test_with_special_date_params() {
     }
     test_type(
         "DATE",
-        &[
+        vec![
             make_check("'1970-01-01'"),
             make_check("'1965-09-25'"),
             make_check("'2010-02-09'"),
             (Date::PosInfinity, "'infinity'"),
             (Date::NegInfinity, "'-infinity'"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_time_params() {
+async fn test_time_params() {
     fn make_check(time: &str) -> (Option<NaiveTime>, &str) {
         (
             Some(NaiveTime::parse_from_str(time, "'%H:%M:%S.%f'").unwrap()),
@@ -138,11 +143,12 @@ fn test_time_params() {
     }
     test_type(
         "TIME",
-        &[
+        vec![
             make_check("'00:00:00.010000000'"),
             make_check("'11:19:33.100314000'"),
             make_check("'23:11:45.120200000'"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }

tokio-postgres/tests/test/types/eui48_04.rs

@@ -2,11 +2,11 @@ use eui48_04::MacAddress;
 
 use crate::types::test_type;
 
-#[test]
+#[tokio::test]
-fn test_eui48_params() {
+async fn test_eui48_params() {
     test_type(
         "MACADDR",
-        &[
+        vec![
             (
                 Some(MacAddress::parse_str("12-34-56-AB-CD-EF").unwrap()),
                 "'12-34-56-ab-cd-ef'",
@@ -14,4 +14,5 @@ fn test_eui48_params() {
             (None, "NULL"),
         ],
     )
+    .await
 }

tokio-postgres/tests/test/types/geo_010.rs

@@ -2,23 +2,24 @@ use geo_010::{Coordinate, LineString, Point, Rect};
 
 use crate::types::test_type;
 
-#[test]
+#[tokio::test]
-fn test_point_params() {
+async fn test_point_params() {
     test_type(
         "POINT",
-        &[
+        vec![
             (Some(Point::new(0.0, 0.0)), "POINT(0, 0)"),
             (Some(Point::new(-3.14, 1.618)), "POINT(-3.14, 1.618)"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_box_params() {
+async fn test_box_params() {
     test_type(
         "BOX",
-        &[
+        vec![
             (
                 Some(Rect {
                     min: Coordinate { x: -3.14, y: 1.618 },
@@ -31,11 +32,12 @@ fn test_box_params() {
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_path_params() {
+async fn test_path_params() {
     let points = vec![
         Coordinate { x: 0., y: 0. },
         Coordinate { x: -3.14, y: 1.618 },
@@ -46,12 +48,13 @@ fn test_path_params() {
     ];
     test_type(
         "PATH",
-        &[
+        vec![
             (
                 Some(LineString(points)),
                 "path '((0, 0), (-3.14, 1.618), (160.0, 69701.5615))'",
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }

tokio-postgres/tests/test/types/mod.rs

@@ -1,4 +1,4 @@
-use futures::{Future, Stream};
+use futures::{TryStreamExt};
 use std::collections::HashMap;
 use std::error::Error;
 use std::f32;
@@ -7,7 +7,6 @@ use std::fmt;
 use std::net::IpAddr;
 use std::result;
 use std::time::{Duration, UNIX_EPOCH};
-use tokio::runtime::current_thread::Runtime;
 use tokio_postgres::to_sql_checked;
 use tokio_postgres::types::{FromSql, FromSqlOwned, IsNull, Kind, ToSql, Type, WrongType};
 
@@ -26,53 +25,63 @@ mod serde_json_1;
 #[cfg(feature = "with-uuid-0_7")]
 mod uuid_07;
 
-fn test_type<T, S>(sql_type: &str, checks: &[(T, S)])
+async fn test_type<T, S>(sql_type: &str, checks: Vec<(T, S)>)
 where
     T: PartialEq + for<'a> FromSqlOwned + ToSql,
     S: fmt::Display,
 {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    for (val, repr) in checks {
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        let stmt = client
-    let connection = connection.map_err(|e| panic!("{}", e));
+            .prepare(&format!("SELECT {}::{}", repr, sql_type))
-    runtime.spawn(connection);
+            .await
-
+            .unwrap();
-    for &(ref val, ref repr) in checks.iter() {
+        let rows = client
-        let prepare = client.prepare(&format!("SELECT {}::{}", repr, sql_type));
+            .query(&stmt, &[])
-        let stmt = runtime.block_on(prepare).unwrap();
+            .await
-        let query = client.query(&stmt, &[]).collect();
+            .unwrap()
-        let rows = runtime.block_on(query).unwrap();
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
         let result = rows[0].get(0);
-        assert_eq!(val, &result);
+        assert_eq!(val, result);
 
-        let prepare = client.prepare(&format!("SELECT $1::{}", sql_type));
+        let stmt = client
-        let stmt = runtime.block_on(prepare).unwrap();
+            .prepare(&format!("SELECT $1::{}", sql_type))
-        let query = client.query(&stmt, &[val]).collect();
+            .await
-        let rows = runtime.block_on(query).unwrap();
+            .unwrap();
+        let rows = client
+            .query(&stmt, &[&val])
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
         let result = rows[0].get(0);
-        assert_eq!(val, &result);
+        assert_eq!(val, result);
     }
 }
 
-#[test]
+#[tokio::test]
-fn test_bool_params() {
+async fn test_bool_params() {
     test_type(
         "BOOL",
-        &[(Some(true), "'t'"), (Some(false), "'f'"), (None, "NULL")],
+        vec![(Some(true), "'t'"), (Some(false), "'f'"), (None, "NULL")],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_i8_params() {
+async fn test_i8_params() {
-    test_type("\"char\"", &[(Some('a' as i8), "'a'"), (None, "NULL")]);
+    test_type("\"char\"", vec![(Some('a' as i8), "'a'"), (None, "NULL")]).await;
 }
 
-#[test]
+#[tokio::test]
-fn test_name_params() {
+async fn test_name_params() {
     test_type(
         "NAME",
-        &[
+        vec![
             (Some("hello world".to_owned()), "'hello world'"),
             (
                 Some("イロハニホヘト チリヌルヲ".to_owned()),
@@ -80,88 +89,95 @@ fn test_name_params() {
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_i16_params() {
+async fn test_i16_params() {
     test_type(
         "SMALLINT",
-        &[
+        vec![
             (Some(15001i16), "15001"),
             (Some(-15001i16), "-15001"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_i32_params() {
+async fn test_i32_params() {
     test_type(
         "INT",
-        &[
+        vec![
             (Some(2_147_483_548i32), "2147483548"),
             (Some(-2_147_483_548i32), "-2147483548"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_oid_params() {
+async fn test_oid_params() {
     test_type(
         "OID",
-        &[
+        vec![
             (Some(2_147_483_548u32), "2147483548"),
             (Some(4_000_000_000), "4000000000"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_i64_params() {
+async fn test_i64_params() {
     test_type(
         "BIGINT",
-        &[
+        vec![
             (Some(9_223_372_036_854_775_708i64), "9223372036854775708"),
             (Some(-9_223_372_036_854_775_708i64), "-9223372036854775708"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_f32_params() {
+async fn test_f32_params() {
     test_type(
         "REAL",
-        &[
+        vec![
             (Some(f32::INFINITY), "'infinity'"),
             (Some(f32::NEG_INFINITY), "'-infinity'"),
             (Some(1000.55), "1000.55"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_f64_params() {
+async fn test_f64_params() {
     test_type(
         "DOUBLE PRECISION",
-        &[
+        vec![
             (Some(f64::INFINITY), "'infinity'"),
             (Some(f64::NEG_INFINITY), "'-infinity'"),
             (Some(10000.55), "10000.55"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_varchar_params() {
+async fn test_varchar_params() {
     test_type(
         "VARCHAR",
-        &[
+        vec![
             (Some("hello world".to_owned()), "'hello world'"),
             (
                 Some("イロハニホヘト チリヌルヲ".to_owned()),
@@ -169,14 +185,15 @@ fn test_varchar_params() {
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_text_params() {
+async fn test_text_params() {
     test_type(
         "TEXT",
-        &[
+        vec![
             (Some("hello world".to_owned()), "'hello world'"),
             (
                 Some("イロハニホヘト チリヌルヲ".to_owned()),
@@ -184,138 +201,149 @@ fn test_text_params() {
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_borrowed_text() {
+async fn test_borrowed_text() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client.prepare("SELECT 'foo'").await.unwrap();
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+    let rows = client
-    let connection = connection.map_err(|e| panic!("{}", e));
+        .query(&stmt, &[])
-    runtime.spawn(connection);
+        .await
-
+        .unwrap()
-    let prepare = client.prepare("SELECT 'foo'");
+        .try_collect::<Vec<_>>()
-    let stmt = runtime.block_on(prepare).unwrap();
+        .await
-    let query = client.query(&stmt, &[]).collect();
+        .unwrap();
-    let rows = runtime.block_on(query).unwrap();
     let s: &str = rows[0].get(0);
     assert_eq!(s, "foo");
 }
 
-#[test]
+#[tokio::test]
-fn test_bpchar_params() {
+async fn test_bpchar_params() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare(
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
-
-    let batch = client
-        .simple_query(
             "CREATE TEMPORARY TABLE foo (
                 id SERIAL PRIMARY KEY,
                 b CHAR(5)
             )",
         )
-        .for_each(|_| Ok(()));
+        .await
-    runtime.block_on(batch).unwrap();
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let prepare = client.prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)")
-    let execute = client.execute(&stmt, &[&"12345", &"123", &None::<&'static str>]);
+        .await
-    runtime.block_on(execute).unwrap();
+        .unwrap();
+    client
+        .execute(&stmt, &[&"12345", &"123", &None::<&'static str>])
+        .await
+        .unwrap();
 
-    let prepare = client.prepare("SELECT b FROM foo ORDER BY id");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("SELECT b FROM foo ORDER BY id")
-    let query = client.query(&stmt, &[]).collect();
+        .await
-    let res = runtime.block_on(query).unwrap();
+        .unwrap();
+    let rows = client
+        .query(&stmt, &[])
+        .await
+        .unwrap()
+        .map_ok(|row| row.get(0))
+        .try_collect::<Vec<Option<String>>>()
+        .await
+        .unwrap();
 
     assert_eq!(
         vec![Some("12345".to_owned()), Some("123  ".to_owned()), None],
-        res.iter().map(|row| row.get(0)).collect::<Vec<_>>()
+        rows,
     );
 }
 
-#[test]
+#[tokio::test]
-fn test_citext_params() {
+async fn test_citext_params() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare(
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
-
-    let batch = client
-        .simple_query(
             "CREATE TEMPORARY TABLE foo (
                 id SERIAL PRIMARY KEY,
                 b CITEXT
             )",
         )
-        .for_each(|_| Ok(()));
+        .await
-    runtime.block_on(batch).unwrap();
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let prepare = client.prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)")
-    let execute = client.execute(&stmt, &[&"foobar", &"FooBar", &None::<&'static str>]);
+        .await
-    runtime.block_on(execute).unwrap();
+        .unwrap();
+    client
+        .execute(&stmt, &[&"foobar", &"FooBar", &None::<&'static str>])
+        .await
+        .unwrap();
 
-    let prepare = client.prepare("SELECT b FROM foo WHERE b = 'FOOBAR' ORDER BY id");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("SELECT b FROM foo WHERE b = 'FOOBAR' ORDER BY id")
-    let query = client.query(&stmt, &[]).collect();
+        .await
-    let res = runtime.block_on(query).unwrap();
+        .unwrap();
+    let rows = client
+        .query(&stmt, &[])
+        .await
+        .unwrap()
+        .map_ok(|row| row.get(0))
+        .try_collect::<Vec<String>>()
+        .await
+        .unwrap();
 
-    assert_eq!(
+    assert_eq!(vec!["foobar".to_string(), "FooBar".to_string()], rows,);
-        vec!["foobar".to_string(), "FooBar".to_string()],
-        res.iter()
-            .map(|row| row.get::<_, String>(0))
-            .collect::<Vec<_>>()
-    );
 }
 
-#[test]
+#[tokio::test]
-fn test_bytea_params() {
+async fn test_bytea_params() {
     test_type(
         "BYTEA",
-        &[
+        vec![
             (Some(vec![0u8, 1, 2, 3, 254, 255]), "'\\x00010203feff'"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_borrowed_bytea() {
+async fn test_borrowed_bytea() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
-
+    let stmt = client.prepare("SELECT 'foo'::BYTEA").await.unwrap();
-    let handshake = connect("user=postgres");
+    let rows = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .query(&stmt, &[])
-    let connection = connection.map_err(|e| panic!("{}", e));
+        .await
-    runtime.spawn(connection);
+        .unwrap()
-
+        .try_collect::<Vec<_>>()
-    let prepare = client.prepare("SELECT 'foo'::BYTEA");
+        .await
-    let stmt = runtime.block_on(prepare).unwrap();
+        .unwrap();
-    let query = client.query(&stmt, &[]).collect();
-    let rows = runtime.block_on(query).unwrap();
     let s: &[u8] = rows[0].get(0);
     assert_eq!(s, b"foo");
 }
 
-#[test]
+macro_rules! make_map {
-fn test_hstore_params() {
+    ($($k:expr => $v:expr),+) => ({
-    macro_rules! make_map {
+        let mut map = HashMap::new();
-        ($($k:expr => $v:expr),+) => ({
+        $(map.insert($k, $v);)+
-            let mut map = HashMap::new();
+        map
-            $(map.insert($k, $v);)+
+    })
-            map
+}
-        })
+
-    }
+#[tokio::test]
+async fn test_hstore_params() {
     test_type(
         "hstore",
-        &[
+        vec![
             (
                 Some(make_map!("a".to_owned() => Some("1".to_owned()))),
                 "'a=>1'",
@@ -328,149 +356,149 @@ fn test_hstore_params() {
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn test_array_params() {
+async fn test_array_params() {
     test_type(
         "integer[]",
-        &[
+        vec![
             (Some(vec![1i32, 2i32]), "ARRAY[1,2]"),
             (Some(vec![1i32]), "ARRAY[1]"),
             (Some(vec![]), "ARRAY[]"),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
 #[allow(clippy::eq_op)]
-fn test_nan_param<T>(sql_type: &str)
+async fn test_nan_param<T>(sql_type: &str)
 where
     T: PartialEq + ToSql + FromSqlOwned,
 {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare(&format!("SELECT 'NaN'::{}", sql_type))
-    let connection = connection.map_err(|e| panic!("{}", e));
+        .await
-    runtime.spawn(connection);
+        .unwrap();
-
+    let rows = client
-    let prepare = client.prepare(&format!("SELECT 'NaN'::{}", sql_type));
+        .query(&stmt, &[])
-    let stmt = runtime.block_on(prepare).unwrap();
+        .await
-    let query = client.query(&stmt, &[]).collect();
+        .unwrap()
-    let rows = runtime.block_on(query).unwrap();
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
     let val: T = rows[0].get(0);
     assert!(val != val);
 }
 
-#[test]
+#[tokio::test]
-fn test_f32_nan_param() {
+async fn test_f32_nan_param() {
-    test_nan_param::<f32>("REAL");
+    test_nan_param::<f32>("REAL").await;
 }
 
-#[test]
+#[tokio::test]
-fn test_f64_nan_param() {
+async fn test_f64_nan_param() {
-    test_nan_param::<f64>("DOUBLE PRECISION");
+    test_nan_param::<f64>("DOUBLE PRECISION").await;
 }
 
-#[test]
+#[tokio::test]
-fn test_pg_database_datname() {
+async fn test_pg_database_datname() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
-
+    let stmt = client
-    let handshake = connect("user=postgres");
+        .prepare("SELECT datname FROM pg_database")
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .await
-    let connection = connection.map_err(|e| panic!("{}", e));
+        .unwrap();
-    runtime.spawn(connection);
+    let rows = client
-
+        .query(&stmt, &[])
-    let prepare = client.prepare("SELECT datname FROM pg_database");
+        .await
-    let stmt = runtime.block_on(prepare).unwrap();
+        .unwrap()
-    let query = client.query(&stmt, &[]).collect();
+        .try_collect::<Vec<_>>()
-    let rows = runtime.block_on(query).unwrap();
+        .await
+        .unwrap();
     assert_eq!(rows[0].get::<_, &str>(0), "postgres");
 }
 
-#[test]
+#[tokio::test]
-fn test_slice() {
+async fn test_slice() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare(
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
-
-    let batch = client
-        .simple_query(
             "CREATE TEMPORARY TABLE foo (
-                id SERIAL PRIMARY KEY,
+            id SERIAL PRIMARY KEY,
-                f TEXT
+            f TEXT
-            );
+        )",
-
-            INSERT INTO foo(f) VALUES ('a'), ('b'), ('c'), ('d');",
         )
-        .for_each(|_| Ok(()));
+        .await
-    runtime.block_on(batch).unwrap();
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let prepare = client.prepare("SELECT f FROM foo WHERE id = ANY($1)");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("INSERT INTO foo (f) VALUES ('a'), ('b'), ('c'), ('d')")
-    let query = client
+        .await
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
+
+    let stmt = client
+        .prepare("SELECT f FROM foo WHERE id = ANY($1)")
+        .await
+        .unwrap();
+    let rows = client
         .query(&stmt, &[&&[1i32, 3, 4][..]])
-        .map(|r| r.get::<_, String>(0))
+        .await
-        .collect();
+        .unwrap()
-    let rows = runtime.block_on(query).unwrap();
+        .map_ok(|r| r.get(0))
+        .try_collect::<Vec<String>>()
+        .await
+        .unwrap();
 
     assert_eq!(vec!["a".to_owned(), "c".to_owned(), "d".to_owned()], rows);
 }
 
-#[test]
+#[tokio::test]
-fn test_slice_wrong_type() {
+async fn test_slice_wrong_type() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare(
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
-
-    let batch = client
-        .simple_query(
             "CREATE TEMPORARY TABLE foo (
-                id SERIAL PRIMARY KEY
+            id SERIAL PRIMARY KEY
-            )",
+        )",
         )
-        .for_each(|_| Ok(()));
+        .await
-    runtime.block_on(batch).unwrap();
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let prepare = client.prepare("SELECT * FROM foo WHERE id = ANY($1)");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("SELECT * FROM foo WHERE id = ANY($1)")
-    let query = client.query(&stmt, &[&&[&"hi"][..]]).collect();
+        .await
-    let err = runtime.block_on(query).err().unwrap();
+        .unwrap();
+    let err = client.query(&stmt, &[&&[&"hi"][..]]).await.err().unwrap();
     match err.source() {
         Some(e) if e.is::<WrongType>() => {}
         _ => panic!("Unexpected error {:?}", err),
     };
 }
 
-#[test]
+#[tokio::test]
-fn test_slice_range() {
+async fn test_slice_range() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client.prepare("SELECT $1::INT8RANGE").await.unwrap();
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+    let err = client.query(&stmt, &[&&[&1i64][..]]).await.err().unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
-
-    let prepare = client.prepare("SELECT $1::INT8RANGE");
-    let stmt = runtime.block_on(prepare).unwrap();
-    let query = client.query(&stmt, &[&&[&1i64][..]]).collect();
-    let err = runtime.block_on(query).err().unwrap();
     match err.source() {
         Some(e) if e.is::<WrongType>() => {}
         _ => panic!("Unexpected error {:?}", err),
     };
 }
 
-#[test]
+#[tokio::test]
-fn domain() {
+async fn domain() {
     #[derive(Debug, PartialEq)]
     struct SessionId(Vec<u8>);
 
@@ -509,57 +537,56 @@ fn domain() {
         }
     }
 
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare("CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16)")
-    let connection = connection.map_err(|e| panic!("{}", e));
+        .await
-    runtime.spawn(connection);
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let batch = client
+    let stmt = client
-        .simple_query(
+        .prepare("CREATE TABLE pg_temp.foo (id pg_temp.session_id)")
-            "CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16);
+        .await
-             CREATE TABLE pg_temp.foo (id pg_temp.session_id);",
+        .unwrap();
-        )
+    client.execute(&stmt, &[]).await.unwrap();
-        .for_each(|_| Ok(()));
-    runtime.block_on(batch).unwrap();
 
     let id = SessionId(b"0123456789abcdef".to_vec());
 
-    let prepare = client.prepare("INSERT INTO pg_temp.foo (id) VALUES ($1)");
+    let stmt = client
-    let stmt = runtime.block_on(prepare).unwrap();
+        .prepare("INSERT INTO pg_temp.foo (id) VALUES ($1)")
-    let execute = client.execute(&stmt, &[&id]);
+        .await
-    runtime.block_on(execute).unwrap();
+        .unwrap();
+    client.execute(&stmt, &[&id]).await.unwrap();
 
-    let prepare = client.prepare("SELECT id FROM pg_temp.foo");
+    let stmt = client.prepare("SELECT id FROM pg_temp.foo").await.unwrap();
-    let stmt = runtime.block_on(prepare).unwrap();
+    let rows = client
-    let query = client.query(&stmt, &[]).collect();
+        .query(&stmt, &[])
-    let rows = runtime.block_on(query).unwrap();
+        .await
+        .unwrap()
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
     assert_eq!(id, rows[0].get(0));
 }
 
-#[test]
+#[tokio::test]
-fn composite() {
+async fn composite() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare(
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
-
-    let batch = client
-        .simple_query(
             "CREATE TYPE pg_temp.inventory_item AS (
-                name TEXT,
+            name TEXT,
-                supplier INTEGER,
+            supplier INTEGER,
-                price NUMERIC
+            price NUMERIC
-            )",
+        )",
         )
-        .for_each(|_| Ok(()));
+        .await
-    runtime.block_on(batch).unwrap();
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let prepare = client.prepare("SELECT $1::inventory_item");
+    let stmt = client.prepare("SELECT $1::inventory_item").await.unwrap();
-    let stmt = runtime.block_on(prepare).unwrap();
     let type_ = &stmt.params()[0];
     assert_eq!(type_.name(), "inventory_item");
     match *type_.kind() {
@@ -575,22 +602,17 @@ fn composite() {
     }
 }
 
-#[test]
+#[tokio::test]
-fn enum_() {
+async fn enum_() {
-    let mut runtime = Runtime::new().unwrap();
+    let mut client = connect("user=postgres").await;
 
-    let handshake = connect("user=postgres");
+    let stmt = client
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+        .prepare("CREATE TYPE pg_temp.mood AS ENUM ('sad', 'ok', 'happy')")
-    let connection = connection.map_err(|e| panic!("{}", e));
+        .await
-    runtime.spawn(connection);
+        .unwrap();
+    client.execute(&stmt, &[]).await.unwrap();
 
-    let batch = client
+    let stmt = client.prepare("SELECT $1::mood").await.unwrap();
-        .simple_query("CREATE TYPE pg_temp.mood AS ENUM ('sad', 'ok', 'happy');")
-        .for_each(|_| Ok(()));
-    runtime.block_on(batch).unwrap();
-
-    let prepare = client.prepare("SELECT $1::mood");
-    let stmt = runtime.block_on(prepare).unwrap();
     let type_ = &stmt.params()[0];
     assert_eq!(type_.name(), "mood");
     match *type_.kind() {
@@ -604,11 +626,11 @@ fn enum_() {
     }
 }
 
-#[test]
+#[tokio::test]
-fn system_time() {
+async fn system_time() {
     test_type(
         "TIMESTAMP",
-        &[
+        vec![
             (
                 Some(UNIX_EPOCH + Duration::from_millis(1_010)),
                 "'1970-01-01 00:00:01.01'",
@@ -623,14 +645,15 @@ fn system_time() {
             ),
             (None, "NULL"),
         ],
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
-fn inet() {
+async fn inet() {
     test_type(
         "INET",
-        &[
+        vec![
             (Some("127.0.0.1".parse::<IpAddr>().unwrap()), "'127.0.0.1'"),
             (
                 Some("127.0.0.1".parse::<IpAddr>().unwrap()),
@@ -653,5 +676,6 @@ fn inet() {
                 "'2001:4f8:3:ba:2e0:81ff:fe22:d1f1/128'",
             ),
         ],
-    );
+    )
+    .await;
 }

tokio-postgres/tests/test/types/serde_json_1.rs

@@ -2,11 +2,11 @@ use serde_json_1::Value;
 
 use crate::types::test_type;
 
-#[test]
+#[tokio::test]
-fn test_json_params() {
+async fn test_json_params() {
     test_type(
         "JSON",
-        &[
+        vec![
             (
                 Some(serde_json_1::from_str::<Value>("[10, 11, 12]").unwrap()),
                 "'[10, 11, 12]'",
@@ -18,13 +18,14 @@ fn test_json_params() {
             (None, "NULL"),
         ],
     )
+    .await
 }
 
-#[test]
+#[tokio::test]
-fn test_jsonb_params() {
+async fn test_jsonb_params() {
     test_type(
         "JSONB",
-        &[
+        vec![
             (
                 Some(serde_json_1::from_str::<Value>("[10, 11, 12]").unwrap()),
                 "'[10, 11, 12]'",
@@ -36,4 +37,5 @@ fn test_jsonb_params() {
             (None, "NULL"),
         ],
     )
+    .await
 }

tokio-postgres/tests/test/types/uuid_07.rs

@@ -2,11 +2,11 @@ use uuid_07::Uuid;
 
 use crate::types::test_type;
 
-#[test]
+#[tokio::test]
-fn test_uuid_params() {
+async fn test_uuid_params() {
     test_type(
         "UUID",
-        &[
+        vec![
             (
                 Some(Uuid::parse_str("a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11").unwrap()),
                 "'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'",
@@ -14,4 +14,5 @@ fn test_uuid_params() {
             (None, "NULL"),
         ],
     )
+    .await
 }