Build v0 example and SSE slices

slhx-axum/src/lib.rs

@@ -2,7 +2,9 @@ use axum::async_trait;
 use axum::body::{to_bytes, Body};
 use axum::extract::{FromRequest, FromRequestParts, Multipart};
 use axum::http::{header, request::Parts, HeaderMap, HeaderValue, Request, Response, StatusCode};
+use axum::response::sse::{Event, Sse};
 use axum::response::IntoResponse;
+use futures_util::{Stream, StreamExt};
 use slhx_core::{BuildFingerprint, EffectBatch, IntoEffect};
 use std::collections::BTreeMap;
 use std::convert::Infallible;
@@ -13,6 +15,7 @@ pub const SLHX_TITLE_HEADER: &str = "x-slhx-title";
 pub const SLHX_CONTENT_TYPE: &str = "application/slhx";
 pub const SLHX_HANDLE_FIELD: &str = "__h";
 pub const SLHX_RUNTIME_CONTENT_TYPE: &str = "application/javascript; charset=utf-8";
+pub const SLHX_SSE_EVENT: &str = "slhx";
 
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 pub struct RuntimeJs;
@@ -426,6 +429,54 @@ pub fn runtime_js_source() -> &'static str {
     slhx_js::RUNTIME_JS
 }
 
+// req: push/001, req: push/003, req: push/004
+pub fn sse<S, E>(batches: S) -> Sse<impl Stream<Item = Result<Event, E>> + Send>
+where
+    S: Stream<Item = Result<EffectBatch, E>> + Send + 'static,
+    E: Into<axum::BoxError>,
+{
+    Sse::new(batches.map(|batch| batch.map(sse_event)))
+}
+
+pub fn sse_event(batch: EffectBatch) -> Event {
+    Event::default()
+        .event(SLHX_SSE_EVENT)
+        .data(encode_sse_batch(&batch))
+}
+
+pub fn encode_sse_batch(batch: &EffectBatch) -> String {
+    base64_url_no_pad(&batch.to_wire())
+}
+
+fn base64_url_no_pad(input: &[u8]) -> String {
+    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
+    let mut out = String::with_capacity((input.len() * 4).div_ceil(3));
+    let mut chunks = input.chunks_exact(3);
+    for chunk in &mut chunks {
+        let n = ((chunk[0] as u32) << 16) | ((chunk[1] as u32) << 8) | chunk[2] as u32;
+        out.push(ALPHABET[((n >> 18) & 0x3f) as usize] as char);
+        out.push(ALPHABET[((n >> 12) & 0x3f) as usize] as char);
+        out.push(ALPHABET[((n >> 6) & 0x3f) as usize] as char);
+        out.push(ALPHABET[(n & 0x3f) as usize] as char);
+    }
+    match chunks.remainder() {
+        [a] => {
+            let n = (*a as u32) << 16;
+            out.push(ALPHABET[((n >> 18) & 0x3f) as usize] as char);
+            out.push(ALPHABET[((n >> 12) & 0x3f) as usize] as char);
+        }
+        [a, b] => {
+            let n = ((*a as u32) << 16) | ((*b as u32) << 8);
+            out.push(ALPHABET[((n >> 18) & 0x3f) as usize] as char);
+            out.push(ALPHABET[((n >> 12) & 0x3f) as usize] as char);
+            out.push(ALPHABET[((n >> 6) & 0x3f) as usize] as char);
+        }
+        [] => {}
+        _ => unreachable!(),
+    }
+    out
+}
+
 fn parse_urlencoded_pairs(body: &[u8]) -> Result<Vec<(String, String)>, InteractionFormRejection> {
     if body.is_empty() {
         return Ok(Vec::new());
@@ -479,8 +530,11 @@ fn hex(byte: u8) -> Option<u8> {
 
 #[cfg(test)]
 mod tests {
-    use super::{html_with_root_fingerprint, BuildFingerprint, InteractionForm};
-    use axum::{body::Body, extract::FromRequest, http::Request};
+    use super::{encode_sse_batch, html_with_root_fingerprint, sse, BuildFingerprint, InteractionForm, SLHX_SSE_EVENT};
+    use axum::{body::{to_bytes, Body}, extract::FromRequest, http::{header, Request}, response::IntoResponse};
+    use futures_util::stream;
+    use slhx_core::{EffectBatch, EFFECT_BATCH_ABI_VERSION};
+    use std::convert::Infallible;
 
     #[test]
     fn root_fingerprint_is_added_to_initial_root() {
@@ -505,6 +559,29 @@ mod tests {
         assert_eq!(html, "<main data-slhx-root data-slhx-fp=\"1\">Docs</main>");
     }
 
+    #[tokio::test]
+    async fn sse_response_streams_base64_url_effect_batches() {
+        let batch = EffectBatch {
+            abi_version: EFFECT_BATCH_ABI_VERSION,
+            fingerprint: BuildFingerprint(11),
+            ops: Vec::new(),
+        };
+        let encoded = encode_sse_batch(&batch);
+
+        let response = sse(stream::iter([Ok::<_, Infallible>(batch)])).into_response();
+        assert_eq!(
+            response.headers().get(header::CONTENT_TYPE).unwrap(),
+            "text/event-stream"
+        );
+        let body = to_bytes(response.into_body(), 1024).await.unwrap();
+
+        assert_eq!(
+            String::from_utf8(body.to_vec()).unwrap(),
+            format!("event: {SLHX_SSE_EVENT}\ndata: {encoded}\n\n")
+        );
+        assert!(!encoded.contains('='));
+    }
+
     #[tokio::test]
     async fn interaction_form_extracts_multipart_fields_and_files() {
         let boundary = "slhx-test-boundary";