Markk116/urus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Markk116
2026-05-26 23:16:45 +02:00
commit 3b6c466210
12 changed files with 2254 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -0,0 +1,2 @@
target
Cargo.lock
Cargo.toml
+30
View File
@@ -0,0 +1,30 @@
[package]
name = "urus"
version = "0.1.0"
edition = "2021"
rust-version = "1.95"
description = "Cowboy/bandit-style HTTP library for the smarm actor runtime"
[dependencies]
smarm = { path = "../smarm" }
httparse = "1.9"
libc = "0.2"
[features]
smarm-trace = ["smarm/smarm-trace"]
[dev-dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
[profile.dev]
panic = "unwind"
[profile.release]
panic = "unwind"
lto = "thin"
codegen-units = 1
[[example]]
name = "crud"
path = "examples/crud.rs"
examples/crud.rs
+270
View File
@@ -0,0 +1,270 @@
//! CRUD example: a tiny user database with JSON persistence.
//!
//! Demonstrates urus and the actor model together:
//! - The pipeline is shared (Arc) across all connection actors.
//! - Handlers do NOT take a lock or share mutable state directly.
//! - A single "store" actor owns the data; handlers send it a request
//! via a channel and block on the reply. Serialization is structural —
//! the store processes one request at a time, no Mutex needed.
//! - On every mutating request the store writes the JSON file. Read
//! requests don't touch disk.
//!
//! Endpoints:
//! GET /users — list
//! POST /users — create
//! GET /users/:id — fetch
//! PUT /users/:id — replace
//! DELETE /users/:id — delete
//!
//! Try it:
//! cargo run --example crud
//! curl -s http://localhost:8080/users
//! curl -s -X POST -d '{"name":"alice","email":"a@x"}' http://localhost:8080/users
//! curl -s http://localhost:8080/users/1
use serde::{Deserialize, Serialize};
use smarm::{channel, Sender};
use std::sync::OnceLock;
use urus::{serve_with, Config, Conn, Next, Pipeline, Router};
// ---------------------------------------------------------------------------
// Domain
// ---------------------------------------------------------------------------
#[derive(Clone, Debug, Serialize, Deserialize)]
struct User {
id: u64,
name: String,
email: String,
}
#[derive(Deserialize)]
struct NewUser {
name: String,
email: String,
}
// ---------------------------------------------------------------------------
// Store actor protocol
// ---------------------------------------------------------------------------
//
// One enum per request kind. Each carries a `reply` Sender for the response.
// Reply types are kept simple: most are JSON byte vectors + HTTP status. The
// store does the serialisation; the handler just writes the bytes.
enum Request {
List { reply: Sender<(u16, Vec<u8>)> },
Get { id: u64, reply: Sender<(u16, Vec<u8>)> },
Create { body: Vec<u8>, reply: Sender<(u16, Vec<u8>)> },
Update { id: u64, body: Vec<u8>, reply: Sender<(u16, Vec<u8>)> },
Delete { id: u64, reply: Sender<(u16, Vec<u8>)> },
}
const DB_PATH: &str = "/tmp/urus-crud.json";
// ---------------------------------------------------------------------------
// Store actor body
// ---------------------------------------------------------------------------
fn store_loop(rx: smarm::Receiver<Request>) {
// Load on start. Missing file = empty store. Corrupt file = panic; we
// don't auto-rebuild because silently losing data is worse than failing
// loud.
let mut users: Vec<User> = match std::fs::read(DB_PATH) {
Ok(bytes) if !bytes.is_empty() =>
serde_json::from_slice(&bytes).expect("urus-crud: db file is not valid JSON"),
_ => Vec::new(),
};
let mut next_id: u64 = users.iter().map(|u| u.id).max().unwrap_or(0) + 1;
loop {
let req = match rx.recv() {
Ok(r) => r,
Err(_) => return, // all senders dropped
};
match req {
Request::List { reply } => {
let body = serde_json::to_vec(&users).unwrap();
let _ = reply.send((200, body));
}
Request::Get { id, reply } => {
match users.iter().find(|u| u.id == id) {
Some(u) => {
let body = serde_json::to_vec(u).unwrap();
let _ = reply.send((200, body));
}
None => {
let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
}
}
}
Request::Create { body, reply } => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => {
let u = User { id: next_id, name: nu.name, email: nu.email };
next_id += 1;
users.push(u.clone());
persist(&users);
let _ = reply.send((201, serde_json::to_vec(&u).unwrap()));
}
Err(_) => {
let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
}
}
}
Request::Update { id, body, reply } => {
match serde_json::from_slice::<NewUser>(&body) {
Ok(nu) => match users.iter_mut().find(|u| u.id == id) {
Some(u) => {
u.name = nu.name;
u.email = nu.email;
let snapshot = u.clone();
persist(&users);
let _ = reply.send((200, serde_json::to_vec(&snapshot).unwrap()));
}
None => {
let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
}
},
Err(_) => {
let _ = reply.send((400, b"{\"error\":\"invalid body\"}".to_vec()));
}
}
}
Request::Delete { id, reply } => {
let before = users.len();
users.retain(|u| u.id != id);
if users.len() < before {
persist(&users);
let _ = reply.send((204, Vec::new()));
} else {
let _ = reply.send((404, b"{\"error\":\"not found\"}".to_vec()));
}
}
}
}
}
fn persist(users: &[User]) {
let json = serde_json::to_vec_pretty(users).unwrap();
// Atomic-ish: write to temp then rename. Avoids half-written files on
// crash. /tmp is on the same filesystem so rename is atomic.
let tmp = format!("{DB_PATH}.tmp");
std::fs::write(&tmp, &json).expect("urus-crud: write tmp failed");
std::fs::rename(&tmp, DB_PATH).expect("urus-crud: rename failed");
}
// ---------------------------------------------------------------------------
// Handler helpers
// ---------------------------------------------------------------------------
//
// Once-cell trick: the store actor is spawned the first time a handler
// runs (smarm requires `spawn` to be called from inside an actor — which
// connection actors are). After that all handlers share the same Sender.
// Simpler than threading the Sender through the pipeline at startup.
static STORE_TX: OnceLock<Sender<Request>> = OnceLock::new();
fn store() -> &'static Sender<Request> {
STORE_TX.get_or_init(|| {
let (tx, rx) = channel::<Request>();
smarm::spawn(move || store_loop(rx));
tx
})
}
fn json(conn: Conn, status: u16, body: Vec<u8>) -> Conn {
conn.put_status(status)
.put_header("content-type", "application/json")
.put_body(body)
}
fn parse_id(s: &str) -> Option<u64> {
s.parse().ok()
}
// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------
fn list(conn: Conn, _next: Next) -> Conn {
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::List { reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
}
fn create(conn: Conn, _next: Next) -> Conn {
let body = conn.body.as_bytes().to_vec();
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Create { body, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
}
fn get_one(conn: Conn, _next: Next) -> Conn {
let id = match conn.params.get("id").and_then(parse_id) {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Get { id, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
}
fn update(conn: Conn, _next: Next) -> Conn {
let id = match conn.params.get("id").and_then(parse_id) {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let body = conn.body.as_bytes().to_vec();
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Update { id, body, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
}
fn delete(conn: Conn, _next: Next) -> Conn {
let id = match conn.params.get("id").and_then(parse_id) {
Some(id) => id,
None => return json(conn, 400, b"{\"error\":\"bad id\"}".to_vec()),
};
let (tx, rx) = channel::<(u16, Vec<u8>)>();
store().send(Request::Delete { id, reply: tx }).ok();
let (status, body) = rx.recv().expect("store dropped");
json(conn, status, body)
}
// ---------------------------------------------------------------------------
// Minimal request logger
// ---------------------------------------------------------------------------
fn logger(conn: Conn, next: Next) -> Conn {
let method = conn.method.as_str().to_string();
let path = conn.path.clone();
let conn = next.run(conn);
println!("{method} {path} -> {}", conn.status.unwrap_or(0));
conn
}
// ---------------------------------------------------------------------------
// main
// ---------------------------------------------------------------------------
fn main() {
let pipeline = Pipeline::new()
.plug(logger)
.plug(
Router::new()
.get( "/users", list)
.post( "/users", create)
.get( "/users/:id", get_one)
.put( "/users/:id", update)
.delete("/users/:id", delete),
);
let cfg = Config::new("127.0.0.1:8080".parse().unwrap());
println!("urus-crud: DB at {DB_PATH}");
serve_with(cfg, pipeline).unwrap();
}
src/conn.rs
+328
View File
@@ -0,0 +1,328 @@
//! `Conn` — the value that flows through the entire plug pipeline.
//!
//! Carries request data, response state being built, and arbitrary user
//! data. Owned and moved at each step. Plugs are unaware of the underlying
//! HTTP version (v1.1 today, /2 later).
use std::any::{Any, TypeId};
use std::collections::HashMap;
// ---------------------------------------------------------------------------
// Method
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Method {
Get,
Post,
Put,
Delete,
Patch,
Head,
Options,
Other(String),
}
impl Method {
pub fn parse(s: &str) -> Self {
match s {
"GET" => Method::Get,
"POST" => Method::Post,
"PUT" => Method::Put,
"DELETE" => Method::Delete,
"PATCH" => Method::Patch,
"HEAD" => Method::Head,
"OPTIONS" => Method::Options,
_ => Method::Other(s.to_string()),
}
}
pub fn as_str(&self) -> &str {
match self {
Method::Get => "GET",
Method::Post => "POST",
Method::Put => "PUT",
Method::Delete => "DELETE",
Method::Patch => "PATCH",
Method::Head => "HEAD",
Method::Options => "OPTIONS",
Method::Other(s) => s.as_str(),
}
}
}
// ---------------------------------------------------------------------------
// HttpVersion
// ---------------------------------------------------------------------------
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HttpVersion {
Http10,
Http11,
}
impl HttpVersion {
pub fn as_str(self) -> &'static str {
match self {
HttpVersion::Http10 => "HTTP/1.0",
HttpVersion::Http11 => "HTTP/1.1",
}
}
}
// ---------------------------------------------------------------------------
// HeaderMap — small-vec of (name, value) string pairs.
// ---------------------------------------------------------------------------
//
// Per spec: most requests have <20 headers, so a linear-scan Vec is cheaper
// than a hash map. Names are case-insensitive (HTTP/1.1), normalised to
// lowercase on insert.
#[derive(Debug, Clone, Default)]
pub struct HeaderMap {
inner: Vec<(String, String)>,
}
impl HeaderMap {
pub fn new() -> Self { Self::default() }
pub fn with_capacity(n: usize) -> Self {
Self { inner: Vec::with_capacity(n) }
}
/// Append a header. Does not deduplicate (HTTP allows repeats; e.g.
/// `Set-Cookie`).
pub fn append(&mut self, name: &str, value: impl Into<String>) {
self.inner.push((name.to_ascii_lowercase(), value.into()));
}
/// Replace any existing values for `name` with a single value.
pub fn set(&mut self, name: &str, value: impl Into<String>) {
let lower = name.to_ascii_lowercase();
self.inner.retain(|(n, _)| n != &lower);
self.inner.push((lower, value.into()));
}
/// First value for `name`, if any.
pub fn get(&self, name: &str) -> Option<&str> {
let lower = name.to_ascii_lowercase();
self.inner.iter()
.find(|(n, _)| n == &lower)
.map(|(_, v)| v.as_str())
}
/// Iterate over (name, value) pairs in insertion order.
pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
self.inner.iter().map(|(n, v)| (n.as_str(), v.as_str()))
}
pub fn len(&self) -> usize { self.inner.len() }
pub fn is_empty(&self) -> bool { self.inner.is_empty() }
}
// ---------------------------------------------------------------------------
// Body — request body, exposed as owned bytes.
// ---------------------------------------------------------------------------
//
// v1: bodies are read in full by the connection actor before the pipeline
// runs, when `Content-Length` is set. Streaming and chunked encoding will
// come later; for now this is the simplest correct thing.
#[derive(Debug, Clone, Default)]
pub struct Body {
bytes: Vec<u8>,
}
impl Body {
pub fn empty() -> Self { Self::default() }
pub fn from_bytes(b: Vec<u8>) -> Self { Self { bytes: b } }
pub fn as_bytes(&self) -> &[u8] { &self.bytes }
pub fn into_bytes(self) -> Vec<u8> { self.bytes }
pub fn len(&self) -> usize { self.bytes.len() }
pub fn is_empty(&self) -> bool { self.bytes.is_empty() }
}
// ---------------------------------------------------------------------------
// RespBody — what plugs put on the wire.
// ---------------------------------------------------------------------------
//
// Enum, not a trait, so the connection actor can pattern-match. Leaves room
// for future variants like Sse, Stream, Chunked without touching the plug
// API.
#[derive(Debug, Clone)]
pub enum RespBody {
Empty,
Bytes(Vec<u8>),
}
impl RespBody {
pub fn len_hint(&self) -> usize {
match self {
RespBody::Empty => 0,
RespBody::Bytes(b) => b.len(),
}
}
}
impl Default for RespBody {
fn default() -> Self { RespBody::Empty }
}
// ---------------------------------------------------------------------------
// Params — path parameters extracted by the router.
// ---------------------------------------------------------------------------
//
// Small-vec semantics: routes have a handful of params; a Vec is the right
// shape. Empty until the router plug populates it.
#[derive(Debug, Clone, Default)]
pub struct Params {
inner: Vec<(String, String)>,
}
impl Params {
pub fn new() -> Self { Self::default() }
pub fn put(&mut self, name: impl Into<String>, value: impl Into<String>) {
self.inner.push((name.into(), value.into()));
}
pub fn get(&self, name: &str) -> Option<&str> {
self.inner.iter()
.find(|(n, _)| n == name)
.map(|(_, v)| v.as_str())
}
pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
self.inner.iter().map(|(n, v)| (n.as_str(), v.as_str()))
}
}
// ---------------------------------------------------------------------------
// Assigns — type-erased map for arbitrary user data.
// ---------------------------------------------------------------------------
//
// Same pattern as Phoenix's `conn.assigns`. The `Option<HashMap>` keeps the
// common case (no assigns) at zero allocation cost.
#[derive(Debug, Default)]
pub struct Assigns {
inner: Option<HashMap<TypeId, Box<dyn Any + Send>>>,
}
impl Assigns {
pub fn new() -> Self { Self::default() }
pub fn put<T: Any + Send>(&mut self, value: T) {
self.inner
.get_or_insert_with(HashMap::new)
.insert(TypeId::of::<T>(), Box::new(value));
}
pub fn get<T: Any + Send>(&self) -> Option<&T> {
self.inner
.as_ref()?
.get(&TypeId::of::<T>())?
.downcast_ref::<T>()
}
pub fn take<T: Any + Send>(&mut self) -> Option<T> {
let boxed = self.inner.as_mut()?.remove(&TypeId::of::<T>())?;
boxed.downcast::<T>().ok().map(|b| *b)
}
}
// ---------------------------------------------------------------------------
// Conn — the pipeline value.
// ---------------------------------------------------------------------------
#[derive(Debug)]
pub struct Conn {
// Request (populated before the pipeline runs).
pub method: Method,
pub path: String,
pub query: Option<String>,
pub version: HttpVersion,
pub headers: HeaderMap,
pub body: Body,
// Response (built up by plugs, written after the pipeline returns).
pub status: Option<u16>,
pub resp_headers: HeaderMap,
pub resp_body: RespBody,
// Pipeline state.
pub params: Params,
pub assigns: Assigns,
pub halted: bool,
}
impl Conn {
/// Construct an empty `Conn`. Tests use this; the connection actor uses
/// `From<RequestParts>` (see parser.rs).
pub fn new() -> Self {
Self {
method: Method::Get,
path: String::new(),
query: None,
version: HttpVersion::Http11,
headers: HeaderMap::new(),
body: Body::empty(),
status: None,
resp_headers: HeaderMap::new(),
resp_body: RespBody::Empty,
params: Params::new(),
assigns: Assigns::new(),
halted: false,
}
}
// ----- Fluent builders. Most handler code uses these. -----
pub fn put_status(mut self, status: u16) -> Self {
self.status = Some(status);
self
}
pub fn put_header(mut self, name: &str, value: impl Into<String>) -> Self {
self.resp_headers.set(name, value);
self
}
pub fn put_body(mut self, body: impl Into<RespBody>) -> Self {
self.resp_body = body.into();
self
}
pub fn put_params(mut self, params: Params) -> Self {
self.params = params;
self
}
pub fn halt(mut self) -> Self {
self.halted = true;
self
}
}
impl Default for Conn {
fn default() -> Self { Self::new() }
}
// ----- Ergonomic conversions into RespBody. -----
impl From<Vec<u8>> for RespBody {
fn from(v: Vec<u8>) -> Self { RespBody::Bytes(v) }
}
impl From<String> for RespBody {
fn from(s: String) -> Self { RespBody::Bytes(s.into_bytes()) }
}
impl From<&'static str> for RespBody {
fn from(s: &'static str) -> Self { RespBody::Bytes(s.as_bytes().to_vec()) }
}
impl From<&[u8]> for RespBody {
fn from(s: &[u8]) -> Self { RespBody::Bytes(s.to_vec()) }
}
src/conn_actor.rs
+309
View File
@@ -0,0 +1,309 @@
//! The connection actor — one per accepted TCP connection.
//!
//! Runs the HTTP/1.1 request loop:
//!
//! loop {
//! read bytes → parse → build Conn
//! pipeline.run(conn) // inline; no spawn
//! write response
//! if !keep_alive { break }
//! }
//!
//! Everything in here happens in one smarm green thread. The actor parks on
//! `wait_readable` between bytes and `wait_writable` during slow writes;
//! during those parks, other connection actors progress freely.
use crate::conn::{Body, Conn, RespBody};
use crate::net::OwnedFd;
use crate::parser::{self, ParseError};
use crate::plug::Pipeline;
use std::io::{self, ErrorKind};
use std::os::fd::RawFd;
use std::time::Duration;
// ---------------------------------------------------------------------------
// Limits
// ---------------------------------------------------------------------------
/// Per-connection settings the connection actor needs to honour.
#[derive(Clone, Copy, Debug)]
pub struct ConnLimits {
pub max_headers: usize,
pub initial_read_buf: usize,
/// Hard cap on the request head to bound buffer growth. 64 KiB is
/// well over Apache's 8 KiB default; protects against pathological
/// clients streaming headers forever.
pub max_head_bytes: usize,
/// Hard cap on Content-Length we'll accept. 16 MiB is enough for a CRUD
/// example; configurable in `Config`.
pub max_body_bytes: usize,
pub keep_alive_timeout: Duration,
}
impl Default for ConnLimits {
fn default() -> Self {
Self {
max_headers: 64,
initial_read_buf: 8 * 1024,
max_head_bytes: 64 * 1024,
max_body_bytes: 16 * 1024 * 1024,
keep_alive_timeout: Duration::from_secs(60),
}
}
}
// ---------------------------------------------------------------------------
// run_connection — entry point spawned by the listener actor.
// ---------------------------------------------------------------------------
pub fn run_connection(fd: OwnedFd, pipeline: Pipeline, limits: ConnLimits) {
// The OwnedFd cleans up via Drop on any exit path (panic, error, or
// normal close). No explicit close calls below.
let raw = fd.as_raw();
let mut buf: Vec<u8> = Vec::with_capacity(limits.initial_read_buf);
loop {
// ----- 1. Read until we have a full request head. -----
let parsed = match read_head(raw, &mut buf, &limits) {
Ok(p) => p,
Err(ReadHeadErr::ClientClosed) => {
// Clean EOF between requests (or before any request). Normal.
return;
}
Err(ReadHeadErr::Io(_)) => {
// Network error or timeout. Best-effort close; we're done.
return;
}
Err(ReadHeadErr::Parse(e)) => {
emit_error_response(raw, &e);
return;
}
};
// ----- 2. Read body. -----
let body_len = parsed.content_length.unwrap_or(0);
if body_len > limits.max_body_bytes {
let _ = write_all(raw, b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n");
return;
}
// If client sent `Expect: 100-continue`, emit it before reading the
// body. RFC 7231 §5.1.1. We don't gate on app logic here; v1 always
// accepts.
if parsed.expect_100 {
if write_all(raw, b"HTTP/1.1 100 Continue\r\n\r\n").is_err() {
return;
}
}
let body = match read_body(raw, &mut buf, parsed.head_len, body_len) {
Ok(b) => b,
Err(_) => return,
};
let keep_alive = parsed.keep_alive;
let version = parsed.version;
let head_len = parsed.head_len;
let conn = parser::build_conn(parsed, Body::from_bytes(body));
// ----- 3. Run the pipeline. -----
// Catch panics at the actor boundary — a panicking handler should
// not take down the whole connection silently with no response.
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
pipeline.run(conn)
}));
let mut response_conn = match result {
Ok(c) => c,
Err(_) => {
// Compose a 500 manually; the original Conn was moved into
// the closure.
let mut c = Conn::new();
c.version = version;
c.put_status(500).put_header("content-length", "0")
}
};
// If no plug touched status, that's a configuration error (no router
// matched, no default handler). Emit 404.
if response_conn.status.is_none() {
response_conn = response_conn.put_status(404)
.put_body(RespBody::Empty);
}
// ----- 4. Write the response. -----
let bytes = parser::serialise_response(&response_conn, keep_alive);
if write_all(raw, &bytes).is_err() {
return;
}
// ----- 5. Loop or close. -----
if !keep_alive {
return;
}
// Drop the request bytes (head + body) from `buf`; anything past
// them is the start of the next pipelined request.
let consumed = head_len + body_len;
buf.drain(..consumed);
}
}
// ---------------------------------------------------------------------------
// read_head
// ---------------------------------------------------------------------------
#[allow(dead_code)] // io::Error is captured for future logging
enum ReadHeadErr {
ClientClosed,
Io(io::Error),
Parse(ParseError),
}
/// Read until `parse_head` succeeds or fails definitively. `buf` may already
/// contain leftover bytes from a previous keep-alive cycle; we try to parse
/// those before reading more from the socket.
fn read_head(
fd: RawFd,
buf: &mut Vec<u8>,
limits: &ConnLimits,
) -> Result<parser::ParsedHead, ReadHeadErr> {
loop {
// Try to parse what we already have. On the first iteration of a
// fresh keep-alive cycle, `buf` may already hold the next request.
if !buf.is_empty() {
match parser::parse_head(buf, limits.max_headers) {
Ok(h) => return Ok(h),
Err(ParseError::Incomplete) => {} // need more bytes
Err(e) => return Err(ReadHeadErr::Parse(e)),
}
}
if buf.len() >= limits.max_head_bytes {
return Err(ReadHeadErr::Parse(ParseError::TooManyHeaders));
}
// Read more.
match read_some(fd, buf, limits.initial_read_buf) {
Ok(0) => return Err(ReadHeadErr::ClientClosed),
Ok(_) => continue,
Err(e) => return Err(ReadHeadErr::Io(e)),
}
}
}
// ---------------------------------------------------------------------------
// read_body
// ---------------------------------------------------------------------------
fn read_body(
fd: RawFd,
buf: &mut Vec<u8>,
head_len: usize,
body_len: usize,
) -> io::Result<Vec<u8>> {
// Bytes already in `buf` past the head belong to the body.
let already = buf.len().saturating_sub(head_len);
let need = body_len.saturating_sub(already);
if need == 0 {
// We have the full body in `buf` already. Extract a copy; `buf` is
// drained later in the connection loop.
return Ok(buf[head_len..head_len + body_len].to_vec());
}
// Read until we have the rest.
let mut total_read = already;
while total_read < body_len {
match read_some(fd, buf, 8 * 1024) {
Ok(0) => return Err(io::Error::new(ErrorKind::UnexpectedEof, "client closed during body")),
Ok(n) => total_read += n,
Err(e) => return Err(e),
}
}
Ok(buf[head_len..head_len + body_len].to_vec())
}
// ---------------------------------------------------------------------------
// read_some — single epoll-park + read loop.
// ---------------------------------------------------------------------------
//
// Appends what it reads onto `buf`. Returns bytes read, 0 for EOF, or the
// last io error.
fn read_some(fd: RawFd, buf: &mut Vec<u8>, chunk: usize) -> io::Result<usize> {
// Loop to absorb EAGAIN: a readable wakeup followed by EAGAIN is
// possible (signal race, etc). Re-park and retry rather than returning
// 0 (which would be confused with EOF by callers).
loop {
smarm::wait_readable(fd)?;
let start = buf.len();
buf.resize(start + chunk, 0);
let n = unsafe {
libc::read(fd, buf.as_mut_ptr().add(start) as *mut _, chunk)
};
if n < 0 {
let err = io::Error::last_os_error();
buf.truncate(start);
if err.kind() == ErrorKind::WouldBlock || err.kind() == ErrorKind::Interrupted {
continue;
}
return Err(err);
}
let n = n as usize;
buf.truncate(start + n);
return Ok(n); // n == 0 here is real EOF
}
}
// ---------------------------------------------------------------------------
// write_all — robust write loop.
// ---------------------------------------------------------------------------
fn write_all(fd: RawFd, mut buf: &[u8]) -> io::Result<()> {
while !buf.is_empty() {
// Park on writability before each syscall.
smarm::wait_writable(fd)?;
let n = unsafe {
libc::write(fd, buf.as_ptr() as *const _, buf.len())
};
if n < 0 {
let err = io::Error::last_os_error();
if err.kind() == ErrorKind::WouldBlock {
continue; // spurious wake; retry
}
return Err(err);
}
if n == 0 {
return Err(io::Error::new(ErrorKind::WriteZero, "write returned 0"));
}
buf = &buf[n as usize..];
}
Ok(())
}
// ---------------------------------------------------------------------------
// Error responses for unparseable / malformed requests.
// ---------------------------------------------------------------------------
fn emit_error_response(fd: RawFd, err: &ParseError) {
let resp: &[u8] = match err {
ParseError::TooManyHeaders =>
b"HTTP/1.1 431 Request Header Fields Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
ParseError::BadContentLength =>
b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
ParseError::Unsupported =>
b"HTTP/1.1 411 Length Required\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
// Incomplete and Malformed both lead here; Incomplete shouldn't
// appear (read_head loops on it).
_ =>
b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
};
let _ = write_all(fd, resp);
}
src/lib.rs
+33
View File
@@ -0,0 +1,33 @@
//! # urus — a cowboy/bandit-style HTTP library for the smarm actor runtime.
//!
//! v1 covers HTTP/1.1, the plug pipeline, and a built-in router. See
//! `urus-spec.md` for the design.
//!
//! ```no_run
//! use urus::{Pipeline, Router, Conn, Next, serve};
//!
//! let pipeline = Pipeline::new().plug(
//! Router::new()
//! .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello"))
//! .get("/users/:id", |c: Conn, _n: Next| {
//! let id = c.params.get("id").unwrap_or("").to_string();
//! c.put_status(200).put_body(id)
//! })
//! );
//!
//! serve("0.0.0.0:8080", pipeline).unwrap();
//! ```
pub mod conn;
pub mod plug;
pub mod router;
pub mod parser;
pub mod net;
pub mod conn_actor;
pub mod serve;
// Re-exports — what most users want at the crate root.
pub use conn::{Assigns, Body, Conn, HeaderMap, HttpVersion, Method, Params, RespBody};
pub use plug::{Next, Pipeline, Plug};
pub use router::Router;
pub use serve::{serve, serve_with, Config};
src/net.rs
+165
View File
@@ -0,0 +1,165 @@
//! Thin TCP socket layer.
//!
//! We don't use `std::net::TcpListener` because we need non-blocking accept
//! integrated with smarm's epoll loop. So this drops down to libc: socket,
//! bind, listen, accept4. All fds carry `O_NONBLOCK | O_CLOEXEC` so smarm's
//! readiness primitives work as documented.
use std::io;
use std::net::SocketAddr;
use std::os::fd::RawFd;
// ---------------------------------------------------------------------------
// OwnedFd — a tiny RAII wrapper that closes on drop.
// ---------------------------------------------------------------------------
//
// Sent through smarm channels (which require `Send`) for zero-copy fd
// handoff between listener and connection actor. `Send` is safe: fd values
// are just integers, and ownership semantics — exactly-one closer — are
// enforced by the type itself.
#[derive(Debug)]
pub struct OwnedFd {
fd: RawFd,
}
impl OwnedFd {
/// Wrap a raw fd. The wrapper now owns the fd and will close it on drop.
pub fn from_raw(fd: RawFd) -> Self {
Self { fd }
}
pub fn as_raw(&self) -> RawFd { self.fd }
/// Release ownership without closing. The caller must close the fd.
pub fn into_raw(self) -> RawFd {
let fd = self.fd;
std::mem::forget(self);
fd
}
}
impl Drop for OwnedFd {
fn drop(&mut self) {
if self.fd >= 0 {
unsafe { libc::close(self.fd); }
}
}
}
// fd handoff: SAFETY: RawFd is an integer; sending one across threads
// transfers ownership in the same way moving an i32 would. The recipient
// becomes the unique closer.
unsafe impl Send for OwnedFd {}
// ---------------------------------------------------------------------------
// bind_and_listen
// ---------------------------------------------------------------------------
const LISTEN_BACKLOG: i32 = 1024;
/// Create a non-blocking, SO_REUSEADDR TCP listener bound to `addr`. The
/// returned fd is ready for `accept4` calls; the caller registers it with
/// smarm via `wait_readable` between accepts.
pub fn bind_and_listen(addr: SocketAddr) -> io::Result<OwnedFd> {
let family = match addr {
SocketAddr::V4(_) => libc::AF_INET,
SocketAddr::V6(_) => libc::AF_INET6,
};
let fd = unsafe {
libc::socket(
family,
libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC,
0,
)
};
if fd < 0 {
return Err(io::Error::last_os_error());
}
let owned = OwnedFd::from_raw(fd);
// SO_REUSEADDR — standard for servers; avoids TIME_WAIT bind failures
// on restart.
let opt: libc::c_int = 1;
let r = unsafe {
libc::setsockopt(
fd,
libc::SOL_SOCKET,
libc::SO_REUSEADDR,
&opt as *const _ as *const libc::c_void,
std::mem::size_of_val(&opt) as libc::socklen_t,
)
};
if r < 0 {
return Err(io::Error::last_os_error());
}
// bind().
match addr {
SocketAddr::V4(a) => {
let sa = libc::sockaddr_in {
sin_family: libc::AF_INET as u16,
sin_port: a.port().to_be(),
sin_addr: libc::in_addr { s_addr: u32::from_ne_bytes(a.ip().octets()) },
sin_zero: [0; 8],
};
let r = unsafe {
libc::bind(
fd,
&sa as *const _ as *const libc::sockaddr,
std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
)
};
if r < 0 { return Err(io::Error::last_os_error()); }
}
SocketAddr::V6(a) => {
let sa = libc::sockaddr_in6 {
sin6_family: libc::AF_INET6 as u16,
sin6_port: a.port().to_be(),
sin6_flowinfo: a.flowinfo(),
sin6_addr: libc::in6_addr { s6_addr: a.ip().octets() },
sin6_scope_id: a.scope_id(),
};
let r = unsafe {
libc::bind(
fd,
&sa as *const _ as *const libc::sockaddr,
std::mem::size_of::<libc::sockaddr_in6>() as libc::socklen_t,
)
};
if r < 0 { return Err(io::Error::last_os_error()); }
}
}
// listen().
let r = unsafe { libc::listen(fd, LISTEN_BACKLOG) };
if r < 0 { return Err(io::Error::last_os_error()); }
Ok(owned)
}
// ---------------------------------------------------------------------------
// accept_nonblocking
// ---------------------------------------------------------------------------
/// One non-blocking `accept4`. Returns the new fd on success,
/// `Err(WouldBlock)` if no connection is pending (caller should park on
/// `wait_readable(listener)` and retry), or other errors directly.
pub fn accept_nonblocking(listener: RawFd) -> io::Result<OwnedFd> {
let mut addr: libc::sockaddr_storage = unsafe { std::mem::zeroed() };
let mut len: libc::socklen_t = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
let fd = unsafe {
libc::accept4(
listener,
&mut addr as *mut _ as *mut libc::sockaddr,
&mut len,
libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC,
)
};
if fd < 0 {
return Err(io::Error::last_os_error());
}
Ok(OwnedFd::from_raw(fd))
}
src/parser.rs
+379
View File
@@ -0,0 +1,379 @@
//! HTTP/1.1 wire protocol — request parsing and response serialisation.
//!
//! Parsing uses `httparse` for the request line + headers (zero-alloc, zero-
//! copy, battle-tested). Body framing, keep-alive logic and response writing
//! are ours.
//!
//! v1 body framing:
//! - `Content-Length: N` — read exactly N bytes.
//! - No body header — empty body.
//! - `Transfer-Encoding: chunked` — deferred. Returns ParseError::Unsupported
//! and the connection actor responds 411 Length Required + close.
//!
//! Keep it stupid simple. Chunked decoding lands when something actually
//! requests it.
use crate::conn::{Body, Conn, HeaderMap, HttpVersion, Method, RespBody};
// ---------------------------------------------------------------------------
// ParseError
// ---------------------------------------------------------------------------
#[derive(Debug)]
pub enum ParseError {
/// Request bytes are well-formed but incomplete; need more from the
/// socket.
Incomplete,
/// Request is malformed; respond 400 and close.
Malformed,
/// Header count exceeded the configured max; respond 431 and close.
TooManyHeaders,
/// `Content-Length` header could not be parsed as an integer.
BadContentLength,
/// A wire feature we haven't implemented yet (e.g. chunked encoding).
/// Connection actor responds 411 + close.
Unsupported,
}
// ---------------------------------------------------------------------------
// ParsedHead — what `parse_head` returns on success.
// ---------------------------------------------------------------------------
//
// `head_len` is how many bytes from the start of the buffer the head
// occupied — the body starts at `&buf[head_len..]`.
pub struct ParsedHead {
pub head_len: usize,
pub method: Method,
pub path: String,
pub query: Option<String>,
pub version: HttpVersion,
pub headers: HeaderMap,
pub content_length: Option<usize>,
pub keep_alive: bool,
pub expect_100: bool,
}
/// Try to parse a request head from `buf`. Returns `Incomplete` if more
/// bytes are needed; the caller should read more and retry with the *same*
/// buffer.
pub fn parse_head(buf: &[u8], max_headers: usize) -> Result<ParsedHead, ParseError> {
// httparse needs a header array; size it from config. 64 is the spec
// default. Stack allocation; if a request has more than max_headers
// they get TooManyHeaders.
let mut header_buf = vec![httparse::EMPTY_HEADER; max_headers];
let mut req = httparse::Request::new(&mut header_buf);
let head_len = match req.parse(buf) {
Ok(httparse::Status::Complete(n)) => n,
Ok(httparse::Status::Partial) => return Err(ParseError::Incomplete),
Err(httparse::Error::TooManyHeaders) => return Err(ParseError::TooManyHeaders),
Err(_) => return Err(ParseError::Malformed),
};
let method_str = req.method.ok_or(ParseError::Malformed)?;
let raw_path = req.path.ok_or(ParseError::Malformed)?;
let v = req.version.ok_or(ParseError::Malformed)?;
let version = match v {
0 => HttpVersion::Http10,
1 => HttpVersion::Http11,
_ => return Err(ParseError::Malformed),
};
// Split path and query. We don't percent-decode the path here — the
// spec calls for it, but for v1 raw is fine; user code can decode if
// they need to. Same with the query.
let (path, query) = match raw_path.split_once('?') {
Some((p, q)) => (p.to_string(), Some(q.to_string())),
None => (raw_path.to_string(), None),
};
// Walk headers, building HeaderMap and pulling out the few we need
// ourselves (Content-Length, Connection, Transfer-Encoding, Expect).
let mut headers = HeaderMap::with_capacity(req.headers.len());
let mut content_length = None;
let mut connection_hdr = None;
let mut chunked = false;
let mut expect_100 = false;
for h in req.headers.iter() {
let name_lower = h.name.to_ascii_lowercase();
let value = std::str::from_utf8(h.value).map_err(|_| ParseError::Malformed)?;
match name_lower.as_str() {
"content-length" => {
content_length = Some(
value.trim()
.parse::<usize>()
.map_err(|_| ParseError::BadContentLength)?
);
}
"transfer-encoding" => {
// We only care whether it includes "chunked". Multiple codings
// can appear; chunked is the only one we'd need to decode.
if value.to_ascii_lowercase().split(',').any(|t| t.trim() == "chunked") {
chunked = true;
}
}
"connection" => {
connection_hdr = Some(value.to_ascii_lowercase());
}
"expect" => {
if value.eq_ignore_ascii_case("100-continue") {
expect_100 = true;
}
}
_ => {}
}
headers.append(&name_lower, value.to_string());
}
if chunked {
return Err(ParseError::Unsupported);
}
// Keep-alive logic, RFC 7230 §6.3:
// HTTP/1.1: keep-alive by default; "Connection: close" overrides.
// HTTP/1.0: close by default; "Connection: keep-alive" overrides.
let keep_alive = match version {
HttpVersion::Http11 => connection_hdr.as_deref() != Some("close"),
HttpVersion::Http10 => connection_hdr.as_deref() == Some("keep-alive"),
};
Ok(ParsedHead {
head_len,
method: Method::parse(method_str),
path,
query,
version,
headers,
content_length,
keep_alive,
expect_100,
})
}
// ---------------------------------------------------------------------------
// Conn assembly
// ---------------------------------------------------------------------------
pub fn build_conn(head: ParsedHead, body: Body) -> Conn {
let mut c = Conn::new();
c.method = head.method;
c.path = head.path;
c.query = head.query;
c.version = head.version;
c.headers = head.headers;
c.body = body;
c
}
// ---------------------------------------------------------------------------
// Response serialisation
// ---------------------------------------------------------------------------
//
// We emit:
// <version> <status> <reason>\r\n
// <header>: <value>\r\n
// ...
// \r\n
// <body bytes>
//
// Headers we always inject (unless the user already set them):
// - Content-Length: from resp_body.len_hint()
// - Connection: close (if keep-alive is off this request)
// - Date: skipped in v1; not required by the standard and adds complexity.
pub fn serialise_response(conn: &Conn, keep_alive: bool) -> Vec<u8> {
// Pre-size: status line ~30 + headers ~50/each + body. Good enough.
let body_len = conn.resp_body.len_hint();
let mut out = Vec::with_capacity(64 + conn.resp_headers.len() * 40 + body_len);
let status = conn.status.unwrap_or(200);
let reason = reason_phrase(status);
// Status line.
out.extend_from_slice(conn.version.as_str().as_bytes());
out.push(b' ');
out.extend_from_slice(status.to_string().as_bytes());
out.push(b' ');
out.extend_from_slice(reason.as_bytes());
out.extend_from_slice(b"\r\n");
// User headers — written first so subsequent injection can skip them.
let mut wrote_content_length = false;
let mut wrote_connection = false;
for (name, value) in conn.resp_headers.iter() {
match name {
"content-length" => wrote_content_length = true,
"connection" => wrote_connection = true,
_ => {}
}
out.extend_from_slice(name.as_bytes());
out.extend_from_slice(b": ");
out.extend_from_slice(value.as_bytes());
out.extend_from_slice(b"\r\n");
}
if !wrote_content_length {
out.extend_from_slice(b"content-length: ");
out.extend_from_slice(body_len.to_string().as_bytes());
out.extend_from_slice(b"\r\n");
}
if !wrote_connection && !keep_alive {
out.extend_from_slice(b"connection: close\r\n");
}
out.extend_from_slice(b"\r\n");
// Body.
match &conn.resp_body {
RespBody::Empty => {}
RespBody::Bytes(b) => out.extend_from_slice(b),
}
out
}
/// HTTP/1.1 reason phrases. Required by RFC 7230 §3.1.2 — clients may
/// display them.
pub fn reason_phrase(status: u16) -> &'static str {
match status {
200 => "OK",
201 => "Created",
202 => "Accepted",
204 => "No Content",
301 => "Moved Permanently",
302 => "Found",
303 => "See Other",
304 => "Not Modified",
307 => "Temporary Redirect",
308 => "Permanent Redirect",
400 => "Bad Request",
401 => "Unauthorized",
403 => "Forbidden",
404 => "Not Found",
405 => "Method Not Allowed",
409 => "Conflict",
411 => "Length Required",
413 => "Payload Too Large",
414 => "URI Too Long",
415 => "Unsupported Media Type",
422 => "Unprocessable Entity",
429 => "Too Many Requests",
431 => "Request Header Fields Too Large",
500 => "Internal Server Error",
501 => "Not Implemented",
502 => "Bad Gateway",
503 => "Service Unavailable",
_ => "OK", // best-effort fallback; spec allows any phrase
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_simple_get() {
let req = b"GET /hello HTTP/1.1\r\nHost: x\r\n\r\n";
let head = parse_head(req, 64).unwrap();
assert_eq!(head.method, Method::Get);
assert_eq!(head.path, "/hello");
assert_eq!(head.version, HttpVersion::Http11);
assert!(head.keep_alive);
assert_eq!(head.content_length, None);
}
#[test]
fn parse_with_query() {
let req = b"GET /users?id=42&active=true HTTP/1.1\r\nHost: x\r\n\r\n";
let head = parse_head(req, 64).unwrap();
assert_eq!(head.path, "/users");
assert_eq!(head.query.as_deref(), Some("id=42&active=true"));
}
#[test]
fn parse_post_with_content_length() {
let req = b"POST /a HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\nhello";
let head = parse_head(req, 64).unwrap();
assert_eq!(head.content_length, Some(5));
// head_len is bytes before the body; the remainder is the body.
assert_eq!(&req[head.head_len..], b"hello");
}
#[test]
fn parse_connection_close() {
let req = b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n";
let head = parse_head(req, 64).unwrap();
assert!(!head.keep_alive);
}
#[test]
fn parse_http10_default_close() {
let req = b"GET / HTTP/1.0\r\nHost: x\r\n\r\n";
let head = parse_head(req, 64).unwrap();
assert!(!head.keep_alive);
}
#[test]
fn parse_http10_keepalive_opt_in() {
let req = b"GET / HTTP/1.0\r\nHost: x\r\nConnection: keep-alive\r\n\r\n";
let head = parse_head(req, 64).unwrap();
assert!(head.keep_alive);
}
#[test]
fn parse_partial() {
let req = b"GET /hello HTTP/1.1\r\nHost: x\r\n"; // no terminator yet
match parse_head(req, 64) {
Err(ParseError::Incomplete) => {}
_ => panic!("expected Incomplete"),
}
}
#[test]
fn parse_chunked_unsupported() {
let req = b"POST /a HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n";
match parse_head(req, 64) {
Err(ParseError::Unsupported) => {}
_ => panic!("expected Unsupported for chunked"),
}
}
#[test]
fn serialise_basic_200() {
let conn = Conn::new().put_status(200).put_body("hi");
let bytes = serialise_response(&conn, true);
let s = std::str::from_utf8(&bytes).unwrap();
assert!(s.starts_with("HTTP/1.1 200 OK\r\n"));
assert!(s.contains("content-length: 2"));
assert!(s.ends_with("\r\n\r\nhi"));
}
#[test]
fn serialise_connection_close_when_not_keepalive() {
let conn = Conn::new().put_status(204);
let bytes = serialise_response(&conn, false);
let s = std::str::from_utf8(&bytes).unwrap();
assert!(s.contains("connection: close"));
}
#[test]
fn serialise_user_content_length_is_respected() {
let conn = Conn::new().put_status(200)
.put_header("content-length", "999")
.put_body("hi"); // mismatch on purpose — user wins
let bytes = serialise_response(&conn, true);
let s = std::str::from_utf8(&bytes).unwrap();
assert!(s.contains("content-length: 999"));
assert!(!s.contains("content-length: 2"));
}
}
src/plug.rs
+100
View File
@@ -0,0 +1,100 @@
//! The plug pipeline — universal abstraction for everything in urus.
//!
//! A `Plug` is anything that takes a `Conn`, optionally calls `next.run(conn)`
//! to delegate to the rest of the pipeline, and returns a `Conn`. Routing,
//! middleware, handlers — all the same shape.
//!
//! The halt is structural: not calling `next.run(conn)` *is* the halt. Setting
//! `conn.halted = true` is only a hint to outer plugs that the pipeline was
//! short-circuited (e.g. a logger that wraps `next`).
use crate::conn::Conn;
use std::sync::Arc;
// ---------------------------------------------------------------------------
// Plug trait
// ---------------------------------------------------------------------------
pub trait Plug: Send + Sync + 'static {
fn call(&self, conn: Conn, next: Next) -> Conn;
}
// Blanket impl for closures. Both `Send + Sync` so plug values can be shared
// across connection actors via the Arc'd pipeline.
impl<F> Plug for F
where
F: Fn(Conn, Next) -> Conn + Send + Sync + 'static,
{
fn call(&self, conn: Conn, next: Next) -> Conn {
self(conn, next)
}
}
// ---------------------------------------------------------------------------
// Next — a handle to the remainder of the pipeline.
// ---------------------------------------------------------------------------
//
// Opaque wrapper around a slice of the remaining plugs. `run(conn)` consumes
// the handle (no double-calling) and dispatches to the head plug. Empty
// remainder is a no-op — the conn passes through unchanged.
pub struct Next<'a> {
plugs: &'a [Arc<dyn Plug>],
}
impl<'a> Next<'a> {
pub(crate) fn new(plugs: &'a [Arc<dyn Plug>]) -> Self {
Self { plugs }
}
/// Continue down the pipeline. Returns the `Conn` produced by the rest.
pub fn run(self, conn: Conn) -> Conn {
match self.plugs.split_first() {
None => conn,
Some((head, tail)) => head.call(conn, Next { plugs: tail }),
}
}
}
// ---------------------------------------------------------------------------
// Pipeline
// ---------------------------------------------------------------------------
//
// Built once at startup, cloned cheaply (Arc bump) into each connection
// actor. Read-only after construction; no per-request allocation in here.
#[derive(Clone)]
pub struct Pipeline {
plugs: Arc<Vec<Arc<dyn Plug>>>,
}
impl Pipeline {
pub fn new() -> Self {
Self { plugs: Arc::new(Vec::new()) }
}
/// Append a plug. Builder pattern — consumes and returns self.
///
/// Mutates the underlying Vec via `Arc::make_mut` when possible (it's
/// unique during construction), which is allocation-free after the first
/// `plug()` until clones are taken.
pub fn plug(mut self, p: impl Plug) -> Self {
let v = Arc::make_mut(&mut self.plugs);
v.push(Arc::new(p));
self
}
/// Run the pipeline against a `Conn`. The connection actor calls this
/// once per request.
pub fn run(&self, conn: Conn) -> Conn {
Next::new(self.plugs.as_slice()).run(conn)
}
/// Number of plugs in the pipeline. Mostly for tests.
pub fn len(&self) -> usize { self.plugs.len() }
pub fn is_empty(&self) -> bool { self.plugs.is_empty() }
}
impl Default for Pipeline {
fn default() -> Self { Self::new() }
}
src/router.rs
+223
View File
@@ -0,0 +1,223 @@
//! The router — itself a plug.
//!
//! v1: linear scan over a Vec of compiled route patterns. The spec calls for
//! a radix trie eventually; for "good enough to build a CRUD app" with
//! typical route counts (<50), the linear scan is dominant noise next to
//! everything else on the request path. Replace with a trie when there's
//! evidence it matters.
//!
//! Pattern syntax:
//! - `/users` — literal
//! - `/users/:id` — `id` param
//! - `/users/:id/posts/:n` — multiple params
//!
//! No wildcards in v1. They are not needed for CRUD.
use crate::conn::{Conn, Method, Params};
use crate::plug::{Next, Plug};
use std::sync::Arc;
// ---------------------------------------------------------------------------
// Pattern
// ---------------------------------------------------------------------------
/// A compiled URL pattern: a list of segments, each either a literal or a
/// named parameter. Segment compilation happens at `Router::get(...)` time;
/// matching at request time is a single-pass loop.
#[derive(Debug, Clone)]
struct Pattern {
segments: Vec<Segment>,
}
#[derive(Debug, Clone)]
enum Segment {
Literal(String),
Param(String),
}
impl Pattern {
fn compile(path: &str) -> Self {
// Treat empty path the same as "/" — single empty segment.
let trimmed = path.trim_start_matches('/');
if trimmed.is_empty() {
return Pattern { segments: Vec::new() };
}
let segments = trimmed
.split('/')
.map(|s| {
if let Some(name) = s.strip_prefix(':') {
Segment::Param(name.to_string())
} else {
Segment::Literal(s.to_string())
}
})
.collect();
Pattern { segments }
}
/// Match `path` against this pattern. Returns extracted params on
/// success, `None` on miss. Strict path equality — trailing slashes
/// matter; that's standard HTTP behaviour and we don't paper over it.
fn match_path(&self, path: &str) -> Option<Params> {
let trimmed = path.trim_start_matches('/');
if trimmed.is_empty() {
return if self.segments.is_empty() { Some(Params::new()) } else { None };
}
let parts: Vec<&str> = trimmed.split('/').collect();
if parts.len() != self.segments.len() {
return None;
}
let mut params = Params::new();
for (seg, part) in self.segments.iter().zip(parts.iter()) {
match seg {
Segment::Literal(lit) => {
if lit != part {
return None;
}
}
Segment::Param(name) => {
params.put(name.clone(), (*part).to_string());
}
}
}
Some(params)
}
}
// ---------------------------------------------------------------------------
// Route
// ---------------------------------------------------------------------------
struct Route {
method: Method,
pattern: Pattern,
handler: Arc<dyn Plug>,
}
// ---------------------------------------------------------------------------
// Router
// ---------------------------------------------------------------------------
pub struct Router {
routes: Vec<Route>,
}
impl Router {
pub fn new() -> Self {
Self { routes: Vec::new() }
}
fn add(mut self, method: Method, path: &str, plug: impl Plug) -> Self {
self.routes.push(Route {
method,
pattern: Pattern::compile(path),
handler: Arc::new(plug),
});
self
}
pub fn get(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Get, path, plug) }
pub fn post(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Post, path, plug) }
pub fn put(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Put, path, plug) }
pub fn delete(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Delete, path, plug) }
pub fn patch(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Patch, path, plug) }
pub fn head(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Head, path, plug) }
pub fn options(self, path: &str, plug: impl Plug) -> Self { self.add(Method::Options, path, plug) }
}
impl Default for Router {
fn default() -> Self { Self::new() }
}
impl Plug for Router {
fn call(&self, conn: Conn, next: Next) -> Conn {
// Two-pass dispatch: first match on method+path; if no method-and-
// path matches but a same-path-different-method does, return 405.
// Otherwise pass through to `next` so outer pipelines can layer a
// 404 handler (or skip and let the connection actor emit a default).
let mut path_seen = false;
for route in &self.routes {
if let Some(params) = route.pattern.match_path(&conn.path) {
if route.method == conn.method {
let conn = conn.put_params(params);
return route.handler.call(conn, next);
}
path_seen = true;
}
}
if path_seen {
// Path is known, method isn't — RFC 7231 §6.5.5.
conn.put_status(405)
.put_header("content-length", "0")
.halt()
} else {
// No route matched at all; let outer plugs / default handler
// deal with it.
next.run(conn)
}
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
use crate::plug::Pipeline;
fn ok_handler(_label: &'static str) -> impl Plug {
move |conn: Conn, _next: Next| conn.put_status(200).put_body(_label)
}
#[test]
fn literal_match() {
let r = Router::new().get("/ping", ok_handler("pong"));
let mut c = Conn::new();
c.method = Method::Get;
c.path = "/ping".into();
let out = Pipeline::new().plug(r).run(c);
assert_eq!(out.status, Some(200));
}
#[test]
fn param_match() {
let r = Router::new().get("/users/:id", |conn: Conn, _n: Next| {
let id = conn.params.get("id").unwrap_or("").to_string();
conn.put_status(200).put_body(id)
});
let mut c = Conn::new();
c.method = Method::Get;
c.path = "/users/42".into();
let out = Pipeline::new().plug(r).run(c);
assert_eq!(out.status, Some(200));
match &out.resp_body {
crate::conn::RespBody::Bytes(b) => assert_eq!(b, b"42"),
_ => panic!("expected bytes"),
}
}
#[test]
fn method_mismatch_yields_405() {
let r = Router::new().get("/users", ok_handler("list"));
let mut c = Conn::new();
c.method = Method::Post;
c.path = "/users".into();
let out = Pipeline::new().plug(r).run(c);
assert_eq!(out.status, Some(405));
}
#[test]
fn no_route_falls_through() {
// Router returns conn unchanged; next plug (a 404 default) handles it.
let r = Router::new().get("/users", ok_handler("list"));
let default_404 = |conn: Conn, _n: Next|
conn.put_status(404).put_body("not found");
let mut c = Conn::new();
c.method = Method::Get;
c.path = "/nowhere".into();
let out = Pipeline::new().plug(r).plug(default_404).run(c);
assert_eq!(out.status, Some(404));
}
}
src/serve.rs
+188
View File
@@ -0,0 +1,188 @@
//! Listener pool and the `serve` entry point.
//!
//! A small fixed pool of listener actors share the same TCP listen fd (via
//! `dup`) — each blocks in non-blocking `accept4` + `wait_readable` on its
//! own copy. When a connection arrives the listener spawns a connection
//! actor with the `OwnedFd` and immediately returns to `accept`. No
//! coordination needed; the kernel serialises `accept` calls across the fds.
//!
//! Sharing via `dup` rather than the same fd is deliberate — Linux's
//! `accept4` is thread-safe on a single fd, but dup'ing per-listener keeps
//! each actor's epoll registration local to its own RawFd value (so smarm's
//! `waiters: HashMap<RawFd, Pid>` doesn't see collisions between listeners
//! waiting on "the same fd").
use crate::conn_actor::{run_connection, ConnLimits};
use crate::net::{accept_nonblocking, bind_and_listen, OwnedFd};
use crate::plug::Pipeline;
use std::io::{self, ErrorKind};
use std::net::{SocketAddr, ToSocketAddrs};
use std::os::fd::RawFd;
use std::time::Duration;
// ---------------------------------------------------------------------------
// Config
// ---------------------------------------------------------------------------
#[derive(Clone, Debug)]
pub struct Config {
pub addr: SocketAddr,
pub listener_pool: usize,
pub keep_alive_timeout: Duration,
pub max_header_count: usize,
pub read_buf_size: usize,
pub request_timeout: Duration,
pub max_body_bytes: usize,
/// Number of smarm scheduler OS threads. `None` means smarm's default
/// (one per CPU). Set this to a small fixed number in tests so multiple
/// concurrent test servers don't oversubscribe the host.
pub scheduler_threads: Option<usize>,
}
impl Config {
pub fn new(addr: SocketAddr) -> Self {
let pool = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(2)
.max(2);
Self {
addr,
listener_pool: pool,
keep_alive_timeout: Duration::from_secs(60),
max_header_count: 64,
read_buf_size: 8 * 1024,
request_timeout: Duration::from_secs(30),
max_body_bytes: 16 * 1024 * 1024,
scheduler_threads: None,
}
}
fn to_conn_limits(&self) -> ConnLimits {
ConnLimits {
max_headers: self.max_header_count,
initial_read_buf: self.read_buf_size,
max_head_bytes: 64 * 1024,
max_body_bytes: self.max_body_bytes,
keep_alive_timeout: self.keep_alive_timeout,
}
}
}
// ---------------------------------------------------------------------------
// dup helper
// ---------------------------------------------------------------------------
fn dup_fd(fd: RawFd) -> io::Result<OwnedFd> {
let new_fd = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) };
if new_fd < 0 {
return Err(io::Error::last_os_error());
}
Ok(OwnedFd::from_raw(new_fd))
}
// ---------------------------------------------------------------------------
// listener actor body
// ---------------------------------------------------------------------------
fn listener_loop(listener: OwnedFd, pipeline: Pipeline, limits: ConnLimits) {
let fd = listener.as_raw();
loop {
match accept_nonblocking(fd) {
Ok(client) => {
// Hand the fd off to a new connection actor. spawn() is
// cheap on smarm — it's a single Vec push under the
// shared lock.
let p = pipeline.clone();
let l = limits;
smarm::spawn(move || run_connection(client, p, l));
}
Err(e) if e.kind() == ErrorKind::WouldBlock => {
// No pending connection. Park until the listener is
// readable again, then retry.
if let Err(we) = smarm::wait_readable(fd) {
// epoll registration failed — fatal for this listener.
eprintln!("urus: listener wait_readable failed: {we}");
return;
}
}
Err(e) if e.kind() == ErrorKind::Interrupted => {
continue;
}
Err(e) => {
// EMFILE / ENFILE / ECONNABORTED etc. Log and continue;
// the system may recover.
eprintln!("urus: accept error: {e}");
// Small backoff via smarm's sleep to avoid spinning if
// the error is sticky.
smarm::sleep(Duration::from_millis(10));
}
}
}
// listener OwnedFd drops here, closing the dup'd fd.
}
// ---------------------------------------------------------------------------
// serve_with — main entry. Boots smarm, spawns listeners, blocks.
// ---------------------------------------------------------------------------
//
// Boots an smarm runtime (one OS thread per CPU by default — see smarm's
// `Config::default()`) and runs until externally killed. We don't wire a
// graceful-shutdown signal in v1; the runtime exits when all actors exit,
// which they don't (listener loops are infinite). Ctrl-C is your friend.
pub fn serve_with(config: Config, pipeline: Pipeline) -> io::Result<()> {
let listener = bind_and_listen(config.addr)?;
println!("urus: listening on {}", config.addr);
// We want one connection-actor-spawning loop per listener pool slot.
// Each gets its own dup'd fd so epoll registrations don't collide.
let mut listener_fds = Vec::with_capacity(config.listener_pool);
listener_fds.push(listener); // primary keeps the original
for _ in 1..config.listener_pool {
let dup = dup_fd(listener_fds[0].as_raw())?;
listener_fds.push(dup);
}
let limits = config.to_conn_limits();
// smarm's runtime API: init(Config) then run(f). The closure is the
// root actor; from there we spawn one listener per fd in the pool.
let smarm_cfg = match config.scheduler_threads {
Some(n) => smarm::Config::exact(n),
None => smarm::Config::default(),
};
let rt = smarm::init(smarm_cfg);
rt.run(move || {
let n = listener_fds.len();
let mut handles = Vec::with_capacity(n);
for (i, lfd) in listener_fds.into_iter().enumerate() {
let p = pipeline.clone();
let h = smarm::spawn(move || {
println!("urus: listener {} starting", i);
listener_loop(lfd, p, limits);
});
handles.push(h);
}
// Block forever (until ctrl-C) by joining the listeners. They
// never exit on their own in v1.
for h in handles {
let _ = h.join();
}
});
Ok(())
}
// ---------------------------------------------------------------------------
// serve — convenience over serve_with.
// ---------------------------------------------------------------------------
pub fn serve(addr: impl ToSocketAddrs, pipeline: Pipeline) -> io::Result<()> {
let addr = addr
.to_socket_addrs()?
.next()
.ok_or_else(|| io::Error::new(ErrorKind::InvalidInput, "no addresses resolved"))?;
serve_with(Config::new(addr), pipeline)
}
tests/integration.rs
+227
View File
@@ -0,0 +1,227 @@
//! End-to-end integration tests.
//!
//! Each test spins up the server in a background OS thread on an
//! ephemeral port, opens a regular `std::net::TcpStream` against it, and
//! exercises the wire. Single thread for the smarm runtime per test so
//! teardown happens cleanly when the test thread is dropped (the process
//! exits at the end of the test binary).
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::time::Duration;
use urus::{serve_with, Conn, Config, Next, Pipeline, Router};
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
/// Reserve a free localhost port by binding briefly and reading back the
/// assigned port number. The port is released the instant we drop the
/// listener; there's a tiny race window before the urus server claims it
/// but it's well below the threshold for flakiness in practice.
fn free_port() -> u16 {
let l = TcpListener::bind("127.0.0.1:0").unwrap();
l.local_addr().unwrap().port()
}
fn spawn_server(pipeline: Pipeline) -> u16 {
let port = free_port();
let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
std::thread::spawn(move || {
let cfg = Config {
listener_pool: 2,
scheduler_threads: Some(2),
..Config::new(addr)
};
serve_with(cfg, pipeline).unwrap();
});
// Wait for the server to actually be listening.
for _ in 0..50 {
if TcpStream::connect(addr).is_ok() {
return port;
}
std::thread::sleep(Duration::from_millis(50));
}
panic!("server didn't come up on {addr}");
}
fn send_request(port: u16, req: &[u8]) -> Vec<u8> {
let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
s.write_all(req).unwrap();
s.shutdown(std::net::Shutdown::Write).ok();
let mut buf = Vec::new();
s.read_to_end(&mut buf).unwrap();
buf
}
fn http_status(resp: &[u8]) -> u16 {
let s = std::str::from_utf8(&resp[..resp.len().min(64)]).unwrap();
let parts: Vec<&str> = s.splitn(3, ' ').collect();
parts[1].parse().unwrap()
}
fn http_body(resp: &[u8]) -> &[u8] {
// Split on the first \r\n\r\n.
for i in 0..resp.len().saturating_sub(3) {
if &resp[i..i + 4] == b"\r\n\r\n" {
return &resp[i + 4..];
}
}
&[]
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[test]
fn hello_world() {
let pipe = Pipeline::new().plug(
Router::new().get("/", |c: Conn, _n: Next| {
c.put_status(200).put_body("hello urus")
})
);
let port = spawn_server(pipe);
let resp = send_request(port, b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
assert_eq!(http_status(&resp), 200);
assert_eq!(http_body(&resp), b"hello urus");
}
#[test]
fn echo_body() {
let pipe = Pipeline::new().plug(
Router::new().post("/echo", |c: Conn, _n: Next| {
let body = c.body.as_bytes().to_vec();
c.put_status(200).put_body(body)
})
);
let port = spawn_server(pipe);
let resp = send_request(
port,
b"POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello",
);
assert_eq!(http_status(&resp), 200);
assert_eq!(http_body(&resp), b"hello");
}
#[test]
fn path_params() {
let pipe = Pipeline::new().plug(
Router::new().get("/users/:id", |c: Conn, _n: Next| {
let id = c.params.get("id").unwrap_or("?").to_string();
c.put_status(200).put_body(format!("user={id}"))
})
);
let port = spawn_server(pipe);
let resp = send_request(port, b"GET /users/42 HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
assert_eq!(http_status(&resp), 200);
assert_eq!(http_body(&resp), b"user=42");
}
#[test]
fn method_not_allowed() {
let pipe = Pipeline::new().plug(
Router::new().get("/only-get", |c: Conn, _n: Next| c.put_status(200))
);
let port = spawn_server(pipe);
let resp = send_request(port, b"POST /only-get HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
assert_eq!(http_status(&resp), 405);
}
#[test]
fn unknown_route_falls_to_404() {
// Router falls through; connection actor's default emits 404.
let pipe = Pipeline::new().plug(
Router::new().get("/known", |c: Conn, _n: Next| c.put_status(200))
);
let port = spawn_server(pipe);
let resp = send_request(port, b"GET /missing HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
assert_eq!(http_status(&resp), 404);
}
#[test]
fn keep_alive_two_requests_on_one_connection() {
let pipe = Pipeline::new().plug(
Router::new().get("/", |c: Conn, _n: Next| c.put_status(200).put_body("ok"))
);
let port = spawn_server(pipe);
let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
// First request, no Connection header (HTTP/1.1 default = keep-alive).
s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
// Read the first response (Content-Length: 2 -> body "ok").
let mut buf = vec![0u8; 1024];
let mut total = 0;
let mut first_end = None;
while first_end.is_none() {
let n = s.read(&mut buf[total..]).unwrap();
assert!(n > 0, "server hung up early");
total += n;
// First response ends at header-terminator + 2 bytes of body.
for i in 0..total.saturating_sub(3) {
if &buf[i..i + 4] == b"\r\n\r\n" {
let body_start = i + 4;
if total >= body_start + 2 {
first_end = Some(body_start + 2);
}
break;
}
}
}
let first = &buf[..first_end.unwrap()];
assert_eq!(http_status(first), 200);
assert_eq!(http_body(first), b"ok");
// Second request on the same socket — proves keep-alive works.
s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").unwrap();
let mut rest = Vec::new();
s.read_to_end(&mut rest).unwrap();
assert_eq!(http_status(&rest), 200);
assert_eq!(http_body(&rest), b"ok");
}
#[test]
fn panicking_handler_yields_500() {
let pipe = Pipeline::new().plug(
Router::new().get("/boom", |_c: Conn, _n: Next| -> Conn {
panic!("handler explosion")
})
);
let port = spawn_server(pipe);
let resp = send_request(port, b"GET /boom HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
assert_eq!(http_status(&resp), 500);
}
#[test]
fn middleware_can_short_circuit() {
// Middleware that requires `X-Auth: secret`; otherwise returns 401
// without calling `next.run(conn)`.
let auth = |c: Conn, n: Next| {
if c.headers.get("x-auth").map(|s| s == "secret").unwrap_or(false) {
n.run(c)
} else {
c.put_status(401).halt()
}
};
let pipe = Pipeline::new()
.plug(auth)
.plug(Router::new().get("/secret", |c: Conn, _n: Next| c.put_status(200).put_body("ok")));
let port = spawn_server(pipe);
// Without auth → 401.
let resp = send_request(port, b"GET /secret HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
assert_eq!(http_status(&resp), 401);
// With auth → 200 + body.
let resp = send_request(
port,
b"GET /secret HTTP/1.1\r\nHost: x\r\nX-Auth: secret\r\nConnection: close\r\n\r\n",
);
assert_eq!(http_status(&resp), 200);
assert_eq!(http_body(&resp), b"ok");
}