feat: add Server-Sent Events support for stream cat

src/api.rs

@@ -14,6 +14,7 @@ use tokio_util::compat::{FuturesAsyncReadCompatExt, FuturesAsyncWriteCompatExt};
 use http_body_util::StreamBody;
 use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full};
 use hyper::body::Bytes;
+use hyper::header::ACCEPT;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Method, Request, Response, StatusCode};
@@ -27,8 +28,14 @@ use crate::thread_pool::ThreadPool;
 type BoxError = Box<dyn std::error::Error + Send + Sync>;
 type HTTPResult = Result<Response<BoxBody<Bytes, BoxError>>, BoxError>;
 
+#[derive(Debug, PartialEq, Clone)]
+enum AcceptType {
+    Ndjson,
+    EventStream,
+}
+
 enum Routes {
-    StreamCat,
+    StreamCat(AcceptType),
     StreamAppend(String),
     StreamItemGet(Scru128Id),
     StreamItemRemove(Scru128Id),
@@ -38,9 +45,15 @@ enum Routes {
     NotFound,
 }
 
-fn match_route(method: &Method, path: &str) -> Routes {
+fn match_route(method: &Method, path: &str, headers: &hyper::HeaderMap) -> Routes {
     match (method, path) {
-        (&Method::GET, "/") => Routes::StreamCat,
+        (&Method::GET, "/") => {
+            let accept_type = match headers.get(ACCEPT) {
+                Some(accept) if accept == "text/event-stream" => AcceptType::EventStream,
+                _ => AcceptType::Ndjson,
+            };
+            Routes::StreamCat(accept_type)
+        }
 
         (&Method::POST, p) if p.starts_with("/pipe/") => {
             if let Some(id_str) = p.strip_prefix("/pipe/") {
@@ -100,23 +113,16 @@ async fn handle(
 ) -> HTTPResult {
     let method = req.method();
     let path = req.uri().path();
+    let headers = req.headers().clone();
 
-    match match_route(method, path) {
-        Routes::StreamCat => {
+    match match_route(method, path, &headers) {
+        Routes::StreamCat(accept_type) => {
             let options = match ReadOptions::from_query(req.uri().query()) {
                 Ok(opts) => opts,
                 Err(err) => return response_400(err.to_string()),
             };
 
-            let rx = store.read(options).await;
-            let stream = ReceiverStream::new(rx);
-            let stream = stream.map(|frame| {
-                let mut encoded = serde_json::to_vec(&frame).unwrap();
-                encoded.push(b'\n');
-                Ok(hyper::body::Frame::data(bytes::Bytes::from(encoded)))
-            });
-            let body = StreamBody::new(stream).boxed();
-            Ok(Response::new(body))
+            handle_stream_cat(&mut store, options, accept_type).await
         }
 
         Routes::StreamAppend(topic) => handle_stream_append(&mut store, req, topic).await,
@@ -149,6 +155,46 @@ async fn handle(
     }
 }
 
+async fn handle_stream_cat(
+    store: &mut Store,
+    options: ReadOptions,
+    accept_type: AcceptType,
+) -> HTTPResult {
+    let rx = store.read(options).await;
+    let stream = ReceiverStream::new(rx);
+
+    let accept_type_clone = accept_type.clone();
+    let stream = stream.map(move |frame| {
+        let bytes = match accept_type_clone {
+            AcceptType::Ndjson => {
+                let mut encoded = serde_json::to_vec(&frame).unwrap();
+                encoded.push(b'\n');
+                encoded
+            }
+            AcceptType::EventStream => format!(
+                "event: {}\nid: {}\ndata: {}\n\n",
+                frame.topic,
+                frame.id,
+                serde_json::to_string(&frame).unwrap_or_default()
+            )
+            .into_bytes(),
+        };
+        Ok(hyper::body::Frame::data(Bytes::from(bytes)))
+    });
+
+    let body = StreamBody::new(stream).boxed();
+
+    let content_type = match accept_type {
+        AcceptType::Ndjson => "application/x-ndjson",
+        AcceptType::EventStream => "text/event-stream",
+    };
+
+    Ok(Response::builder()
+        .status(StatusCode::OK)
+        .header("Content-Type", content_type)
+        .body(body)?)
+}
+
 async fn handle_stream_append(
     store: &mut Store,
     req: Request<hyper::body::Incoming>,

src/client.rs

@@ -51,6 +51,7 @@ pub async fn cat(
     tail: bool,
     last_id: Option<String>,
     limit: Option<u64>,
+    sse: bool,
 ) -> Result<Receiver<Bytes>, BoxError> {
     let stream = connect(addr).await?;
     let io = TokioIo::new(stream);
@@ -91,10 +92,13 @@ pub async fn cat(
         "http://localhost/".to_string()
     };
 
-    let req = Request::builder()
-        .method(Method::GET)
-        .uri(uri)
-        .body(empty())?;
+    let mut req = Request::builder().method(Method::GET).uri(uri);
+
+    if sse {
+        req = req.header("Accept", "text/event-stream");
+    }
+
+    let req = req.body(empty())?;
 
     let res = sender.send_request(req).await?;
 

src/main.rs

@@ -73,6 +73,10 @@ struct CommandCat {
     /// Limit the number of events
     #[clap(long)]
     limit: Option<u64>,
+
+    /// Use Server-Sent Events format
+    #[clap(long)]
+    sse: bool,
 }
 
 #[derive(Parser, Debug)]
@@ -184,6 +188,7 @@ async fn cat(args: CommandCat) -> Result<(), Box<dyn std::error::Error + Send +
         args.tail,
         args.last_id.clone(),
         args.limit,
+        args.sse,
     )
     .await?;
     let mut stdout = tokio::io::stdout();