Client::poll_ready and send_request may return Connection Errors (#132)

Closes #131

src/proto/error.rs

@@ -1,4 +1,4 @@
-use codec::RecvError;
+use codec::{RecvError, SendError};
 use frame::Reason;
 
 use std::io;
@@ -11,12 +11,13 @@ pub enum Error {
 }
 
 impl Error {
-    pub fn into_connection_recv_error(self) -> RecvError {
-        use self::Error::*;
-
-        match self {
-            Proto(reason) => RecvError::Connection(reason),
-            Io(e) => RecvError::Io(e),
+    /// Clone the error for internal purposes.
+    ///
+    /// `io::Error` is not `Clone`, so we only copy the `ErrorKind`.
+    pub(super) fn shallow_clone(&self) -> Error {
+        match *self {
+            Error::Proto(reason) => Error::Proto(reason),
+            Error::Io(ref io) => Error::Io(io::Error::from(io.kind())),
         }
     }
 }
@@ -32,3 +33,21 @@ impl From<io::Error> for Error {
         Error::Io(src)
     }
 }
+
+impl From<Error> for RecvError {
+    fn from(src: Error) -> RecvError {
+        match src {
+            Error::Proto(reason) => RecvError::Connection(reason),
+            Error::Io(e) => RecvError::Io(e),
+        }
+    }
+}
+
+impl From<Error> for SendError {
+    fn from(src: Error) -> SendError {
+        match src {
+            Error::Proto(reason) => SendError::Connection(reason),
+            Error::Io(e) => SendError::Io(e),
+        }
+    }
+}

src/proto/streams/recv.rs

@@ -337,10 +337,7 @@ where
         self.ensure_can_reserve(frame.promised_id())?;
 
         // Make sure that the stream state is valid
-        store[stream]
-            .state
-            .ensure_recv_open()
-            .map_err(|e| e.into_connection_recv_error())?;
+        store[stream].state.ensure_recv_open()?;
 
         // TODO: Streams in the reserved states do not count towards the concurrency
         // limit. However, it seems like there should be a cap otherwise this

src/proto/streams/streams.rs

@@ -56,6 +56,9 @@ where
 
     /// Task that calls `poll_complete`.
     task: Option<task::Task>,
+
+    /// If the connection errors, a copy is kept for any StreamRefs.
+    conn_error: Option<proto::Error>,
 }
 
 impl<B, P> Streams<B, P>
@@ -71,6 +74,7 @@ where
                     recv: Recv::new(&config),
                     send: Send::new(&config),
                     task: None,
+                    conn_error: None,
                 },
                 store: Store::new(),
             })),
@@ -193,6 +197,8 @@ where
             })
             .unwrap();
 
+        actions.conn_error = Some(err.shallow_clone());
+
         last_processed_id
     }
 
@@ -337,6 +343,7 @@ where
             let mut me = self.inner.lock().unwrap();
             let me = &mut *me;
 
+            me.actions.ensure_no_conn_error()?;
             me.actions.send.ensure_next_stream_id()?;
 
             // The `pending` argument is provided by the `Client`, and holds
@@ -424,6 +431,7 @@ where
         let mut me = self.inner.lock().unwrap();
         let me = &mut *me;
 
+        me.actions.ensure_no_conn_error()?;
         me.actions.send.ensure_next_stream_id()?;
 
         if let Some(key) = key {
@@ -733,4 +741,12 @@ where
             self.recv.ensure_not_idle(id)
         }
     }
+
+    fn ensure_no_conn_error(&self) -> Result<(), proto::Error> {
+        if let Some(ref err) = self.conn_error {
+            Err(err.shallow_clone())
+        } else {
+            Ok(())
+        }
+    }
 }