Make request errors more useful (#12)

By making the HttpRequest error type only yield status code type
failures, we can more easily take action on those errors in the SDK. For
other configuration related errors specific to request building, we have
introduced a new enum variant -- InvalidParameter.

Author: keelerm84

src/client.rs

@@ -42,7 +42,9 @@ pub struct ClientBuilder {
 impl ClientBuilder {
     /// Set a HTTP header on the SSE request.
     pub fn header(mut self, key: &'static str, value: &str) -> Result<ClientBuilder> {
-        let value = value.parse().map_err(|e| Error::HttpRequest(Box::new(e)))?;
+        let value = value
+            .parse()
+            .map_err(|e| Error::InvalidParameter(Box::new(e)))?;
         self.headers.insert(key, value);
         Ok(self)
     }
@@ -113,7 +115,9 @@ impl Client<()> {
     /// [`ClientBuilder`]: struct.ClientBuilder.html
     /// [`.stream()`]: #method.stream
     pub fn for_url(url: &str) -> Result<ClientBuilder> {
-        let url = url.parse().map_err(|e| Error::HttpRequest(Box::new(e)))?;
+        let url = url
+            .parse()
+            .map_err(|e| Error::InvalidParameter(Box::new(e)))?;
         Ok(ClientBuilder {
             url,
             headers: HeaderMap::new(),
@@ -205,7 +209,7 @@ impl<C> ReconnectingRequest<C> {
         *request.headers_mut().unwrap() = self.props.headers.clone();
         let request = request
             .body(Body::empty())
-            .map_err(|e| Error::HttpRequest(Box::new(e)))?;
+            .map_err(|e| Error::InvalidParameter(Box::new(e)))?;
         Ok(self.http.request(request))
     }
 
@@ -257,40 +261,50 @@ where
         loop {
             trace!("ReconnectingRequest::poll loop({:?})", &self.state);
 
-            let state = self.as_mut().project().state.project();
-            let new_state = match state {
+            let this = self.as_mut().project();
+            let state = this.state.project();
+
+            match state {
                 // New immediately transitions to Connecting, and exists only
                 // to ensure that we only connect when polled.
                 StateProj::New => {
                     let resp = match self.send_request() {
                         Err(e) => return Poll::Ready(Some(Err(e))),
                         Ok(r) => r,
                     };
-                    State::Connecting {
-                        resp,
-                        retry: self.props.reconnect_opts.retry_initial,
-                    }
+                    let retry = self.props.reconnect_opts.retry_initial;
+                    self.as_mut()
+                        .project()
+                        .state
+                        .set(State::Connecting { resp, retry })
                 }
                 StateProj::Connecting { retry, resp } => match ready!(resp.poll(cx)) {
                     Ok(resp) => {
                         debug!("HTTP response: {:#?}", resp);
 
                         if !resp.status().is_success() {
-                            let e = StatusError {
-                                status: resp.status(),
-                            };
-                            return Poll::Ready(Some(Err(Error::HttpRequest(Box::new(e)))));
+                            self.as_mut().project().state.set(State::New);
+                            return Poll::Ready(Some(Err(Error::HttpRequest(resp.status()))));
                         }
 
                         self.as_mut().reset_backoff();
-                        State::Connected(resp.into_body())
+                        self.as_mut()
+                            .project()
+                            .state
+                            .set(State::Connected(resp.into_body()))
                     }
                     Err(e) => {
                         warn!("request returned an error: {}", e);
                         if !*retry {
+                            self.as_mut().project().state.set(State::New);
                             return Poll::Ready(Some(Err(Error::HttpStream(Box::new(e)))));
                         }
-                        State::WaitingToReconnect(delay(self.as_mut().backoff(), "retrying"))
+
+                        let duration = self.as_mut().backoff();
+                        self.as_mut()
+                            .project()
+                            .state
+                            .set(State::WaitingToReconnect(delay(duration, "retrying")))
                     }
                 },
                 StateProj::Connected(body) => match ready!(body.poll_data(cx)) {
@@ -300,10 +314,11 @@ where
                     res => {
                         // reconnect
                         if self.props.reconnect_opts.reconnect {
-                            State::WaitingToReconnect(delay(
-                                self.as_mut().backoff(),
-                                "reconnecting",
-                            ))
+                            let duration = self.as_mut().backoff();
+                            self.as_mut()
+                                .project()
+                                .state
+                                .set(State::WaitingToReconnect(delay(duration, "reconnecting")))
                         } else {
                             return Poll::Ready(
                                 res.map(|r| r.map_err(|e| Error::HttpStream(Box::new(e)))),
@@ -315,10 +330,12 @@ where
                     ready!(delay.poll(cx));
                     info!("Reconnecting");
                     let resp = self.send_request()?;
-                    State::Connecting { retry: true, resp }
+                    self.as_mut()
+                        .project()
+                        .state
+                        .set(State::Connecting { retry: true, resp })
                 }
             };
-            self.as_mut().project().state.set(new_state);
         }
     }
 }

src/decode.rs

@@ -569,7 +569,7 @@ mod tests {
             Ready(Some(Err(err))) => {
                 assert!(err.is_http_stream_error());
                 let description = format!("{}", err.source().unwrap());
-                assert!(description.contains("read error"), description);
+                assert!(description.contains("read error"), "{}", description);
             }
             res => panic!("expected HttpStream error, got {:?}", res),
         }
@@ -582,7 +582,7 @@ mod tests {
             Ready(Some(Err(err))) => {
                 assert!(err.is_http_stream_error());
                 let description = format!("{}", err.source().unwrap());
-                assert!(description.contains("read error"), description);
+                assert!(description.contains("read error"), "{}", description);
             }
             res => panic!("expected HttpStream error, got {:?}", res),
         }
@@ -600,7 +600,7 @@ mod tests {
             Ready(Some(Err(err))) => {
                 assert!(err.is_http_stream_error());
                 let description = format!("{}", err.source().unwrap());
-                assert!(description.contains("read error"), description);
+                assert!(description.contains("read error"), "{}", description);
             }
             res => panic!("expected HttpStream error, got {:?}", res),
         }
@@ -687,7 +687,7 @@ mod tests {
         let mut cx = Context::from_waker(&waker);
         match s.poll_next_unpin(&mut cx) {
             Poll::Ready(None) => (),
-            Poll::Ready(Some(event)) => panic!(format!("expected eof, got {:?}", event)),
+            Poll::Ready(Some(event)) => panic!("expected eof, got {:?}", event),
             Poll::Pending => panic!("expected eof, got Pending"),
         }
     }
@@ -700,7 +700,7 @@ mod tests {
         let mut cx = Context::from_waker(&waker);
         match s.poll_next_unpin(&mut cx) {
             Poll::Ready(Some(Ok(event))) => event,
-            Poll::Ready(Some(Err(e))) => panic!(format!("expected eof, got error: {:?}", e)),
+            Poll::Ready(Some(Err(e))) => panic!("expected eof, got error: {:?}", e),
             Poll::Ready(None) => panic!("expected event, got eof"),
             Poll::Pending => panic!("expected event, got Pending"),
         }

src/error.rs

@@ -1,8 +1,12 @@
+use hyper::StatusCode;
+
 /// Error type returned from this library's functions.
 #[derive(Debug)]
 pub enum Error {
+    /// An invalid request parameter
+    InvalidParameter(Box<dyn std::error::Error + Send + 'static>),
     /// The HTTP request failed.
-    HttpRequest(Box<dyn std::error::Error + Send + 'static>),
+    HttpRequest(StatusCode),
     /// An error reading from the HTTP response body.
     HttpStream(Box<dyn std::error::Error + Send + 'static>),
     /// The HTTP response stream ended unexpectedly (e.g. in the
@@ -38,7 +42,6 @@ impl Error {
 
     pub fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
-            Error::HttpRequest(err) => Some(err.as_ref()),
             Error::HttpStream(err) => Some(err.as_ref()),
             Error::Unexpected(err) => Some(err.as_ref()),
             _ => None,