Start prepping for futures closing over parameters

Change the slice-consuming methods to requiring &(dyn ToSql + Sync),
which makes the overall value Send. If you have non-Sync values for
whatever reason, you can still use the iterator-based methods.

Author: sfackler

tokio-postgres/src/client.rs

@@ -197,9 +197,9 @@ impl Client {
     pub fn query(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Stream<Item = Result<Row, Error>> {
-        let buf = query::encode(statement, params.iter().cloned());
+        let buf = query::encode(statement, params.iter().map(|s| *s as _));
         query::query(self.inner(), statement.clone(), buf)
     }
 
@@ -229,9 +229,9 @@ impl Client {
     pub fn execute(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Future<Output = Result<u64, Error>> {
-        let buf = query::encode(statement, params.iter().cloned());
+        let buf = query::encode(statement, params.iter().map(|s| *s as _));
         query::execute(self.inner(), buf)
     }
 
@@ -262,7 +262,7 @@ impl Client {
     pub fn copy_in<S>(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
         stream: S,
     ) -> impl Future<Output = Result<u64, Error>>
     where
@@ -271,7 +271,7 @@ impl Client {
         <S::Ok as IntoBuf>::Buf: 'static + Send,
         S::Error: Into<Box<dyn error::Error + Sync + Send>>,
     {
-        let buf = query::encode(statement, params.iter().cloned());
+        let buf = query::encode(statement, params.iter().map(|s| *s as _));
         copy_in::copy_in(self.inner(), buf, stream)
     }
 
@@ -283,9 +283,9 @@ impl Client {
     pub fn copy_out(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Stream<Item = Result<Bytes, Error>> {
-        let buf = query::encode(statement, params.iter().cloned());
+        let buf = query::encode(statement, params.iter().map(|s| *s as _));
         copy_out::copy_out(self.inner(), buf)
     }
 

tokio-postgres/src/transaction.rs

@@ -95,7 +95,7 @@ impl<'a> Transaction<'a> {
     pub fn query(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Stream<Item = Result<Row, Error>> {
         self.client.query(statement, params)
     }
@@ -119,7 +119,7 @@ impl<'a> Transaction<'a> {
     pub fn execute(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Future<Output = Result<u64, Error>> {
         self.client.execute(statement, params)
     }
@@ -150,10 +150,10 @@ impl<'a> Transaction<'a> {
     pub fn bind(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Future<Output = Result<Portal, Error>> {
         // https://github.com/rust-lang/rust/issues/63032
-        let buf = bind::encode(statement, params.iter().cloned());
+        let buf = bind::encode(statement, params.iter().map(|s| *s as _));
         bind::bind(self.client.inner(), statement.clone(), buf)
     }
 
@@ -189,7 +189,7 @@ impl<'a> Transaction<'a> {
     pub fn copy_in<S>(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
         stream: S,
     ) -> impl Future<Output = Result<u64, Error>>
     where
@@ -205,7 +205,7 @@ impl<'a> Transaction<'a> {
     pub fn copy_out(
         &mut self,
         statement: &Statement,
-        params: &[&dyn ToSql],
+        params: &[&(dyn ToSql + Sync)],
     ) -> impl Stream<Item = Result<Bytes, Error>> {
         self.client.copy_out(statement, params)
     }

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

@@ -27,7 +27,7 @@ mod uuid_07;
 
 async fn test_type<T, S>(sql_type: &str, checks: &[(T, S)])
 where
-    T: PartialEq + for<'a> FromSqlOwned + ToSql,
+    T: PartialEq + for<'a> FromSqlOwned + ToSql + Sync,
     S: fmt::Display,
 {
     let mut client = connect("user=postgres").await;
@@ -656,3 +656,30 @@ async fn inet() {
     )
     .await;
 }
+
+#[tokio::test]
+async fn check_send() {
+    fn is_send<T: Send>(_: &T) {}
+
+    let mut client = connect("user=postgres").await;
+
+    let f = client.prepare("SELECT $1::TEXT");
+    is_send(&f);
+    let stmt = f.await.unwrap();
+
+    let f = client.query(&stmt, &[&"hello"]);
+    is_send(&f);
+
+    let f = client.execute(&stmt, &[&"hello"]);
+    is_send(&f);
+
+    let f = client.transaction();
+    is_send(&f);
+    let mut trans = f.await.unwrap();
+
+    let f = trans.query(&stmt, &[&"hello"]);
+    is_send(&f);
+
+    let f = trans.execute(&stmt, &[&"hello"]);
+    is_send(&f);
+}