chore: Emit SSE::Connected event when stream is established

Author: Zarathustra2

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

@@ -50,6 +50,7 @@ struct Config {
 #[derive(Serialize, Debug)]
 #[serde(tag = "kind", rename_all = "camelCase")]
 enum EventType {
+    Connected {},
     Event { event: Event },
     Comment { comment: String },
     Error { error: String },
@@ -66,6 +67,7 @@ impl From<es::SSE> for EventType {
                 },
             },
             es::SSE::Comment(comment) => Self::Comment { comment },
+            es::SSE::Connected(_) => Self::Connected {},
         }
     }
 }

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

@@ -36,6 +36,10 @@ impl Inner {
             match stream.try_next().await {
                 Ok(Some(event)) => {
                     let event_type: EventType = event.into();
+                    if let EventType::Connected {} = event_type {
+                        continue;
+                    }
+
                     if !self.send_message(event_type, &client).await {
                         break;
                     }

eventsource-client/examples/tail.rs

@@ -46,6 +46,9 @@ fn tail_events(client: impl es::Client) -> impl Stream<Item = Result<(), ()>> {
             es::SSE::Comment(comment) => {
                 println!("got a comment: \n{}", comment)
             }
+            es::SSE::Connected(headers) => {
+                println!("got a connection start with headers: \n{:?}", headers)
+            }
         })
         .map_err(|err| eprintln!("error streaming events: {:?}", err))
 }

eventsource-client/src/client.rs

@@ -403,6 +403,7 @@ where
                         Poll::Ready(Some(Ok(event)))
                     }
                     SSE::Comment(_) => Poll::Ready(Some(Ok(event))),
+                    SSE::Connected(_) => Poll::Ready(Some(Ok(event))),
                 };
             }
 
@@ -436,13 +437,15 @@ where
                         debug!("HTTP response: {:#?}", resp);
 
                         if resp.status().is_success() {
+                            let reply =
+                                Poll::Ready(Some(Ok(SSE::Connected(resp.headers().to_owned()))));
                             self.as_mut().project().retry_strategy.reset(Instant::now());
                             self.as_mut().reset_redirects();
                             self.as_mut()
                                 .project()
                                 .state
                                 .set(State::Connected(resp.into_body()));
-                            continue;
+                            return reply;
                         }
 
                         if resp.status() == 301 || resp.status() == 307 {
@@ -575,19 +578,6 @@ fn delay(dur: Duration, description: &str) -> Sleep {
     tokio::time::sleep(dur)
 }
 
-#[derive(Debug)]
-struct StatusError {
-    status: StatusCode,
-}
-
-impl Display for StatusError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        write!(f, "Invalid status code: {}", self.status)
-    }
-}
-
-impl std::error::Error for StatusError {}
-
 mod private {
     use crate::client::ClientImpl;
 

eventsource-client/src/event_parser.rs

@@ -1,6 +1,6 @@
 use std::{collections::VecDeque, convert::TryFrom, str::from_utf8};
 
-use hyper::body::Bytes;
+use hyper::{body::Bytes, http::HeaderValue, HeaderMap};
 use log::{debug, log_enabled, trace};
 use pin_project::pin_project;
 
@@ -32,6 +32,7 @@ impl EventData {
 
 #[derive(Debug, Eq, PartialEq)]
 pub enum SSE {
+    Connected(HeaderMap<HeaderValue>),
     Event(Event),
     Comment(String),
 }

eventsource-client/src/lib.rs

@@ -14,7 +14,8 @@
 //! let mut stream = Box::pin(client.stream())
 //!     .map_ok(|event| match event {
 //!         SSE::Comment(comment) => println!("got a comment event: {:?}", comment),
-//!         SSE::Event(evt) => println!("got an event: {}", evt.event_type)
+//!         SSE::Event(evt) => println!("got an event: {}", evt.event_type),
+//!         SSE::Connected(headers) => println!("got connection start with headers: {:?}", headers)
 //!     })
 //!     .map_err(|e| println!("error streaming events: {:?}", e));
 //! # while let Ok(Some(_)) = stream.try_next().await {}