Ignore invalid event IDs & persist existing event IDs (#25)

* Ignore invalid event IDs
* Persist already-seen event IDs into new events that lack IDs
* Push utf decoding earlier into event parsing process
* Enable 'headers' and 'last-event-id' capabilities

Author: cwaldren-ld

Makefile

@@ -1,6 +1,5 @@
 TEMP_TEST_OUTPUT=/tmp/contract-test-service.log
-SKIPFLAGS = -skip 'reconnection' -skip 'basic parsing/ID field is ignored if it contains a null' -skip 'basic parsing/last ID persists if not overridden by later event' \
--skip 'HTTP behavior/client follows 301 redirect' -skip 'HTTP behavior/client follows 307 redirect'
+SKIPFLAGS = -skip 'reconnection' -skip 'HTTP behavior/client follows 301 redirect' -skip 'HTTP behavior/client follows 307 redirect'
 
 
 build-contract-tests:

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

@@ -61,13 +61,11 @@ impl From<es::SSE> for EventType {
             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(),
+                    data: evt.data,
+                    id: evt.id,
                 },
             },
-            es::SSE::Comment(comment) => Self::Comment {
-                comment: String::from_utf8(comment).unwrap(),
-            },
+            es::SSE::Comment(comment) => Self::Comment { comment },
         }
     }
 }
@@ -77,7 +75,7 @@ struct Event {
     #[serde(rename = "type")]
     event_type: String,
     data: String,
-    id: String,
+    id: Option<String>,
 }
 
 async fn status() -> impl Responder {
@@ -86,8 +84,8 @@ async fn status() -> impl Responder {
             // "comments".to_string(),
             // "post".to_string(),
             // "report".to_string(),
-            // "headers".to_string(),
-            // "last-event-id".to_string(),
+            "headers".to_string(),
+            "last-event-id".to_string(),
         ],
     })
 }

eventsource-client/examples/tail.rs

@@ -1,6 +1,6 @@
 use es::Client;
 use futures::{Stream, TryStreamExt};
-use std::{env, process, str::from_utf8, time::Duration};
+use std::{env, process, time::Duration};
 
 use eventsource_client as es;
 
@@ -42,17 +42,10 @@ fn tail_events(client: impl Client) -> impl Stream<Item = Result<(), ()>> {
         .stream()
         .map_ok(|event| match event {
             es::SSE::Event(ev) => {
-                println!(
-                    "got an event: {}\n{}",
-                    ev.event_type,
-                    from_utf8(&ev.data).unwrap_or_default()
-                )
+                println!("got an event: {}\n{}", ev.event_type, ev.data)
             }
             es::SSE::Comment(comment) => {
-                println!(
-                    "got a comment: \n{}",
-                    from_utf8(&comment).unwrap_or_default()
-                )
+                println!("got a comment: \n{}", comment)
             }
         })
         .map_err(|err| eprintln!("error streaming events: {:?}", err))

eventsource-client/src/client.rs

@@ -65,7 +65,7 @@ pub struct ClientBuilder {
     headers: HeaderMap,
     reconnect_opts: ReconnectOptions,
     read_timeout: Option<Duration>,
-    last_event_id: String,
+    last_event_id: Option<String>,
     method: String,
     body: Option<String>,
 }
@@ -86,7 +86,7 @@ impl ClientBuilder {
             headers: header_map,
             reconnect_opts: ReconnectOptions::default(),
             read_timeout: None,
-            last_event_id: String::new(),
+            last_event_id: None,
             method: String::from("GET"),
             body: None,
         })
@@ -107,7 +107,7 @@ impl ClientBuilder {
     /// Set the last event id for a stream when it is created. If it is set, it will be sent to the
     /// server in case it can replay missed events.
     pub fn last_event_id(mut self, last_event_id: String) -> ClientBuilder {
-        self.last_event_id = last_event_id;
+        self.last_event_id = Some(last_event_id);
         self
     }
 
@@ -209,7 +209,7 @@ struct RequestProps {
 struct ClientImpl<C> {
     http: hyper::Client<C>,
     request_props: RequestProps,
-    last_event_id: String,
+    last_event_id: Option<String>,
 }
 
 impl<C> Client for ClientImpl<C>
@@ -274,14 +274,14 @@ pub struct ReconnectingRequest<C> {
     state: State,
     next_reconnect_delay: Duration,
     event_parser: EventParser,
-    last_event_id: String,
+    last_event_id: Option<String>,
 }
 
 impl<C> ReconnectingRequest<C> {
     fn new(
         http: hyper::Client<C>,
         props: RequestProps,
-        last_event_id: String,
+        last_event_id: Option<String>,
     ) -> ReconnectingRequest<C> {
         let reconnect_delay = props.reconnect_opts.delay;
         ReconnectingRequest {
@@ -306,11 +306,11 @@ impl<C> ReconnectingRequest<C> {
             request_builder = request_builder.header(name, value);
         }
 
-        if !self.last_event_id.is_empty() {
-            request_builder = request_builder.header(
-                "last-event-id",
-                HeaderValue::from_str(&self.last_event_id.clone()).unwrap(),
-            );
+        if self.last_event_id.is_some() {
+            let id_as_header = HeaderValue::from_str(self.last_event_id.as_ref().unwrap())
+                .map_err(|e| Error::InvalidParameter(Box::new(e)))?;
+
+            request_builder = request_builder.header("last-event-id", id_as_header);
         }
 
         let body = match &self.props.body {
@@ -357,8 +357,8 @@ where
             if let Some(event) = this.event_parser.get_event() {
                 return match event {
                     SSE::Event(ref evt) => {
-                        if !evt.id.is_empty() {
-                            *this.last_event_id = String::from_utf8(evt.id.clone()).unwrap();
+                        if evt.id.is_some() {
+                            *this.last_event_id = evt.id.clone();
                         }
 
                         if let Some(retry) = evt.retry {

eventsource-client/src/error.rs

@@ -18,8 +18,6 @@ pub enum Error {
     UnexpectedEof,
     /// Encountered a line not conforming to the SSE protocol.
     InvalidLine(String),
-    /// Encountered an event type that is not a valid UTF-8 byte sequence.
-    InvalidEventType(std::str::Utf8Error),
     InvalidEvent,
     /// An unexpected failure occurred.
     Unexpected(Box<dyn std::error::Error + Send + 'static>),