Merge pull request #456 from http-rs/sse

Add initial SSE support

Author: yoshuawuyts

Cargo.toml

@@ -29,6 +29,7 @@ docs = ["unstable"]
 unstable = []
 
 [dependencies]
+async-sse = "2.1.0"
 http-types = "1.0.1"
 http-service = "0.5.0"
 http-service-h1 = { version = "0.1.0", optional = true }

examples/sse.rs

@@ -0,0 +1,13 @@
+use tide::sse;
+
+#[async_std::main]
+async fn main() -> Result<(), std::io::Error> {
+    let mut app = tide::new();
+    app.at("/sse").get(sse::endpoint(|_req, sender| async move {
+        sender.send("fruit", "banana", None).await;
+        sender.send("fruit", "apple", None).await;
+        Ok(())
+    }));
+    app.listen("localhost:8080").await?;
+    Ok(())
+}

src/lib.rs

@@ -201,6 +201,7 @@ pub mod security;
 
 pub use endpoint::Endpoint;
 pub use request::Request;
+pub mod sse;
 
 #[doc(inline)]
 pub use http_types::{Error, Result, Status};

src/response/mod.rs

@@ -23,7 +23,7 @@ pub(crate) enum CookieEvent {
 /// An HTTP response
 #[derive(Debug)]
 pub struct Response {
-    res: http_service::Response,
+    pub(crate) res: http_service::Response,
     // tracking here
     pub(crate) cookie_events: Vec<CookieEvent>,
 }

src/sse/endpoint.rs

@@ -0,0 +1,70 @@
+use crate::http::{mime, Body, StatusCode};
+use crate::log;
+use crate::sse::Sender;
+use crate::utils::BoxFuture;
+use crate::{Endpoint, Request, Response, Result};
+
+use async_std::future::Future;
+use async_std::io::BufReader;
+use async_std::task;
+
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+/// Create an endpoint that can handle SSE connections.
+pub fn endpoint<F, Fut, State>(handler: F) -> SseEndpoint<F, Fut, State>
+where
+    State: Send + Sync + 'static,
+    F: Fn(Request<State>, Sender) -> Fut + Send + Sync + 'static,
+    Fut: Future<Output = Result<()>> + Send + Sync + 'static,
+{
+    SseEndpoint {
+        handler: Arc::new(handler),
+        __state: PhantomData,
+        __fut: PhantomData,
+    }
+}
+
+/// An endpoint that can handle SSE connections.
+#[derive(Debug)]
+pub struct SseEndpoint<F, Fut, State>
+where
+    State: Send + Sync + 'static,
+    F: Fn(Request<State>, Sender) -> Fut + Send + Sync + 'static,
+    Fut: Future<Output = Result<()>> + Send + Sync + 'static,
+{
+    handler: Arc<F>,
+    __state: PhantomData<State>,
+    __fut: PhantomData<Fut>,
+}
+
+impl<F, Fut, State> Endpoint<State> for SseEndpoint<F, Fut, State>
+where
+    State: Send + Sync + 'static,
+    F: Fn(Request<State>, Sender) -> Fut + Send + Sync + 'static,
+    Fut: Future<Output = Result<()>> + Send + Sync + 'static,
+{
+    fn call<'a>(&'a self, req: Request<State>) -> BoxFuture<'a, Result<Response>> {
+        let handler = self.handler.clone();
+        Box::pin(async move {
+            let (sender, encoder) = async_sse::encode();
+            task::spawn(async move {
+                let sender = Sender::new(sender);
+                if let Err(err) = handler(req, sender).await {
+                    log::error!("SSE handler error: {:?}", err);
+                }
+            });
+
+            // Perform the handshake as described here:
+            // https://html.spec.whatwg.org/multipage/server-sent-events.html#sse-processing-model
+            let mut res = Response::new(StatusCode::Ok);
+            res.res.insert_header("Cache-Control", "no-cache").unwrap();
+            res.res.set_content_type(mime::SSE);
+
+            let body = Body::from_reader(BufReader::new(encoder), None);
+            res.set_body(body);
+
+            Ok(res)
+        })
+    }
+}

src/sse/mod.rs

@@ -0,0 +1,35 @@
+//! Server-Sent Events (SSE) types.
+//!
+//! # Errors
+//!
+//! Errors originating in the SSE handler will be logged. Errors originating
+//! during the encoding of the SSE stream will be handled by the backend engine
+//! the way any other IO error is handled.
+//!
+//! In the future we may introduce a better mechanism to handle errors that
+//! originate outside of regular endpoints.
+//!
+//! # Examples
+//!
+//! ```no_run
+//! # fn main() -> Result<(), std::io::Error> { async_std::task::block_on(async {
+//! #
+//! use tide::sse;
+//!
+//! let mut app = tide::new();
+//! app.at("/sse").get(sse::endpoint(|_req, sender| async move {
+//!     sender.send("fruit", "banana", None).await;
+//!     sender.send("fruit", "apple", None).await;
+//!     Ok(())
+//! }));
+//! app.listen("localhost:8080").await?;
+//! # Ok(()) }) }
+//! ```
+
+mod endpoint;
+mod sender;
+mod upgrade;
+
+pub use endpoint::{endpoint, SseEndpoint};
+pub use sender::Sender;
+pub use upgrade::upgrade;

src/sse/sender.rs

@@ -0,0 +1,19 @@
+/// An SSE message sender.
+#[derive(Debug)]
+pub struct Sender {
+    sender: async_sse::Sender,
+}
+
+impl Sender {
+    /// Create a new instance of `Sender`.
+    pub(crate) fn new(sender: async_sse::Sender) -> Self {
+        Self { sender }
+    }
+
+    /// Send data from the SSE channel.
+    ///
+    /// Each message constists of a "name" and "data".
+    pub async fn send(&self, name: &str, data: impl AsRef<str>, id: Option<&str>) {
+        self.sender.send(name, data.as_ref().as_bytes(), id).await;
+    }
+}

src/sse/upgrade.rs

@@ -0,0 +1,36 @@
+use crate::http::{mime, Body, StatusCode};
+use crate::log;
+use crate::{Request, Response, Result};
+
+use super::Sender;
+
+use async_std::future::Future;
+use async_std::io::BufReader;
+use async_std::task;
+
+/// Upgrade an existing HTTP connection to an SSE connection.
+pub fn upgrade<F, Fut, State>(req: Request<State>, handler: F) -> Response
+where
+    State: Send + Sync + 'static,
+    F: Fn(Request<State>, Sender) -> Fut + Send + Sync + 'static,
+    Fut: Future<Output = Result<()>> + Send + Sync + 'static,
+{
+    let (sender, encoder) = async_sse::encode();
+    task::spawn(async move {
+        let sender = Sender::new(sender);
+        if let Err(err) = handler(req, sender).await {
+            log::error!("SSE handler error: {:?}", err);
+        }
+    });
+
+    // Perform the handshake as described here:
+    // https://html.spec.whatwg.org/multipage/server-sent-events.html#sse-processing-model
+    let mut res = Response::new(StatusCode::Ok);
+    res.res.insert_header("Cache-Control", "no-cache").unwrap();
+    res.res.set_content_type(mime::SSE);
+
+    let body = Body::from_reader(BufReader::new(encoder), None);
+    res.set_body(body);
+
+    res
+}