Start on borrow overhaul

Author: sfackler

tokio-postgres/src/client.rs

@@ -173,50 +173,50 @@ impl Client {
     ///
     /// Prepared statements can be executed repeatedly, and may contain query parameters (indicated by `$1`, `$2`, etc),
     /// which are set when executed. Prepared statements can only be used with the connection that created them.
-    pub fn prepare(&mut self, query: &str) -> impl Future<Output = Result<Statement, Error>> {
-        self.prepare_typed(query, &[])
+    pub async fn prepare(&self, query: &str) -> Result<Statement, Error> {
+        self.prepare_typed(query, &[]).await
     }
 
     /// Like `prepare`, but allows the types of query parameters to be explicitly specified.
     ///
     /// The list of types may be smaller than the number of parameters - the types of the remaining parameters will be
     /// inferred. For example, `client.prepare_typed(query, &[])` is equivalent to `client.prepare(query)`.
-    pub fn prepare_typed(
-        &mut self,
+    pub async fn prepare_typed(
+        &self,
         query: &str,
         parameter_types: &[Type],
-    ) -> impl Future<Output = Result<Statement, Error>> {
-        prepare::prepare(self.inner(), query, parameter_types)
+    ) -> Result<Statement, Error> {
+        prepare::prepare(&self.inner, query, parameter_types).await
     }
 
     /// Executes a statement, returning a stream of the resulting rows.
     ///
     /// # Panics
     ///
     /// Panics if the number of parameters provided does not match the number expected.
-    pub fn query(
-        &mut self,
-        statement: &Statement,
-        params: &[&(dyn ToSql + Sync)],
-    ) -> impl Stream<Item = Result<Row, Error>> {
+    pub fn query<'a>(
+        &'a self,
+        statement: &'a Statement,
+        params: &'a [&'a (dyn ToSql + Sync)],
+    ) -> impl Stream<Item = Result<Row, Error>> + 'a {
         let buf = query::encode(statement, params.iter().map(|s| *s as _));
-        query::query(self.inner(), statement.clone(), buf)
+        query::query(&self.inner, statement, buf)
     }
 
     /// Like [`query`], but takes an iterator of parameters rather than a slice.
     ///
     /// [`query`]: #method.query
     pub fn query_iter<'a, I>(
-        &mut self,
-        statement: &Statement,
+        &'a self,
+        statement: &'a Statement,
         params: I,
-    ) -> impl Stream<Item = Result<Row, Error>>
+    ) -> impl Stream<Item = Result<Row, Error>> + 'a
     where
-        I: IntoIterator<Item = &'a dyn ToSql>,
+        I: IntoIterator<Item = &'a dyn ToSql> + 'a,
         I::IntoIter: ExactSizeIterator,
     {
         let buf = query::encode(statement, params);
-        query::query(self.inner(), statement.clone(), buf)
+        query::query(&self.inner, statement, buf)
     }
 
     /// Executes a statement, returning the number of rows modified.
@@ -226,29 +226,29 @@ impl Client {
     /// # Panics
     ///
     /// Panics if the number of parameters provided does not match the number expected.
-    pub fn execute(
-        &mut self,
+    pub async fn execute(
+        &self,
         statement: &Statement,
         params: &[&(dyn ToSql + Sync)],
-    ) -> impl Future<Output = Result<u64, Error>> {
+    ) -> Result<u64, Error> {
         let buf = query::encode(statement, params.iter().map(|s| *s as _));
-        query::execute(self.inner(), buf)
+        query::execute(&self.inner, buf).await
     }
 
     /// Like [`execute`], but takes an iterator of parameters rather than a slice.
     ///
     /// [`execute`]: #method.execute
-    pub fn execute_iter<'a, I>(
-        &mut self,
+    pub async fn execute_iter<'a, I>(
+        &self,
         statement: &Statement,
         params: I,
-    ) -> impl Future<Output = Result<u64, Error>>
+    ) -> Result<u64, Error>
     where
         I: IntoIterator<Item = &'a dyn ToSql>,
         I::IntoIter: ExactSizeIterator,
     {
         let buf = query::encode(statement, params);
-        query::execute(self.inner(), buf)
+        query::execute(&self.inner, buf).await
     }
 
     /// Executes a `COPY FROM STDIN` statement, returning the number of rows created.
@@ -260,7 +260,7 @@ impl Client {
     ///
     /// Panics if the number of parameters provided does not match the number expected.
     pub fn copy_in<S>(
-        &mut self,
+        &self,
         statement: &Statement,
         params: &[&(dyn ToSql + Sync)],
         stream: S,
@@ -281,7 +281,7 @@ impl Client {
     ///
     /// Panics if the number of parameters provided does not match the number expected.
     pub fn copy_out(
-        &mut self,
+        &self,
         statement: &Statement,
         params: &[&(dyn ToSql + Sync)],
     ) -> impl Stream<Item = Result<Bytes, Error>> {
@@ -303,7 +303,7 @@ impl Client {
     /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
     pub fn simple_query(
-        &mut self,
+        &self,
         query: &str,
     ) -> impl Stream<Item = Result<SimpleQueryMessage, Error>> {
         simple_query::simple_query(self.inner(), query)
@@ -319,7 +319,7 @@ impl Client {
     /// Prepared statements should be use for any query which contains user-specified data, as they provided the
     /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
-    pub fn batch_execute(&mut self, query: &str) -> impl Future<Output = Result<(), Error>> {
+    pub fn batch_execute(&self, query: &str) -> impl Future<Output = Result<(), Error>> {
         simple_query::batch_execute(self.inner(), query)
     }
 
@@ -338,7 +338,7 @@ impl Client {
     ///
     /// Requires the `runtime` Cargo feature (enabled by default).
     #[cfg(feature = "runtime")]
-    pub fn cancel_query<T>(&mut self, tls: T) -> impl Future<Output = Result<(), Error>>
+    pub fn cancel_query<T>(&self, tls: T) -> impl Future<Output = Result<(), Error>>
     where
         T: MakeTlsConnect<Socket>,
     {
@@ -354,7 +354,7 @@ impl Client {
     /// Like `cancel_query`, but uses a stream which is already connected to the server rather than opening a new
     /// connection itself.
     pub fn cancel_query_raw<S, T>(
-        &mut self,
+        &self,
         stream: S,
         tls: T,
     ) -> impl Future<Output = Result<(), Error>>

tokio-postgres/src/lib.rs

@@ -79,14 +79,14 @@
 //! use std::future::Future;
 //! use tokio_postgres::{Client, Error, Statement};
 //!
-//! fn pipelined_prepare(
-//!     client: &mut Client,
-//! ) -> impl Future<Output = Result<(Statement, Statement), Error>>
+//! async fn pipelined_prepare(
+//!     client: &Client,
+//! ) -> Result<(Statement, Statement), Error>
 //! {
 //!     future::try_join(
 //!         client.prepare("SELECT * FROM foo"),
 //!         client.prepare("INSERT INTO bar (id, name) VALUES ($1, $2)")
-//!     )
+//!     ).await
 //! }
 //! ```
 //!

tokio-postgres/src/prepare.rs

@@ -57,15 +57,14 @@ ORDER BY attnum
 
 static NEXT_ID: AtomicUsize = AtomicUsize::new(0);
 
-pub fn prepare(
-    client: Arc<InnerClient>,
+pub async fn prepare(
+    client: &Arc<InnerClient>,
     query: &str,
     types: &[Type],
-) -> impl Future<Output = Result<Statement, Error>> + 'static {
+) -> 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)))?;
 
@@ -103,14 +102,13 @@ pub fn prepare(
         }
 
         Ok(Statement::new(&client, name, parameters, columns))
-    }
 }
 
-fn prepare_rec(
-    client: Arc<InnerClient>,
-    query: &str,
-    types: &[Type],
-) -> Pin<Box<dyn Future<Output = Result<Statement, Error>> + 'static + Send>> {
+fn prepare_rec<'a>(
+    client: &'a Arc<InnerClient>,
+    query: &'a str,
+    types: &'a [Type],
+) -> Pin<Box<dyn Future<Output = Result<Statement, Error>> + 'a + Send>> {
     Box::pin(prepare(client, query, types))
 }
 
@@ -135,7 +133,7 @@ async fn get_type(client: &Arc<InnerClient>, oid: Oid) -> Result<Type, Error> {
     let stmt = typeinfo_statement(client).await?;
 
     let buf = query::encode(&stmt, (&[&oid as &dyn ToSql]).iter().cloned());
-    let rows = query::query(client.clone(), stmt, buf);
+    let rows = query::query(client, &stmt, buf);
     pin_mut!(rows);
 
     let row = match rows.try_next().await? {
@@ -190,10 +188,10 @@ async fn typeinfo_statement(client: &Arc<InnerClient>) -> Result<Statement, Erro
         return Ok(stmt);
     }
 
-    let stmt = match prepare_rec(client.clone(), TYPEINFO_QUERY, &[]).await {
+    let stmt = match prepare_rec(client, TYPEINFO_QUERY, &[]).await {
         Ok(stmt) => stmt,
         Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_TABLE) => {
-            prepare_rec(client.clone(), TYPEINFO_FALLBACK_QUERY, &[]).await?
+            prepare_rec(client, TYPEINFO_FALLBACK_QUERY, &[]).await?
         }
         Err(e) => return Err(e),
     };
@@ -206,7 +204,7 @@ async fn get_enum_variants(client: &Arc<InnerClient>, oid: Oid) -> Result<Vec<St
     let stmt = typeinfo_enum_statement(client).await?;
 
     let buf = query::encode(&stmt, (&[&oid as &dyn ToSql]).iter().cloned());
-    query::query(client.clone(), stmt, buf)
+    query::query(client, &stmt, buf)
         .and_then(|row| future::ready(row.try_get(0)))
         .try_collect()
         .await
@@ -217,10 +215,10 @@ async fn typeinfo_enum_statement(client: &Arc<InnerClient>) -> Result<Statement,
         return Ok(stmt);
     }
 
-    let stmt = match prepare_rec(client.clone(), TYPEINFO_ENUM_QUERY, &[]).await {
+    let stmt = match prepare_rec(client, TYPEINFO_ENUM_QUERY, &[]).await {
         Ok(stmt) => stmt,
         Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_COLUMN) => {
-            prepare_rec(client.clone(), TYPEINFO_ENUM_FALLBACK_QUERY, &[]).await?
+            prepare_rec(client, TYPEINFO_ENUM_FALLBACK_QUERY, &[]).await?
         }
         Err(e) => return Err(e),
     };
@@ -233,7 +231,7 @@ async fn get_composite_fields(client: &Arc<InnerClient>, oid: Oid) -> Result<Vec
     let stmt = typeinfo_composite_statement(client).await?;
 
     let buf = query::encode(&stmt, (&[&oid as &dyn ToSql]).iter().cloned());
-    let rows = query::query(client.clone(), stmt, buf)
+    let rows = query::query(client, &stmt, buf)
         .try_collect::<Vec<_>>()
         .await?;
 
@@ -253,7 +251,7 @@ async fn typeinfo_composite_statement(client: &Arc<InnerClient>) -> Result<State
         return Ok(stmt);
     }
 
-    let stmt = prepare_rec(client.clone(), TYPEINFO_COMPOSITE_QUERY, &[]).await?;
+    let stmt = prepare_rec(client, TYPEINFO_COMPOSITE_QUERY, &[]).await?;
 
     client.set_typeinfo_composite(&stmt);
     Ok(stmt)

tokio-postgres/src/query.rs

@@ -10,17 +10,16 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-pub fn query(
-    client: Arc<InnerClient>,
-    statement: Statement,
+pub fn query<'a>(
+    client: &'a Arc<InnerClient>,
+    statement: &'a Statement,
     buf: Result<Vec<u8>, Error>,
-) -> impl Stream<Item = Result<Row, Error>> {
-    start(client, buf)
-        .map_ok(|responses| Query {
-            statement,
-            responses,
-        })
-        .try_flatten_stream()
+) -> impl Stream<Item = Result<Row, Error>> + 'a {
+    let f = async move {
+        let responses = start(client, buf).await?;
+        Ok(Query { statement: statement.clone(), responses })
+    };
+    f.try_flatten_stream()
 }
 
 pub fn query_portal(
@@ -44,7 +43,7 @@ pub fn query_portal(
     start.try_flatten_stream()
 }
 
-pub async fn execute(client: Arc<InnerClient>, buf: Result<Vec<u8>, Error>) -> Result<u64, Error> {
+pub async fn execute(client: &InnerClient, buf: Result<Vec<u8>, Error>) -> Result<u64, Error> {
     let mut responses = start(client, buf).await?;
 
     loop {
@@ -67,7 +66,7 @@ pub async fn execute(client: Arc<InnerClient>, buf: Result<Vec<u8>, Error>) -> R
     }
 }
 
-async fn start(client: Arc<InnerClient>, buf: Result<Vec<u8>, Error>) -> Result<Responses, Error> {
+async fn start(client: &InnerClient, buf: Result<Vec<u8>, Error>) -> Result<Responses, Error> {
     let buf = buf?;
     let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;