Implement reporting of connection errors

Author: tailhook

examples/client.rs

@@ -17,6 +17,7 @@ use argparse::{ArgumentParser, Store};
 use rotor::{Scope, Time};
 use rotor_http::client::{connect_tcp, Request, Head, Client, RecvMode};
 use rotor_http::client::{Connection, Requester, Task, Version, ResponseError};
+use rotor_http::client::{ProtocolError};
 
 struct Context;
 
@@ -44,6 +45,12 @@ impl Client for Cli {
             }
         }
     }
+    fn connection_error(self, err: &ProtocolError,
+        _scope: &mut Scope<Context>)
+    {
+        writeln!(&mut stderr(), "----- Bad response: {} -----", err).ok();
+        exit(1);
+    }
     fn wakeup(self,
         _connection: &Connection,
         _scope: &mut Scope<<Self::Requester as Requester>::Context>)

src/client/error.rs

@@ -1,4 +1,7 @@
+use std::io;
+
 use httparse;
+use rotor_stream;
 
 
 quick_error!{
@@ -19,5 +22,48 @@ quick_error!{
             from()
             description("error parsing chunk size")
         }
+        Connection(err: ProtocolError) {
+            from()
+            description("connection error")
+            display("connection error: {}", err)
+        }
+    }
+}
+
+quick_error!{
+    /// Error
+    #[derive(Debug)]
+    pub enum ProtocolError {
+        /// Error when connecting
+        ConnectionError(err: io::Error) {
+            description("connection error")
+            display("connection error: {}", err)
+        }
+        /// Error on idle connection
+        ConnectionClosed {
+            description("connection closed")
+            display("connection closed")
+        }
+        ReadError(err: io::Error) {
+            description("error when reading from stream")
+            display("read error: {}", err)
+        }
+        WriteError(err: io::Error) {
+            description("error when writing to stream")
+            display("write error: {}", err)
+        }
+    }
+}
+
+impl From<rotor_stream::Exception> for ProtocolError {
+    fn from(e: rotor_stream::Exception) -> ProtocolError {
+        use rotor_stream::Exception as S;
+        use self::ProtocolError as D;
+        match e {
+            S::EndOfStream => D::ConnectionClosed,
+            S::LimitReached => unreachable!(),
+            S::ReadError(e) => D::ReadError(e),
+            S::WriteError(e) => D::WriteError(e),
+        }
     }
 }

src/client/mod.rs

@@ -24,7 +24,7 @@ pub use version::Version;
 pub use self::request::{Request};
 pub use self::protocol::{Client, Requester, Task};
 pub use self::head::Head;
-pub use self::error::ResponseError;
+pub use self::error::{ResponseError, ProtocolError};
 pub use recvmode::RecvMode;
 
 use self::parser::Parser;

src/client/parser.rs

@@ -4,7 +4,7 @@ use std::cmp::min;
 use std::fmt;
 
 use rotor::{Scope, Time};
-use rotor_stream::{Protocol, StreamSocket};
+use rotor_stream::{Protocol, StreamSocket, Exception};
 use rotor_stream::{Intent, Expectation as E, Transport};
 use rotor_stream::Buf;
 use httparse;
@@ -508,6 +508,27 @@ impl<M, S> Protocol for Parser<M, S>
             }
         }
     }
+    fn exception(self, _transport: &mut Transport<Self::Socket>,
+        reason: Exception, scope: &mut Scope<Self::Context>)
+        -> Intent<Self>
+    {
+        use self::ParserImpl::*;
+        let mut reason = reason.into();
+        match self.1 {
+            ReadHeaders { machine, .. } | Response { machine, .. } => {
+                let err = ResponseError::Connection(reason);
+                machine.bad_response(&err, scope);
+                reason = if let ResponseError::Connection(r) = err {
+                    r
+                } else {
+                    unreachable!();
+                }
+            }
+            _ => {}
+        }
+        self.0.connection_error(&reason, scope);
+        Intent::done()
+    }
     fn timeout(self, transport: &mut Transport<Self::Socket>,
         scope: &mut Scope<Self::Context>)
         -> Intent<Self>
@@ -556,7 +577,7 @@ mod test {
     use rotor::{Scope, EventSet, Time, Machine};
     use rotor_test::{MemIo, MockLoop};
     use client::{Client, Requester, Connection, Task, Request, Version};
-    use client::{Head, RecvMode, Fsm, ResponseError};
+    use client::{Head, RecvMode, Fsm, ResponseError, ProtocolError};
 
     #[derive(Debug, Default, PartialEq, Eq)]
     struct Context {
@@ -594,6 +615,11 @@ mod test {
                 Task::Sleep(self, scope.now() + Duration::new(100, 0))
             }
         }
+        fn connection_error(self, _err: &ProtocolError,
+            scope: &mut Scope<Context>)
+        {
+            scope.errors += 1;
+        }
         fn wakeup(self,
             _connection: &Connection,
             _scope: &mut Scope<<Self::Requester as Requester>::Context>)

src/client/protocol.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 use rotor::{Scope, Time};
 
 use recvmode::RecvMode;
-use super::{Head, Request, ResponseError};
+use super::{Head, Request, ResponseError, ProtocolError};
 use super::{Connection};
 
 pub enum Task<M: Client> {
@@ -32,6 +32,11 @@ pub trait Client: Sized {
         scope: &mut Scope<<Self::Requester as Requester>::Context>)
         -> Task<Self>;
 
+    /// Error when establishing connection or connection closed when in idle
+    fn connection_error(self,
+        reason: &ProtocolError,
+        scope: &mut Scope<<Self::Requester as Requester>::Context>);
+
     /// Standard rotor's wakeup handler
     ///
     /// If `connection.is_idle()` you may initiate a new request