Refactor to make compatible with existing client

Author: cwaldren-ld

Cargo.toml

@@ -25,6 +25,7 @@ actix = { version = "0.12.0", optional = true }
 actix-web = { version = "4.0.0-beta.10", optional = true }
 reqwest = { version = "0.11.6", features = ["json"], optional = true }
 env_logger = { version = "0.7.1", optional = true }
+hyper-timeout = "0.4.1"
 
 [dev-dependencies]
 env_logger = "0.7.1"

examples/tail.rs

@@ -36,7 +36,9 @@ async fn main() -> Result<(), es::Error> {
     Ok(())
 }
 
-fn tail_events(client: es::Client<es::HttpsConnector>) -> impl Stream<Item = Result<(), ()>> {
+fn tail_events(
+    client: es::Client<es::TimeoutConnector<es::HttpsConnector>>,
+) -> impl Stream<Item = Result<(), ()>> {
     client
         .stream()
         .map_ok(|event| {

src/bin/sse-test-api/main.rs

@@ -21,8 +21,6 @@ struct Config {
     stream_url: String,
     /// The URL of a callback endpoint created by the test harness .
     callback_url: String,
-    /// A string describing the current test, if desired for logging.
-    tag: Option<String>,
     /// An optional integer specifying the initial reconnection delay parameter, in
     /// milliseconds. Not all SSE client implementations allow this to be configured, but the
     /// test harness will send a value anyway in an attempt to avoid having reconnection tests
@@ -31,47 +29,30 @@ struct Config {
     /// An optional integer specifying the read timeout for the connection, in
     /// milliseconds.
     read_timeout_ms: Option<u64>,
-    /// An optional string which should be sent as the Last-Event-Id header in the initial
-    /// HTTP request. The test harness will only set this property if the test service has the
-    /// "last-event-id" capability.
-    last_event_id: Option<String>,
     /// A JSON object containing additional HTTP header names and string values. The SSE
     /// client should be configured to add these headers to its HTTP requests; the test harness
     /// will then verify that it receives those headers. The test harness will only set this
     /// property if the test service has the "headers" capability. Header names can be assumed
     /// to all be lowercase.
     headers: Option<HashMap<String, String>>,
-    /// A string specifying an HTTP method to use instead of GET. The test harness will only
-    /// set this property if the test service has the "post" or "report" capability.
-    method: Option<String>,
-    /// A string specifying data to be sent in the HTTP request body. The test harness will
-    /// only set this property if the test service has the "post" or "report" capability.
-    body: Option<String>,
 }
 
 #[derive(Serialize, Debug)]
 #[serde(tag = "kind")]
 enum EventType {
     #[serde(rename = "event")]
     Event { event: Event },
-    #[serde(rename = "comment")]
-    Comment { comment: String },
     #[serde(rename = "error")]
     Error { error: String },
 }
 
-impl From<es::SSE> for EventType {
-    fn from(event: es::SSE) -> Self {
-        match event {
-            es::SSE::Event(evt) => Self::Event {
-                event: Event {
-                    event_type: evt.event_type,
-                    data: String::from_utf8(evt.data.to_vec()).unwrap(),
-                    id: String::from_utf8(evt.id.to_vec()).unwrap(),
-                },
-            },
-            es::SSE::Comment(comment) => Self::Comment {
-                comment: String::from_utf8(comment).unwrap(),
+impl From<es::Event> for EventType {
+    fn from(event: es::Event) -> Self {
+        Self::Event {
+            event: Event {
+                event_type: event.event_type.clone(),
+                data: String::from_utf8(event.field("data").unwrap_or_default().to_vec()).unwrap(),
+                id: String::from_utf8(event.field("id").unwrap_or_default().to_vec()).unwrap(),
             },
         }
     }

src/bin/sse-test-api/stream_entity.rs

@@ -105,18 +105,6 @@ impl Inner {
             client_builder = client_builder.read_timeout(Duration::from_millis(read_timeout_ms));
         }
 
-        if let Some(last_event_id) = &config.last_event_id {
-            client_builder = client_builder.last_event_id(last_event_id.clone());
-        }
-
-        if let Some(method) = &config.method {
-            client_builder = client_builder.method(method.to_string());
-        }
-
-        if let Some(body) = &config.body {
-            client_builder = client_builder.body(body.to_string());
-        }
-
         if let Some(headers) = &config.headers {
             for (name, value) in headers {
                 client_builder = match client_builder.header(name, value) {

src/client.rs

@@ -3,30 +3,41 @@ use std::{
     future::Future,
     mem,
     pin::Pin,
+    str::FromStr,
     task::{Context, Poll},
     time::Duration,
 };
 
 use futures::{ready, Stream};
 use hyper::{
     body::{Bytes, HttpBody},
-    client::{connect::Connect, ResponseFuture},
-    header::HeaderMap,
+    client::{
+        connect::{Connect, Connection},
+        ResponseFuture,
+    },
+    header::{HeaderMap, HeaderName, HeaderValue},
+    service::Service,
     Body, Request, StatusCode, Uri,
 };
 #[cfg(feature = "rustls")]
 use hyper_rustls::HttpsConnector as RustlsConnector;
 use log::{debug, info, trace, warn};
 use pin_project::pin_project;
-use tokio::time::Sleep;
+
+use tokio::{
+    io::{AsyncRead, AsyncWrite},
+    time::Sleep,
+};
 
 use super::config::ReconnectOptions;
 use super::decode::Decoded;
 use super::error::{Error, Result};
 
 pub use hyper::client::HttpConnector;
+
 #[cfg(feature = "rustls")]
 pub type HttpsConnector = RustlsConnector<HttpConnector>;
+pub use hyper_timeout::TimeoutConnector;
 
 /*
  * TODO remove debug output
@@ -37,18 +48,28 @@ pub struct ClientBuilder {
     url: Uri,
     headers: HeaderMap,
     reconnect_opts: ReconnectOptions,
+    read_timeout: Option<Duration>,
 }
 
 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::InvalidParameter(Box::new(e)))?;
-        self.headers.insert(key, value);
+    pub fn header(mut self, name: &str, value: &str) -> Result<ClientBuilder> {
+        let name =
+            HeaderName::from_str(name).map_err(|_| Error::HttpRequest(StatusCode::BAD_REQUEST))?;
+
+        let value = HeaderValue::from_str(value)
+            .map_err(|_| Error::HttpRequest(StatusCode::BAD_REQUEST))?;
+
+        self.headers.insert(name, value);
         Ok(self)
     }
 
+    /// Set a read timeout for the underlying connection. There is no read timeout by default.
+    pub fn read_timeout(mut self, read_timeout: Duration) -> ClientBuilder {
+        self.read_timeout = Some(read_timeout);
+        self
+    }
+
     /// Configure the client's reconnect behaviour according to the supplied
     /// [`ReconnectOptions`].
     ///
@@ -58,12 +79,19 @@ impl ClientBuilder {
         self
     }
 
-    pub fn build_with_conn<C>(self, conn: C) -> Client<C>
+    pub fn build_with_conn<C>(self, conn: C) -> Client<TimeoutConnector<C>>
     where
-        C: Connect + Clone,
+        C: Service<Uri> + Send + 'static + std::clone::Clone,
+        C::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+        C::Future: Unpin + Send,
+        C::Response: AsyncRead + AsyncWrite + Connection + Unpin + Send,
     {
+        let mut connector = TimeoutConnector::new(conn);
+        connector.set_read_timeout(self.read_timeout);
+        let client = hyper::Client::builder().build::<_, hyper::Body>(connector);
+
         Client {
-            http: hyper::Client::builder().build(conn),
+            http: client,
             request_props: RequestProps {
                 url: self.url,
                 headers: self.headers,
@@ -72,12 +100,12 @@ impl ClientBuilder {
         }
     }
 
-    pub fn build_http(self) -> Client<HttpConnector> {
+    pub fn build_http(self) -> Client<TimeoutConnector<HttpConnector>> {
         self.build_with_conn(HttpConnector::new())
     }
 
     #[cfg(feature = "rustls")]
-    pub fn build(self) -> Client<HttpsConnector> {
+    pub fn build(self) -> Client<TimeoutConnector<HttpsConnector>> {
         let conn = HttpsConnector::with_native_roots();
         self.build_with_conn(conn)
     }
@@ -122,6 +150,7 @@ impl Client<()> {
             url,
             headers: HeaderMap::new(),
             reconnect_opts: ReconnectOptions::default(),
+            read_timeout: None,
         })
     }
 }

src/error.rs

@@ -3,6 +3,7 @@ use hyper::StatusCode;
 /// Error type returned from this library's functions.
 #[derive(Debug)]
 pub enum Error {
+    StreamClosed,
     /// An invalid request parameter
     InvalidParameter(Box<dyn std::error::Error + Send + 'static>),
     /// The HTTP request failed.