Repository: seanmonstar/warp (language: rust)

Issue:
Change SSE module to be more like WS
The `sse` module uses `trait SseEvent` and helper functions that return `impl SseEvent`, and while clever, these cause confusion (see #296) and make it difficult to match types up (see #643). This issue describes how we should change `sse` to be more like `ws`, which just uses plain structs (`ws::Messasge`).

- Create a `struct Event`.
- This should probably have private fields for `name`, `id`, `data`, and `retry`. Some should probably all be optional? Some research there will help us to know which can be optional.


Discussion / hints:

Fix patch:
diff --git a/examples/sse.rs b/examples/sse.rs
index 9c883a7c9..7177d91c0 100644
--- a/examples/sse.rs
+++ b/examples/sse.rs
@@ -2,11 +2,11 @@ use futures::StreamExt;
 use std::convert::Infallible;
 use std::time::Duration;
 use tokio::time::interval;
-use warp::{sse::ServerSentEvent, Filter};
+use warp::{sse::Event, Filter};
 
 // create server-sent event
-fn sse_counter(counter: u64) -> Result<impl ServerSentEvent, Infallible> {
-    Ok(warp::sse::data(counter))
+fn sse_counter(counter: u64) -> Result<Event, Infallible> {
+    Ok(warp::sse::Event::default().data(counter.to_string()))
 }
 
 #[tokio::main]
diff --git a/examples/sse_chat.rs b/examples/sse_chat.rs
index 693a87f55..b6b6221c7 100644
--- a/examples/sse_chat.rs
+++ b/examples/sse_chat.rs
@@ -5,7 +5,7 @@ use std::sync::{
     Arc, Mutex,
 };
 use tokio::sync::mpsc;
-use warp::{sse::ServerSentEvent, Filter};
+use warp::{sse::Event, Filter};
 
 #[tokio::main]
 async fn main() {
@@ -74,10 +74,7 @@ impl warp::reject::Reject for NotUtf8 {}
 /// - Value is a sender of `Message`
 type Users = Arc<Mutex<HashMap<usize, mpsc::UnboundedSender<Message>>>>;
 
-fn user_connected(
-    users: Users,
-) -> impl Stream<Item = Result<impl ServerSentEvent + Send + 'static, warp::Error>> + Send + 'static
-{
+fn user_connected(users: Users) -> impl Stream<Item = Result<Event, warp::Error>> + Send + 'static {
     // Use a counter to assign a new unique ID for this user.
     let my_id = NEXT_USER_ID.fetch_add(1, Ordering::Relaxed);
 
@@ -96,8 +93,8 @@ fn user_connected(
 
     // Convert messages into Server-Sent Events and return resulting stream.
     rx.map(|msg| match msg {
-        Message::UserId(my_id) => Ok((warp::sse::event("user"), warp::sse::data(my_id)).into_a()),
-        Message::Reply(reply) => Ok(warp::sse::data(reply).into_b()),
+        Message::UserId(my_id) => Ok(Event::default().event("user").data(my_id.to_string())),
+        Message::Reply(reply) => Ok(Event::default().data(reply)),
     })
 }
 
diff --git a/src/filters/sse.rs b/src/filters/sse.rs
index 8c4eda58f..659c74225 100644
--- a/src/filters/sse.rs
+++ b/src/filters/sse.rs
@@ -6,22 +6,22 @@
 //!
 //! use std::time::Duration;
 //! use std::convert::Infallible;
-//! use warp::{Filter, sse::ServerSentEvent};
+//! use warp::{Filter, sse::Event};
 //! use futures::{stream::iter, Stream};
 //!
-//! fn sse_events() -> impl Stream<Item = Result<impl ServerSentEvent, Infallible>> {
+//! fn sse_events() -> impl Stream<Item = Result<Event, Infallible>> {
 //!     iter(vec![
-//!         Ok(warp::sse::data("unnamed event").into_a()),
-//!         Ok((
-//!             warp::sse::event("chat"),
-//!             warp::sse::data("chat message"),
-//!         ).into_a().into_b()),
-//!         Ok((
-//!             warp::sse::id(13),
-//!             warp::sse::event("chat"),
-//!             warp::sse::data("other chat message\nwith next line"),
-//!             warp::sse::retry(Duration::from_millis(5000)),
-//!         ).into_b().into_b()),
+//!         Ok(Event::default().data("unnamed event")),
+//!         Ok(
+//!             Event::default().event("chat")
+//!             .data("chat message")
+//!         ),
+//!         Ok(
+//!             Event::default().id(13.to_string())
+//!             .event("chat")
+//!             .data("other chat message\nwith next line")
+//!             .retry(Duration::from_millis(5000))
+//!         )
 //!     ])
 //! }
 //!
@@ -39,6 +39,7 @@
 //! which specifies the expected behavior of Server Sent Events.
 //!
 
+use serde::Serialize;
 use std::borrow::Cow;
 use std::error::Error as StdError;
 use std::fmt::{self, Display, Formatter, Write};
@@ -52,119 +53,118 @@ use futures::{future, Stream, TryStream, TryStreamExt};
 use http::header::{HeaderValue, CACHE_CONTROL, CONTENT_TYPE};
 use hyper::Body;
 use pin_project::pin_project;
-use serde::Serialize;
-use serde_json;
+use serde_json::{self, Error};
 use tokio::time::{self, Delay};
 
-use self::sealed::{
-    BoxedServerSentEvent, EitherServerSentEvent, SseError, SseField, SseFormat, SseWrapper,
-};
+use self::sealed::SseError;
 use super::header;
 use crate::filter::One;
 use crate::reply::Response;
 use crate::{Filter, Rejection, Reply};
 
-/// Server-sent event message
-pub trait ServerSentEvent: SseFormat + Sized + Send + Sync + 'static {
-    /// Convert to either A
-    fn into_a<B>(self) -> EitherServerSentEvent<Self, B> {
-        EitherServerSentEvent::A(self)
+// Server-sent event data type
+#[derive(Debug)]
+enum DataType {
+    Text(String),
+    Json(String),
+}
+
+/// Server-sent event
+#[derive(Default, Debug)]
+pub struct Event {
+    name: Option<String>,
+    id: Option<String>,
+    data: Option<DataType>,
+    event: Option<String>,
+    comment: Option<String>,
+    retry: Option<Duration>,
+}
+
+impl Event {
+    /// Set Server-sent event data
+    /// data field(s) ("data:<content>")
+    pub fn data<T: Into<String>>(mut self, data: T) -> Event {
+        self.data = Some(DataType::Text(data.into()));
+        self
     }
 
-    /// Convert to either B
-    fn into_b<A>(self) -> EitherServerSentEvent<A, Self> {
-        EitherServerSentEvent::B(self)
+    /// Set Server-sent event data
+    /// data field(s) ("data:<content>")
+    pub fn json_data<T: Serialize>(mut self, data: T) -> Result<Event, Error> {
+        self.data = Some(DataType::Json(serde_json::to_string(&data)?));
+        Ok(self)
     }
 
-    /// Convert to boxed
-    fn boxed(self) -> BoxedServerSentEvent {
-        BoxedServerSentEvent(Box::new(self))
+    /// Set Server-sent event comment
+    /// Comment field (":<comment-text>")
+    pub fn comment<T: Into<String>>(mut self, comment: T) -> Event {
+        self.comment = Some(comment.into());
+        self
     }
-}
 
-impl<T: SseFormat + Send + Sync + 'static> ServerSentEvent for T {}
+    /// Set Server-sent event event
+    /// Event name field ("event:<event-name>")
+    pub fn event<T: Into<String>>(mut self, event: T) -> Event {
+        self.event = Some(event.into());
+        self
+    }
 
-#[allow(missing_debug_implementations)]
-struct SseComment<T>(T);
+    /// Set Server-sent event retry
+    /// Retry timeout field ("retry:<timeout>")
+    pub fn retry(mut self, duration: Duration) -> Event {
+        self.retry = Some(duration.into());
+        self
+    }
 
-/// Comment field (":<comment-text>")
-pub fn comment<T>(comment: T) -> impl ServerSentEvent
-where
-    T: Display + Send + Sync + 'static,
-{
-    SseComment(comment)
+    /// Set Server-sent event id
+    /// Identifier field ("id:<identifier>")
+    pub fn id<T: Into<String>>(mut self, id: T) -> Event {
+        self.id = Some(id.into());
+        self
+    }
 }
 
-impl<T: Display> SseFormat for SseComment<T> {
-    fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-        if let SseField::Comment = k {
-            k.fmt(f)?;
-            self.0.fmt(f)?;
+impl Display for Event {
+    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+        if let Some(ref comment) = &self.comment {
+            ":".fmt(f)?;
+            comment.fmt(f)?;
             f.write_char('\n')?;
         }
-        Ok(())
-    }
-}
 
-#[allow(missing_debug_implementations)]
-struct SseEvent<T>(T);
-
-/// Event name field ("event:<event-name>")
-pub fn event<T>(event: T) -> impl ServerSentEvent
-where
-    T: Display + Send + Sync + 'static,
-{
-    SseEvent(event)
-}
-
-impl<T: Display> SseFormat for SseEvent<T> {
-    fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-        if let SseField::Event = k {
-            k.fmt(f)?;
-            self.0.fmt(f)?;
+        if let Some(ref event) = &self.event {
+            "event:".fmt(f)?;
+            event.fmt(f)?;
             f.write_char('\n')?;
         }
-        Ok(())
-    }
-}
-
-#[allow(missing_debug_implementations)]
-struct SseId<T>(T);
 
-/// Identifier field ("id:<identifier>")
-pub fn id<T>(id: T) -> impl ServerSentEvent
-where
-    T: Display + Send + Sync + 'static,
-{
-    SseId(id)
-}
+        match self.data {
+            Some(DataType::Text(ref data)) => {
+                for line in data.split('\n') {
+                    "data:".fmt(f)?;
+                    line.fmt(f)?;
+                    f.write_char('\n')?;
+                }
+            }
+            Some(DataType::Json(ref data)) => {
+                "data:".fmt(f)?;
+                data.fmt(f)?;
+                f.write_char('\n')?;
+            }
+            None => {}
+        }
 
-impl<T: Display> SseFormat for SseId<T> {
-    fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-        if let SseField::Id = k {
-            k.fmt(f)?;
-            self.0.fmt(f)?;
+        if let Some(ref id) = &self.id {
+            "id:".fmt(f)?;
+            id.fmt(f)?;
             f.write_char('\n')?;
         }
-        Ok(())
-    }
-}
 
-#[allow(missing_debug_implementations)]
-struct SseRetry(Duration);
+        if let Some(ref duration) = &self.retry {
+            "retry:".fmt(f)?;
 
-/// Retry timeout field ("retry:<timeout>")
-pub fn retry(time: Duration) -> impl ServerSentEvent {
-    SseRetry(time)
-}
-
-impl SseFormat for SseRetry {
-    fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-        if let SseField::Retry = k {
-            k.fmt(f)?;
-
-            let secs = self.0.as_secs();
-            let millis = self.0.subsec_millis();
+            let secs = duration.as_secs();
+            let millis = duration.subsec_millis();
 
             if secs > 0 {
                 // format seconds
@@ -183,86 +183,12 @@ impl SseFormat for SseRetry {
 
             f.write_char('\n')?;
         }
-        Ok(())
-    }
-}
-
-#[allow(missing_debug_implementations)]
-struct SseData<T>(T);
-
-/// Data field(s) ("data:<content>")
-///
-/// The multiline content will be transferred
-/// using sequential data fields, one per line.
-pub fn data<T>(data: T) -> impl ServerSentEvent
-where
-    T: Display + Send + Sync + 'static,
-{
-    SseData(data)
-}
-
-impl<T: Display> SseFormat for SseData<T> {
-    fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-        if let SseField::Data = k {
-            for line in self.0.to_string().split('\n') {
-                k.fmt(f)?;
-                line.fmt(f)?;
-                f.write_char('\n')?;
-            }
-        }
-        Ok(())
-    }
-}
-
-#[allow(missing_debug_implementations)]
-struct SseJson<T>(T);
 
-/// Data field with JSON content ("data:<json-content>")
-pub fn json<T>(data: T) -> impl ServerSentEvent
-where
-    T: Serialize + Send + Sync + 'static,
-{
-    SseJson(data)
-}
-
-impl<T: Serialize> SseFormat for SseJson<T> {
-    fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-        if let SseField::Data = k {
-            k.fmt(f)?;
-            serde_json::to_string(&self.0)
-                .map_err(|error| {
-                    tracing::error!("sse::json error {}", error);
-                    fmt::Error
-                })
-                .and_then(|data| data.fmt(f))?;
-            f.write_char('\n')?;
-        }
+        f.write_char('\n')?;
         Ok(())
     }
 }
 
-macro_rules! tuple_fmt {
-    (($($t:ident),+) => ($($i:tt),+)) => {
-        impl<$($t),+> SseFormat for ($($t),+)
-        where
-            $($t: SseFormat,)+
-        {
-            fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-                $(self.$i.fmt_field(f, k)?;)+
-                Ok(())
-            }
-        }
-    };
-}
-
-tuple_fmt!((A, B) => (0, 1));
-tuple_fmt!((A, B, C) => (0, 1, 2));
-tuple_fmt!((A, B, C, D) => (0, 1, 2, 3));
-tuple_fmt!((A, B, C, D, E) => (0, 1, 2, 3, 4));
-tuple_fmt!((A, B, C, D, E, F) => (0, 1, 2, 3, 4, 5));
-tuple_fmt!((A, B, C, D, E, F, G) => (0, 1, 2, 3, 4, 5, 6));
-tuple_fmt!((A, B, C, D, E, F, G, H) => (0, 1, 2, 3, 4, 5, 6, 7));
-
 /// Gets the optional last event id from request.
 /// Typically this identifier represented as number or string.
 ///
@@ -322,7 +248,7 @@ where
 /// use futures::Stream;
 /// use futures::stream::iter;
 /// use std::convert::Infallible;
-/// use warp::{Filter, sse::ServerSentEvent};
+/// use warp::{Filter, sse::Event};
 /// use serde_derive::Serialize;
 ///
 /// #[derive(Serialize)]
@@ -331,25 +257,25 @@ where
 ///     text: String,
 /// }
 ///
-/// fn event_stream() -> impl Stream<Item = Result<impl ServerSentEvent, Infallible>> {
+/// fn event_stream() -> impl Stream<Item = Result<Event, Infallible>> {
 ///         iter(vec![
 ///             // Unnamed event with data only
-///             Ok(warp::sse::data("payload").boxed()),
+///             Ok(Event::default().data("payload")),
 ///             // Named event with ID and retry timeout
-///             Ok((
-///                 warp::sse::data("other message\nwith next line"),
-///                 warp::sse::event("chat"),
-///                 warp::sse::id(1),
-///                 warp::sse::retry(Duration::from_millis(15000))
-///             ).boxed()),
+///             Ok(
+///                 Event::default().data("other message\nwith next line")
+///                 .event("chat")
+///                 .id(1.to_string())
+///                 .retry(Duration::from_millis(15000))
+///             ),
 ///             // Event with JSON data
-///             Ok((
-///                 warp::sse::id(2),
-///                 warp::sse::json(Msg {
+///             Ok(
+///                 Event::default().id(2.to_string())
+///                 .json_data(Msg {
 ///                     from: 2,
 ///                     text: "hello".into(),
-///                 }),
-///             ).boxed()),
+///                 }).unwrap(),
+///             )
 ///         ])
 /// }
 ///
@@ -385,8 +311,7 @@ where
 /// ```
 pub fn reply<S>(event_stream: S) -> impl Reply
 where
-    S: TryStream + Send + 'static,
-    S::Ok: ServerSentEvent,
+    S: TryStream<Ok = Event> + Send + 'static,
     S::Error: StdError + Send + Sync + 'static,
 {
     SseReply { event_stream }
@@ -399,8 +324,7 @@ struct SseReply<S> {
 
 impl<S> Reply for SseReply<S>
 where
-    S: TryStream + Send + 'static,
-    S::Ok: ServerSentEvent,
+    S: TryStream<Ok = Event> + Send + 'static,
     S::Error: StdError + Send + Sync + 'static,
 {
     #[inline]
@@ -409,11 +333,11 @@ where
             .event_stream
             .map_err(|error| {
                 // FIXME: error logging
-                tracing::error!("sse stream error: {}", error);
+                log::error!("sse stream error: {}", error);
                 SseError
             })
             .into_stream()
-            .and_then(|event| future::ready(SseWrapper::format(&event)));
+            .and_then(|event| future::ready(Ok(event.to_string())));
 
         let mut res = Response::new(Body::wrap_stream(body_stream));
         // Set appropriate content type
@@ -457,14 +381,9 @@ impl KeepAlive {
     pub fn stream<S>(
         self,
         event_stream: S,
-    ) -> impl TryStream<
-        Ok = impl ServerSentEvent + Send + 'static,
-        Error = impl StdError + Send + Sync + 'static,
-    > + Send
-           + 'static
+    ) -> impl TryStream<Ok = Event, Error = impl StdError + Send + Sync + 'static> + Send + 'static
     where
-        S: TryStream + Send + 'static,
-        S::Ok: ServerSentEvent + Send,
+        S: TryStream<Ok = Event> + Send + 'static,
         S::Error: StdError + Send + Sync + 'static,
     {
         let alive_timer = time::delay_for(self.max_interval);
@@ -487,33 +406,6 @@ struct SseKeepAlive<S> {
     alive_timer: Delay,
 }
 
-#[doc(hidden)]
-#[deprecated(note = "use warp::see:keep_alive() instead")]
-pub fn keep<S>(
-    event_stream: S,
-    keep_interval: impl Into<Option<Duration>>,
-) -> impl TryStream<
-    Ok = impl ServerSentEvent + Send + 'static,
-    Error = impl StdError + Send + Sync + 'static,
-> + Send
-       + 'static
-where
-    S: TryStream + Send + 'static,
-    S::Ok: ServerSentEvent + Send,
-    S::Error: StdError + Send + Sync + 'static,
-{
-    let max_interval = keep_interval
-        .into()
-        .unwrap_or_else(|| Duration::from_secs(15));
-    let alive_timer = time::delay_for(max_interval);
-    SseKeepAlive {
-        event_stream,
-        comment_text: Cow::Borrowed(""),
-        max_interval,
-        alive_timer,
-    }
-}
-
 /// Keeps event source connection alive when no events sent over a some time.
 ///
 /// Some proxy servers may drop HTTP connection after a some timeout of inactivity.
@@ -529,11 +421,11 @@ where
 /// use std::convert::Infallible;
 /// use futures::StreamExt;
 /// use tokio::time::interval;
-/// use warp::{Filter, Stream, sse::ServerSentEvent};
+/// use warp::{Filter, Stream, sse::Event};
 ///
 /// // create server-sent event
-/// fn sse_counter(counter: u64) ->  Result<impl ServerSentEvent, Infallible> {
-///     Ok(warp::sse::data(counter))
+/// fn sse_counter(counter: u64) ->  Result<Event, Infallible> {
+///     Ok(Event::default().data(counter.to_string()))
 /// }
 ///
 /// fn main() {
@@ -565,11 +457,10 @@ pub fn keep_alive() -> KeepAlive {
 
 impl<S> Stream for SseKeepAlive<S>
 where
-    S: TryStream + Send + 'static,
-    S::Ok: ServerSentEvent,
+    S: TryStream<Ok = Event> + Send + 'static,
     S::Error: StdError + Send + Sync + 'static,
 {
-    type Item = Result<EitherServerSentEvent<S::Ok, SseComment<Cow<'static, str>>>, SseError>;
+    type Item = Result<Event, SseError>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         let mut pin = self.project();
@@ -581,18 +472,19 @@ where
                     pin.alive_timer
                         .reset(tokio::time::Instant::now() + *pin.max_interval);
                     let comment_str = pin.comment_text.clone();
-                    Poll::Ready(Some(Ok(EitherServerSentEvent::B(SseComment(comment_str)))))
+                    let event = Event::default().comment(comment_str);
+                    Poll::Ready(Some(Ok(event)))
                 }
             },
             Poll::Ready(Some(Ok(event))) => {
                 // restart timer
                 pin.alive_timer
                     .reset(tokio::time::Instant::now() + *pin.max_interval);
-                Poll::Ready(Some(Ok(EitherServerSentEvent::A(event))))
+                Poll::Ready(Some(Ok(event)))
             }
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Ready(Some(Err(error))) => {
-                tracing::error!("sse::keep error: {}", error);
+                log::error!("sse::keep error: {}", error);
                 Poll::Ready(Some(Err(SseError)))
             }
         }
@@ -613,103 +505,4 @@ mod sealed {
     }
 
     impl StdError for SseError {}
-
-    impl Display for SseField {
-        fn fmt(&self, f: &mut Formatter) -> fmt::Result {
-            use self::SseField::*;
-            f.write_str(match self {
-                Event => "event:",
-                Id => "id:",
-                Data => "data:",
-                Retry => "retry:",
-                Comment => ":",
-            })
-        }
-    }
-
-    /// SSE field kind
-    #[allow(missing_debug_implementations)]
-    pub enum SseField {
-        /// Event name field
-        Event,
-        /// Event id field
-        Id,
-        /// Event data field
-        Data,
-        /// Retry timeout field
-        Retry,
-        /// Comment field
-        Comment,
-    }
-
-    /// SSE formatter trait
-    pub trait SseFormat {
-        /// format message field
-        fn fmt_field(&self, _f: &mut Formatter, _key: &SseField) -> fmt::Result {
-            Ok(())
-        }
-    }
-
-    /// SSE wrapper to help formatting messages
-    #[allow(missing_debug_implementations)]
-    pub struct SseWrapper<'a, T: 'a>(&'a T);
-
-    impl<'a, T> SseWrapper<'a, T>
-    where
-        T: SseFormat + 'a,
-    {
-        pub fn format(event: &'a T) -> Result<String, SseError> {
-            let mut buf = String::new();
-            buf.write_fmt(format_args!("{}", SseWrapper(event)))
-                .map_err(|_| SseError)?;
-            buf.shrink_to_fit();
-            Ok(buf)
-        }
-    }
-
-    impl<'a, T> Display for SseWrapper<'a, T>
-    where
-        T: SseFormat,
-    {
-        fn fmt(&self, f: &mut Formatter) -> fmt::Result {
-            self.0.fmt_field(f, &SseField::Comment)?;
-            // The event name usually transferred before the other fields.
-            self.0.fmt_field(f, &SseField::Event)?;
-            // It is important that the data will be transferred before
-            // the identifier to prevent possible losing events when
-            // resuming connection.
-            self.0.fmt_field(f, &SseField::Data)?;
-            self.0.fmt_field(f, &SseField::Id)?;
-            self.0.fmt_field(f, &SseField::Retry)?;
-            f.write_char('\n')
-        }
-    }
-
-    #[allow(missing_debug_implementations)]
-    pub struct BoxedServerSentEvent(pub(super) Box<dyn SseFormat + Send + Sync>);
-
-    impl SseFormat for BoxedServerSentEvent {
-        fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-            self.0.fmt_field(f, k)
-        }
-    }
-
-    #[allow(missing_debug_implementations)]
-    pub enum EitherServerSentEvent<A, B> {
-        A(A),
-        B(B),
-    }
-
-    impl<A, B> SseFormat for EitherServerSentEvent<A, B>
-    where
-        A: SseFormat,
-        B: SseFormat,
-    {
-        fn fmt_field(&self, f: &mut Formatter, k: &SseField) -> fmt::Result {
-            match self {
-                EitherServerSentEvent::A(a) => a.fmt_field(f, k),
-                EitherServerSentEvent::B(b) => b.fmt_field(f, k),
-            }
-        }
-    }
 }


Test patch: