experiment async copy in with backpressure

postgres/src/client.rs

@@ -5,16 +5,14 @@ use std::{
     sync::{Arc, Mutex},
 };
 
+use xitca_io::bytes::BytesMut;
 use xitca_unsafe_collection::no_hash::NoHashBuilder;
 
 use super::{
     copy::{CopyIn, CopyOut},
     driver::{
         DriverTx,
-        codec::{
-            Response,
-            encode::{self, Encode},
-        },
+        codec::{Response, encode::Encode},
     },
     error::Error,
     query::Query,
@@ -131,6 +129,7 @@ where
 /// [`Driver`]: crate::driver::Driver
 pub struct Client {
     pub(crate) tx: DriverTx,
+    pub(crate) buf: Mutex<BytesMut>,
     pub(crate) cache: Box<ClientCache>,
 }
 
@@ -260,6 +259,7 @@ impl Client {
     pub(crate) fn new(tx: DriverTx, session: Session) -> Self {
         Self {
             tx,
+            buf: Mutex::new(BytesMut::with_capacity(4096)),
             cache: Box::new(ClientCache {
                 session,
                 type_info: Mutex::new(CachedTypeInfo {
@@ -286,7 +286,15 @@ impl Query for Client {
     where
         S: Encode,
     {
-        encode::send_encode_query(&self.tx, stmt)
+        let (res1, buf) = {
+            let mut buf = self.buf.lock().unwrap();
+            let len = buf.len();
+            let res1 = stmt.encode(&mut buf).inspect_err(|_| buf.truncate(len))?;
+            (res1, buf.split())
+        };
+
+        let res2 = self.tx.send(buf)?;
+        Ok((res1, res2))
     }
 }
 

postgres/src/driver/codec.rs

@@ -15,12 +15,15 @@ use crate::{
     types::BorrowToSql,
 };
 
-use super::DriverTx;
+pub struct Request {
+    pub(super) tx: ResponseSender,
+    pub(super) buf: BytesMut,
+}
 
-pub(super) fn request_pair() -> (ResponseSender, Response) {
+pub(super) fn request_pair(buf: BytesMut) -> (Request, Response) {
     let (tx, rx) = unbounded_channel();
     (
-        tx,
+        Request { tx, buf },
         Response {
             rx,
             buf: BytesMut::new(),

postgres/src/driver/codec/encode.rs

@@ -15,7 +15,7 @@ use crate::{
 };
 
 use super::{
-    AsParams, DriverTx, Response,
+    AsParams,
     response::{
         IntoResponse, IntoRowStreamGuard, NoOpIntoRowStream, StatementCreateResponse, StatementCreateResponseBlocking,
     },
@@ -236,13 +236,6 @@ impl<'s> Encode for PortalQuery<'s> {
     }
 }
 
-pub(crate) fn send_encode_query<S>(tx: &DriverTx, stmt: S) -> Result<(S::Output, Response), Error>
-where
-    S: Encode,
-{
-    tx.send(|buf| stmt.encode(buf))
-}
-
 fn encode_bind<P>(stmt: &str, types: &[Type], params: P, portal: &str, buf: &mut BytesMut) -> Result<(), Error>
 where
     P: AsParams,