docs: README streaming/SSE sections, ROADMAP v0.3 done, crud SSE ticker (v0.3 chunk 4)
README: write_timeout in the Config sample, read/write timeout-split semantics spelled out, new Streaming Responses and Server-Sent Events sections (producer contract, 1.0 fallback, heartbeat liveness), roadmap summary bumped. ROADMAP: v0.3 marked done with the as-landed decisions (pull shape per the user's fork answer; Q3 timeout split as proposed) and verified exit criteria. crud example: GET /ticker SSE route — one event/second, producer exits on SseClosed or the example's shutdown flag (so an open ticker doesn't block AllDone past the drain).
This commit is contained in:
@@ -132,7 +132,8 @@ let cfg = Config {
|
||||
listener_pool: 2, // Supervised accept-loop actors
|
||||
scheduler_threads: Some(2), // smarm worker threads (None = one per CPU)
|
||||
keep_alive_timeout: Duration::from_secs(60), // Idle budget between requests
|
||||
request_timeout: Duration::from_secs(30), // Whole-request read deadline
|
||||
request_timeout: Duration::from_secs(30), // Whole-request READ deadline
|
||||
write_timeout: Duration::from_secs(30), // Per-write response budget
|
||||
max_header_count: 64,
|
||||
read_buf_size: 8 * 1024,
|
||||
max_body_bytes: 1 << 20,
|
||||
@@ -152,7 +153,14 @@ serve_with(cfg, pipeline).unwrap();
|
||||
until its head and body are fully read. Expiry mid-head gets a
|
||||
best-effort `408 Request Timeout`; expiry mid-body just closes. Deadlines
|
||||
are absolute instants, so a trickling (slowloris-style) client can't
|
||||
reset its budget by sending one byte at a time.
|
||||
reset its budget by sending one byte at a time. Covers the **read phase
|
||||
only** — a streaming response may legitimately outlive any whole-request
|
||||
clock.
|
||||
- `write_timeout` — a per-write budget for response bytes: the fixed
|
||||
head+body write, and *each* streamed chunk, must complete within it. A
|
||||
client that stops reading is dropped once its socket buffer fills and a
|
||||
write stalls past the budget (the write-side twin of slowloris). Streamed
|
||||
chunks each get a fresh budget; a stream as a whole has no deadline.
|
||||
|
||||
### Graceful Shutdown
|
||||
|
||||
@@ -185,6 +193,62 @@ If every `Handle` is dropped, shutdown can never be signalled and the
|
||||
server runs forever — exactly `serve_with`'s semantics (it does this
|
||||
internally).
|
||||
|
||||
### Streaming Responses
|
||||
|
||||
A handler can return a body it doesn't have yet: hand urus the read end of
|
||||
a channel and feed it from a producer actor (the conn actor owns the
|
||||
socket and all write deadlines; the producer never touches the fd):
|
||||
|
||||
```rust
|
||||
use urus::{Conn, Next};
|
||||
|
||||
|c: Conn, _n: Next| {
|
||||
let (tx, rx) = smarm::channel::<Vec<u8>>();
|
||||
smarm::spawn(move || {
|
||||
for part in ["chunk one", "chunk two"] {
|
||||
if tx.send(part.as_bytes().to_vec()).is_err() {
|
||||
return; // connection died — stop producing
|
||||
}
|
||||
}
|
||||
// tx drops: end of stream.
|
||||
});
|
||||
c.put_status(200).put_body(rx)
|
||||
}
|
||||
```
|
||||
|
||||
On HTTP/1.1 the response uses `Transfer-Encoding: chunked` and the
|
||||
connection is reusable afterwards; on HTTP/1.0 (no chunked framing) bytes
|
||||
go out raw and the connection closes to delimit the body. End of stream is
|
||||
signalled by dropping every `Sender`. The producer contract: a `send`
|
||||
error means the connection is gone — exit. Chunked **request** bodies are
|
||||
decoded transparently; handlers see `conn.body` either way.
|
||||
|
||||
### Server-Sent Events
|
||||
|
||||
`Conn::sse()` is the streaming body with the SSE headers and a heartbeat
|
||||
wired up:
|
||||
|
||||
```rust
|
||||
|c: Conn, _n: Next| {
|
||||
let (c, events) = c.sse(); // or c.sse_with_heartbeat(interval)
|
||||
smarm::spawn(move || {
|
||||
loop {
|
||||
if events.send("tick", "data").is_err() {
|
||||
return; // SseClosed: client gone or server draining
|
||||
}
|
||||
smarm::sleep(std::time::Duration::from_secs(1));
|
||||
}
|
||||
});
|
||||
c
|
||||
}
|
||||
```
|
||||
|
||||
While the producer is silent the connection actor emits `: keep-alive`
|
||||
comment chunks every `HEARTBEAT_INTERVAL` (15s default). Liveness comes
|
||||
from that ping: a vanished client is detected when a write stalls past
|
||||
`write_timeout`. No request clock applies to an open SSE stream, and a
|
||||
graceful shutdown force-stops it at the drain deadline.
|
||||
|
||||
### Named Actors
|
||||
|
||||
For introspection and debugging, the server registers itself in smarm's
|
||||
@@ -262,9 +326,9 @@ This avoids the complexity of async ecosystems while maintaining full concurrenc
|
||||
## Roadmap
|
||||
|
||||
See [`ROADMAP.md`](ROADMAP.md). Summary: v0.2 (supervised listener pool,
|
||||
graceful shutdown, enforced timeouts, registry names) is done; next up are
|
||||
streaming bodies + SSE (v0.3), WebSocket (v0.4), PubSub (v0.5), and
|
||||
Phoenix-style channels (v0.6).
|
||||
graceful shutdown, enforced timeouts, registry names) and v0.3 (streaming
|
||||
bodies, chunked requests, SSE) are done; next up are WebSocket (v0.4),
|
||||
PubSub (v0.5), and Phoenix-style channels (v0.6).
|
||||
|
||||
Refer to `urus-spec.md` and `urus-v1-build-notes.md` in the artifact persistence for the original design and implementation notes.
|
||||
|
||||
|
||||
+33
-12
@@ -119,22 +119,43 @@ panicking listener restarts under load without dropped accepts (test each).
|
||||
|
||||
---
|
||||
|
||||
## v0.3 — Streaming bodies + SSE
|
||||
## v0.3 — Streaming bodies + SSE ✅ (landed 2026-06, chunks 1-3)
|
||||
|
||||
The slipped v1 goal, and the WebSocket prerequisite (an upgraded connection
|
||||
is "a response that never ends").
|
||||
|
||||
1. `RespBody::Stream(...)` — handler hands back a writer callback or a
|
||||
`Receiver<Vec<u8>>`; conn actor switches to chunked transfer-encoding
|
||||
(HTTP/1.1) and pumps until the stream closes. Keep-alive after a
|
||||
completed chunked response.
|
||||
2. Chunked **request** bodies (`parser.rs` currently returns 411) — same
|
||||
cycle, the codec knowledge is fresh.
|
||||
3. SSE sugar: `Conn::sse()` → an `EventSender` (`.send(event, data)`),
|
||||
sets `content-type: text/event-stream`, disables buffering, heartbeats
|
||||
via comment lines on a timer. An SSE conn is just a Stream body with
|
||||
the per-read deadline disabled (or set to the heartbeat interval) —
|
||||
liveness comes from the keep-alive ping, not `request_timeout`.
|
||||
1. `RespBody::Stream(StreamBody)` ✅ — **decided: pull** (user call at the
|
||||
v0.3 design fork): the handler hands back a `smarm::Receiver<Vec<u8>>`
|
||||
and the producer is an actor the handler spawned; the conn actor pumps
|
||||
(`pump_stream`) and keeps sole ownership of the socket and write
|
||||
deadlines. The recv() park between chunks is stoppable by the draining
|
||||
registry, so infinite streams die at the drain deadline with no
|
||||
polling. HTTP/1.1: chunked TE, keep-alive after the terminator;
|
||||
HTTP/1.0: raw bytes, forced close, EOF-delimited. End of stream =
|
||||
every Sender dropped = 0-chunk. Timeout split (handoff Q3, as
|
||||
proposed): `request_timeout` covers the READ phase only; new
|
||||
`Config.write_timeout` (default 30s) bounds every response write —
|
||||
fresh budget per streamed chunk, and the fixed-body + error-response
|
||||
writes went deadline-bounded too (write-side slowloris closed).
|
||||
2. Chunked **request** bodies ✅ — `read_chunked_body` decodes
|
||||
incrementally on the request deadline; extensions ignored, trailers
|
||||
consumed + discarded, decoded size capped at max_body_bytes (413),
|
||||
size-line/trailer spam capped (400). CL + TE: chunked, or chunked on
|
||||
HTTP/1.0, rejected 400 at parse (smuggling ambiguity). Keep-alive
|
||||
drain accounts RAW framing length, so pipelined requests behind a
|
||||
chunked body land exactly.
|
||||
3. SSE sugar ✅ — `Conn::sse()` / `sse_with_heartbeat` → `EventSender`
|
||||
(`send(event, data)` / `data` / `comment`, Clone, `Err(SseClosed)`
|
||||
once the stream is gone). Heartbeat lives in the pump as leaned: with
|
||||
`StreamBody.heartbeat` set, the chunk wait is `recv_timeout(interval)`
|
||||
and expiry writes a `: keep-alive` comment chunk. Liveness from the
|
||||
ping (dead client = heartbeat write stalls past write_timeout), not
|
||||
from any request clock.
|
||||
|
||||
**Exit criteria (verified):** chunked round-trips both directions with
|
||||
keep-alive and pipelining intact; an infinite SSE stream heartbeats while
|
||||
silent, is killed by write_timeout when the client stops reading, and is
|
||||
force-stopped at the drain deadline on shutdown; full suite green.
|
||||
|
||||
---
|
||||
|
||||
|
||||
+24
-1
@@ -204,6 +204,28 @@ fn parse_id(s: &str) -> Option<u64> {
|
||||
// Handlers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// SSE demo: `curl -N localhost:8080/ticker` streams a tick every second
|
||||
/// (with `: keep-alive` comments if it ever goes quiet). The producer
|
||||
/// exits on SseClosed (client gone / write timeout / shutdown drain) or
|
||||
/// when the example is shutting down.
|
||||
fn ticker(conn: Conn, _next: Next) -> Conn {
|
||||
let (conn, events) = conn.sse();
|
||||
smarm::spawn(move || {
|
||||
let mut n: u64 = 0;
|
||||
loop {
|
||||
if SHUTTING_DOWN.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
return; // dropping `events` ends the stream cleanly
|
||||
}
|
||||
if events.send("tick", &n.to_string()).is_err() {
|
||||
return; // SseClosed
|
||||
}
|
||||
n += 1;
|
||||
smarm::sleep(std::time::Duration::from_secs(1));
|
||||
}
|
||||
});
|
||||
conn
|
||||
}
|
||||
|
||||
fn list(conn: Conn, _next: Next) -> Conn {
|
||||
let (tx, rx) = channel::<(u16, Vec<u8>)>();
|
||||
store().send(Request::List { reply: tx }).ok();
|
||||
@@ -278,7 +300,8 @@ fn main() {
|
||||
.post( "/users", create)
|
||||
.get( "/users/:id", get_one)
|
||||
.put( "/users/:id", update)
|
||||
.delete("/users/:id", delete),
|
||||
.delete("/users/:id", delete)
|
||||
.get( "/ticker", ticker),
|
||||
);
|
||||
|
||||
let cfg = Config::new("127.0.0.1:8080".parse().unwrap());
|
||||
|
||||
Reference in New Issue
Block a user