add error type for driver down when receiving response.

http/src/h2/proto/hpack/table.rs

@@ -1,10 +1,11 @@
-use std::{
+use core::{
     cmp,
-    collections::VecDeque,
     hash::{Hash, Hasher},
-    mem, usize,
+    mem,
 };
 
+use std::collections::VecDeque;
+
 use fnv::FnvHasher;
 
 use crate::http::{header, method::Method};

postgres/src/driver/codec.rs

@@ -7,7 +7,7 @@ use postgres_protocol::message::backend;
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use xitca_io::bytes::BytesMut;
 
-use crate::error::{DriverDown, Error};
+use crate::error::{DriverDownReceiving, Error};
 
 pub struct Response {
     rx: ResponseReceiver,
@@ -22,15 +22,14 @@ impl Response {
         }
     }
 
-    pub(crate) fn recv(&mut self) -> impl Future<Output = Result<backend::Message, Error>> + '_ {
+    pub(crate) fn recv(&mut self) -> impl Future<Output = Result<backend::Message, Error>> + Send + '_ {
         poll_fn(|cx| {
             if self.buf.is_empty() {
-                self.buf = ready!(self.rx.poll_recv(cx)).ok_or_else(|| DriverDown(BytesMut::new()))?;
+                self.buf = ready!(self.rx.poll_recv(cx)).ok_or_else(|| DriverDownReceiving)?;
             }
 
             let res = match backend::Message::parse(&mut self.buf)?.expect("must not parse message from empty buffer.")
             {
-                // TODO: error response.
                 backend::Message::ErrorResponse(_body) => Err(Error::todo()),
                 msg => Ok(msg),
             };
@@ -46,23 +45,30 @@ pub(crate) struct ResponseSender {
     msg_count: usize,
 }
 
+pub(super) enum SenderState {
+    Continue,
+    Finish,
+}
+
 impl ResponseSender {
     fn new(tx: UnboundedSender<BytesMut>, msg_count: usize) -> Self {
         Self { tx, msg_count }
     }
 
-    pub(super) fn send(&mut self, msg: BytesMut) {
-        let _ = self.tx.send(msg);
-    }
+    pub(super) fn send(&mut self, msg: BytesMut, complete: bool) -> SenderState {
+        debug_assert!(self.msg_count > 0);
 
-    pub(super) fn complete(&mut self, current_msg_complete: bool) -> bool {
-        assert!(self.msg_count > 0);
+        let _ = self.tx.send(msg);
 
-        if current_msg_complete {
+        if complete {
             self.msg_count -= 1;
         }
 
-        self.msg_count == 0
+        if self.msg_count == 0 {
+            SenderState::Finish
+        } else {
+            SenderState::Continue
+        }
     }
 }
 

postgres/src/driver/generic.rs

@@ -17,7 +17,7 @@ use xitca_unsafe_collection::futures::{Select as _, SelectOutput};
 use crate::{error::Error, iter::AsyncLendingIterator};
 
 use super::{
-    codec::{Request, Response, ResponseMessage, ResponseSender},
+    codec::{Request, Response, ResponseMessage, ResponseSender, SenderState},
     Drive,
 };
 
@@ -183,11 +183,12 @@ where
         while let Some(res) = ResponseMessage::try_from_buf(self.read_buf.get_mut())? {
             match res {
                 ResponseMessage::Normal { buf, complete } => {
-                    if let Some(front) = self.res.front_mut() {
-                        front.send(buf);
-                        if front.complete(complete) {
+                    let front = self.res.front_mut().expect("server respond out of bound");
+                    match front.send(buf, complete) {
+                        SenderState::Finish => {
                             self.res.pop_front();
                         }
+                        SenderState::Continue => {}
                     }
                 }
                 ResponseMessage::Async(msg) => return Ok(Some(msg)),

postgres/src/driver/io_uring.rs

@@ -16,7 +16,7 @@ use xitca_unsafe_collection::futures::{ReusableLocalBoxFuture, Select, SelectOut
 use crate::error::Error;
 
 use super::{
-    codec::{ResponseMessage, ResponseSender},
+    codec::{ResponseMessage, ResponseSender, SenderState},
     generic::GenericDriverRx,
 };
 
@@ -97,10 +97,12 @@ where
                 while let Some(res) = ResponseMessage::try_from_buf(buf)? {
                     match res {
                         ResponseMessage::Normal { buf, complete } => {
-                            let front = self.res.front_mut().expect("out of bound must not happen");
-                            front.send(buf);
-                            if front.complete(complete) {
-                                self.res.pop_front();
+                            let front = self.res.front_mut().expect("server respond out of bound");
+                            match front.send(buf, complete) {
+                                SenderState::Finish => {
+                                    self.res.pop_front();
+                                }
+                                SenderState::Continue => {}
                             }
                         }
                         ResponseMessage::Async(msg) => return Ok(Some(msg)),

postgres/src/error.rs

@@ -76,7 +76,8 @@ impl error::Error for Error {
     }
 }
 
-/// error indicate [Client]'s [Driver] is dropped and can't be accessed anymore.
+/// error indicate [Client]'s [Driver] is dropped and can't be accessed anymore when sending
+/// request to driver.
 ///
 /// the field inside error contains the raw bytes buffer of query message that are ready to be
 /// sent to the [Driver] for transporting. It's possible to construct a new [Client] and [Driver]
@@ -107,6 +108,25 @@ impl From<DriverDown> for Error {
     }
 }
 
+/// error indicate [Client]'s [Driver] is dropped and can't be accessed anymore when receiving response
+/// from server. any mid flight response and unfinished response data are lost and can't be recovered.
+#[derive(Debug)]
+pub struct DriverDownReceiving;
+
+impl fmt::Display for DriverDownReceiving {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("Driver is dropped and unaccessible. Response data is lost and unrecoverable.")
+    }
+}
+
+impl error::Error for DriverDownReceiving {}
+
+impl From<DriverDownReceiving> for Error {
+    fn from(e: DriverDownReceiving) -> Self {
+        Self(Box::new(e))
+    }
+}
+
 pub struct InvalidColumnIndex(pub String);
 
 impl fmt::Debug for InvalidColumnIndex {