⚡ p3: Rust-native SSE streaming (sse_stream_response, ADR-088)

swadhinbiswas committed 2 months ago

1281187

1 parent f5b3f46

commit 1281187

4 files changed

+19 -7

Files

crates

justapi-core

src

server

tests

Showing 4 changed files+19-7

▼crates/justapi-core/src/multipart.rs

+2-2

@@ -8,7 +8,7 @@ use multer::Multipart;
8	8	use std::path::PathBuf;
9	9
10	10	/// A file uploaded via `multipart/form-data`.
11		-#[derive(Debug)]
	11	+#[derive(Debug, Clone)]
12	12	pub struct UploadFile {
13	13	pub field_name: String,
14	14	pub filename: Option<String>,
@@ -21,7 +21,7 @@ pub struct UploadFile {
21	21	}
22	22
23	23	/// Parsed multipart form data containing both text fields and uploaded files.
24		-#[derive(Debug, Default)]
	24	+#[derive(Debug, Default, Clone)]
25	25	pub struct MultipartForm {
26	26	pub fields: HashMap<String, String>,
27	27	pub files: Vec<UploadFile>,
28	28

▼crates/justapi-core/src/server/mod.rs

+1-1

@@ -47,7 +47,7 @@ pub use ws::{default_ws_echo, WsConnInfo, WsHandler, WsRead, WsWrite};
47	47	mod sse_ws;
48	48	#[cfg(feature = "ws")]
49	49	pub use sse_ws::dispatch_ws;
50		-pub use sse_ws::sse_response;
	50	+pub use sse_ws::{sse_response, sse_stream_response};
51	51
52	52	mod handler_exec;
53	53	use handler_exec::execute_handler;
54	54

▼crates/justapi-core/src/server/sse_ws.rs

+15-3

@@ -10,15 +10,27 @@ use super::{WsConnInfo, WsHandler, WsRead, WsWrite};
10	10
11	11	/// Build a demo SSE endpoint that sends 10 events at 100ms intervals.
12	12	pub fn sse_response() -> Response<ResponseBody> {
	13	+ sse_stream_response(10, 100)
	14	+}
	15	+
	16	+/// Rust-native SSE stream: emits `count` events at `interval_ms` spacing,
	17	+/// entirely in Rust (tokio + mpsc + stream). No Python, no GIL — this is the
	18	+/// "server runs on Rust" streaming path (ADR-088).
	19	+///
	20	+/// Each event is `data: {"n":<i>}\n\n`. The producer task runs on the tokio
	21	+/// runtime; the response body is an mpsc-backed stream the HTTP layer drains.
	22	+pub fn sse_stream_response(count: u64, interval_ms: u64) -> Response<ResponseBody> {
13	23	let (tx, rx) = tokio::sync::mpsc::channel::<Result<Bytes, anyhow::Error>>(16);
14	24
15	25	tokio::spawn(async move {
16		- for i in 1..=10 {
17		- let msg = format!("data: {{\"count\":{}}}\n\n", i);
	26	+ for i in 1..=count {
	27	+ let msg = format!("data: {{\"n\":{}}}\n\n", i);
18	28	if tx.send(Ok(Bytes::from(msg))).await.is_err() {
19	29	break;
20	30	}
21		- tokio::time::sleep(std::time::Duration::from_millis(100)).await;
	31	+ if interval_ms > 0 {
	32	+ tokio::time::sleep(std::time::Duration::from_millis(interval_ms)).await;
	33	+ }
22	34	}
23	35	});
24	36
25	37

▼crates/justapi-core/tests/integration.rs

+1-1

@@ -95,7 +95,7 @@ async fn test_loopback_sse() {
95	95	assert_eq!(status, StatusCode::OK);
96	96	let body_str = String::from_utf8(body).unwrap();
97	97	assert!(body_str.contains("data:"));
98		- assert!(body_str.contains("\"count\":10"));
	98	+ assert!(body_str.contains("\"n\":10"));
99	99	}
100	100
101	101	// --- Phase 9: Compression tests ---
102	102