Cameleeringsimoom

How do I stream bodies and server-sent events with simoom?

Reply::stream and Reply::sse return a BodySink you write from any thread; large uploads arrive through a BodySource read off the loop thread.

Some responses are too large to build in memory, or never end: a log export, a progress feed, a live dashboard. Some uploads are too large to buffer. simoom streams both directions with backpressure, so a slow client slows the writer instead of filling memory.

Streaming a response

reply.stream() sends the status and headers at once and returns a BodySink for the body. The body goes out chunked on HTTP/1.1 and as DATA frames on HTTP/2 and HTTP/3. BodySink is Send and implements std::io::Write, so a thread can write to it:

rust
use simoom::{Outcome, Reply, Request};
use std::io::Write;

fn numbers(_: &mut Request<'_>, reply: Reply<'_>) -> Outcome {
    let mut sink = reply.header("content-type", "text/plain").stream();
    std::thread::spawn(move || {
        for i in 0..1000 {
            // write waits while 64 KiB are pending, so a slow client slows this thread
            if writeln!(sink, "line {i}").is_err() {
                return; // the client went away
            }
        }
        sink.finish();
    });
    Outcome::Done
}

At most Limits::stream_buffer bytes (64 KiB by default) are handed to the connection and not yet sent. Past that, write waits. That is fine on your own thread or an offload worker and wrong on the loop thread, where it would stop the loop. On the loop thread use try_write, which never waits and fails with WouldBlock when the buffer is full:

rust
use simoom::{Outcome, Reply, Request};

fn small(_: &mut Request<'_>, reply: Reply<'_>) -> Outcome {
    let mut sink = reply.stream();
    // on the loop thread: try_write never waits
    let _ = sink.try_write(b"first part, ");
    let _ = sink.try_write(b"second part");
    sink.finish();
    Outcome::Done
}

BodySink also has poll_ready and a ready() future for code that runs its own executor; no runtime is attached.

Ending the body:

  • finish(), or dropping the sink, ends it normally.
  • finish_with_trailers(&[("grpc-status", "0")]) ends it with trailer fields (trailing HEADERS on HTTP/2 and HTTP/3, a trailer section on HTTP/1.1).
  • abort() cuts it off: the HTTP/1.1 connection closes, or the HTTP/2 or HTTP/3 stream is reset, so the client knows the body is incomplete.
  • is_closed() tells you the client is gone; write then fails with BrokenPipe.
rust
use simoom::{Outcome, Reply, Request};

fn grpc_like(_: &mut Request<'_>, reply: Reply<'_>) -> Outcome {
    let mut sink = reply.header("content-type", "application/grpc").stream();
    std::thread::spawn(move || {
        let _ = sink.write(b"\0\0\0\0\0");
        sink.finish_with_trailers(&[("grpc-status", "0")]);
    });
    Outcome::Done
}

Server-sent events

reply.sse() is stream() with content-type: text/event-stream and cache-control: no-cache. send_event writes one event: each line of the text becomes a data: line, followed by a blank line.

rust
use simoom::{Outcome, Reply, Request, Router};
use std::time::Duration;

fn router() -> Router {
    Router::new().get("/events", |_: &mut Request<'_>, r: Reply<'_>| -> Outcome {
        let mut sink = r.sse();
        std::thread::spawn(move || {
            for i in 0.. {
                if sink.send_event(&format!("tick {i}")).is_err() {
                    return; // the browser closed the page
                }
                std::thread::sleep(Duration::from_secs(1));
            }
        });
        Outcome::Done
    })
}

In the browser, new EventSource("/events") receives the ticks. For named events or ids, write the fields yourself with sink.write(b"event: update\ndata: ...\n\n").

A thread per stream is simple and fine for tens or hundreds of streams. For many thousands, keep the sinks in one place and have a single thread (or a timer) write to all of them.

Long-lived streams over HTTP/2 can outlive a client that vanished without closing its connection. The h2-keepalive feature with Limits::h2_keepalive sends PINGs to find such clients (HTTP/2 and HTTP/3).

Request bodies

Bodies up to Limits::body_buffer (64 KiB by default) are read whole before your service is called, and req.body() returns them:

rust
use simoom::{Outcome, Reply, Request};

fn upload(req: &mut Request<'_>, reply: Reply<'_>) -> Outcome {
    if req.is_streamed() {
        return reply.status(413).body_static(b"send less than 64 KiB here");
    }
    let n = req.body().len();
    reply.body(format!("got {n} bytes"))
}

Larger bodies, and bodies of unknown length, stream: req.is_streamed() is true, req.body() is empty, and req.take_body() returns a BodySource. The connection reads from the socket only while the source has room, so a slow reader slows the client down (pull-based backpressure).

BodySource implements std::io::Read, which blocks, so read it on the offload pool or your own thread:

rust
use simoom::{blocking, OwnedRequest, Response};
use std::io::Read;

fn service() -> impl simoom::Service {
    blocking(|mut req: OwnedRequest| {
        let mut total = 0u64;
        if let Some(mut body) = req.take_body() {
            let mut buf = [0u8; 16 * 1024];
            loop {
                match body.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => total += n as u64, // hash it, write it to disk, ...
                    Err(_) => return Response::new(400).body("upload broken off"),
                }
            }
        } else {
            total = req.body().len() as u64;
        }
        Response::text(format!("received {total} bytes"))
    })
}

When you just want the whole body, OwnedRequest::read_body() returns the buffered body or reads the stream to its end:

rust
use simoom::{blocking, OwnedRequest, Response};

fn service() -> impl simoom::Service {
    blocking(|mut req: OwnedRequest| match req.read_body() {
        Ok(body) => Response::text(format!("{} bytes", body.len())),
        Err(_) => Response::new(400).body("bad body"),
    })
}

BodySource::try_read and poll_read don't block, if you need to read on the loop thread. Trailers of a streamed body are available from BodySource::trailers() once it is complete.

Limits that apply to request bodies:

  • Limits::max_body (10 MiB): a larger Content-Length is refused with 413 before any body byte is read; a chunked body that grows past it fails the read.
  • Limits::min_body_rate (1 KiB/s after a 5-second grace): a client that sends slower is cut off.

Common questions

Can I stream a file to the client?

Yes: open it on your own thread or an offload worker and copy it into the sink with std::io::copy(&mut file, &mut sink). simoom has no static-file server, so paths, content types and range requests are up to you (the conditional feature helps with ETags and ranges).

Does SSE work over HTTP/2 and HTTP/3?

Yes. The same sse() call works on every protocol; on HTTP/2 and HTTP/3 many event streams share one connection.

How do I know the client disconnected?

write and send_event return an error, and is_closed() returns true. Stop producing and drop the sink.

Can response compression apply to streams?

Yes, with the compression feature and Compression::new().streams(true); each compressed stream then holds a compressor of about 256 KiB while it runs.

Edit this page on GitHub