fix: compose SSE reload-abort with brotli compression

SSE streams with brotli compression were not being cancelled on hot
reload because the two concerns were in mutually exclusive branches.
Restructure stream building as composable layers: reload-abort first,
then optional brotli compression on top.

Author: cablehead

README.md

@@ -133,7 +133,8 @@ Or from a file:
 $ http-nu :3001 ./serve.nu
 ```
 
-Try the [live examples](https://http-nu.cross.stream/examples/) or run them locally with the [examples hub](examples/README.md):
+Try the [live examples](https://http-nu.cross.stream/examples/) or run them
+locally with the [examples hub](examples/README.md):
 
 ```bash
 $ http-nu --datastar :3001 examples/serve.nu

src/handler.rs

@@ -1,9 +1,10 @@
 use std::net::SocketAddr;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Instant;
 
 use arc_swap::ArcSwap;
-use futures_util::StreamExt;
+use futures_util::{Stream, StreamExt};
 use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full, StreamBody};
 use hyper::body::{Bytes, Frame};
 use tokio_stream::wrappers::ReceiverStream;
@@ -475,30 +476,31 @@ async fn build_normal_response(
             }
         }
         ResponseTransport::Stream(rx) => {
-            if use_brotli {
-                compression::compress_stream(rx)
-            } else if is_sse {
+            // Layer 1: base byte stream, with reload-abort for SSE
+            let byte_stream: Pin<Box<dyn Stream<Item = Vec<u8>> + Send + Sync>> = if is_sse {
                 // SSE streams abort on reload (error triggers client retry)
-                let stream = futures_util::stream::try_unfold(
+                Box::pin(futures_util::stream::unfold(
                     (ReceiverStream::new(rx), reload_token),
                     |(mut data_rx, token)| async move {
                         tokio::select! {
                             biased;
-                            _ = token.cancelled() => {
-                                Err(std::io::Error::other("reload").into())
-                            }
+                            _ = token.cancelled() => None,
                             item = StreamExt::next(&mut data_rx) => {
-                                match item {
-                                    Some(data) => Ok(Some((Frame::data(Bytes::from(data)), (data_rx, token)))),
-                                    None => Ok(None),
-                                }
+                                item.map(|data| (data, (data_rx, token)))
                             }
                         }
                     },
-                );
-                BodyExt::boxed(StreamBody::new(stream))
+                ))
+            } else {
+                Box::pin(ReceiverStream::new(rx))
+            };
+
+            // Layer 2: optionally compress, then frame
+            if use_brotli {
+                let brotli = compression::BrotliStream::new(byte_stream);
+                BodyExt::boxed(StreamBody::new(brotli))
             } else {
-                let stream = ReceiverStream::new(rx).map(|data| Ok(Frame::data(Bytes::from(data))));
+                let stream = byte_stream.map(|data| Ok(Frame::data(Bytes::from(data))));
                 BodyExt::boxed(StreamBody::new(stream))
             }
         }

tests/server_test.rs

@@ -2109,6 +2109,96 @@ async fn test_sse_cancelled_on_hot_reload() {
     child.kill().await.ok();
 }
 
+#[tokio::test]
+async fn test_sse_cancelled_on_hot_reload_with_brotli() {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    // Spawn server with an SSE endpoint that streams indefinitely
+    let (mut child, mut stdin, addr_rx) = TestServerWithStdin::spawn("127.0.0.1:0", false);
+
+    // Send initial SSE script
+    let sse_script = r#"{|req|
+        1..100 | each {|i|
+            sleep 100ms
+            {data: $"event-($i)"}
+        } | to sse
+    }"#;
+    stdin.write_all(sse_script.as_bytes()).await.unwrap();
+    stdin.write_all(b"\0").await.unwrap();
+    stdin.flush().await.unwrap();
+
+    let address = tokio::time::timeout(std::time::Duration::from_secs(5), addr_rx)
+        .await
+        .expect("Server didn't start")
+        .expect("Channel closed");
+
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    // Start curl with brotli compression
+    let mut sse_child = tokio::process::Command::new("curl")
+        .arg("-sN")
+        .arg("-H")
+        .arg("Accept-Encoding: br")
+        .arg(&address)
+        .stdout(std::process::Stdio::piped())
+        .spawn()
+        .expect("Failed to start curl");
+
+    let stdout = sse_child.stdout.take().expect("Failed to get stdout");
+    let mut reader = stdout;
+
+    // Read some data to confirm SSE is streaming
+    let mut buf = vec![0u8; 256];
+    let n = tokio::time::timeout(std::time::Duration::from_secs(2), reader.read(&mut buf))
+        .await
+        .expect("Timeout reading initial SSE data")
+        .expect("Read error");
+    assert!(n > 0, "Should have received SSE data");
+
+    // Trigger hot reload
+    let new_script = r#"{|req| "reloaded"}"#;
+    stdin.write_all(new_script.as_bytes()).await.unwrap();
+    stdin.write_all(b"\0").await.unwrap();
+    stdin.flush().await.unwrap();
+
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    // After reload, the compressed SSE stream should end.
+    // Drain remaining data (brotli FINISH frame) and check for EOF.
+    let stream_ended = tokio::time::timeout(std::time::Duration::from_secs(2), async {
+        loop {
+            match reader.read(&mut buf).await {
+                Ok(0) => return true,  // EOF - stream ended
+                Ok(_) => continue,     // Drain remaining data (e.g. brotli FINISH frame)
+                Err(_) => return true, // Read error
+            }
+        }
+    })
+    .await;
+
+    assert!(
+        matches!(stream_ended, Ok(true)),
+        "Compressed SSE stream should stop after reload"
+    );
+
+    sse_child.kill().await.ok();
+
+    // Verify the new handler works
+    let output = std::process::Command::new("curl")
+        .arg("-s")
+        .arg(&address)
+        .output()
+        .expect("curl failed");
+
+    assert_eq!(
+        String::from_utf8_lossy(&output.stdout).trim(),
+        "reloaded",
+        "New handler should be active after reload"
+    );
+
+    child.kill().await.ok();
+}
+
 /// Tests that --topic with -w loads a handler from the store, serves a placeholder
 /// when the topic is empty, reloads when the topic is appended, and reloads again
 /// when the topic is updated.