add http runtime
This commit is contained in:
parent
2525eaf39b
commit
e06db83934
12 changed files with 1188 additions and 579 deletions
402
Cargo.lock
generated
402
Cargo.lock
generated
|
|
@ -8,90 +8,12 @@ version = "1.0.102"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
||||
|
||||
[[package]]
|
||||
name = "async-trait"
|
||||
version = "0.1.89"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atomic-waker"
|
||||
version = "1.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
|
||||
|
||||
[[package]]
|
||||
name = "axum"
|
||||
version = "0.7.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum-core",
|
||||
"bytes",
|
||||
"futures-util",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"itoa",
|
||||
"matchit",
|
||||
"memchr",
|
||||
"mime",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"rustversion",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"serde_path_to_error",
|
||||
"serde_urlencoded",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tower 0.5.3",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "axum-core"
|
||||
version = "0.4.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"futures-util",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"mime",
|
||||
"pin-project-lite",
|
||||
"rustversion",
|
||||
"sync_wrapper",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bitflags"
|
||||
version = "2.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af"
|
||||
|
||||
[[package]]
|
||||
name = "bytes"
|
||||
version = "1.11.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
|
||||
|
||||
[[package]]
|
||||
name = "cfg-if"
|
||||
version = "1.0.4"
|
||||
|
|
@ -126,48 +48,6 @@ version = "0.1.5"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
|
||||
|
||||
[[package]]
|
||||
name = "form_urlencoded"
|
||||
version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf"
|
||||
dependencies = [
|
||||
"percent-encoding",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-channel"
|
||||
version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-core"
|
||||
version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
|
||||
|
||||
[[package]]
|
||||
name = "futures-task"
|
||||
version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
|
||||
|
||||
[[package]]
|
||||
name = "futures-util"
|
||||
version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-task",
|
||||
"pin-project-lite",
|
||||
"slab",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "getrandom"
|
||||
version = "0.4.2"
|
||||
|
|
@ -202,86 +82,6 @@ version = "0.5.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
|
||||
|
||||
[[package]]
|
||||
name = "http"
|
||||
version = "1.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"itoa",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "http-body"
|
||||
version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"http",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "http-body-util"
|
||||
version = "0.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"http",
|
||||
"http-body",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "httparse"
|
||||
version = "1.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
|
||||
|
||||
[[package]]
|
||||
name = "httpdate"
|
||||
version = "1.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
|
||||
|
||||
[[package]]
|
||||
name = "hyper"
|
||||
version = "1.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"http",
|
||||
"http-body",
|
||||
"httparse",
|
||||
"httpdate",
|
||||
"itoa",
|
||||
"pin-project-lite",
|
||||
"smallvec",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-util"
|
||||
version = "0.1.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"http",
|
||||
"http-body",
|
||||
"hyper",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "id-arena"
|
||||
version = "2.3.0"
|
||||
|
|
@ -330,53 +130,18 @@ version = "0.4.29"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
|
||||
|
||||
[[package]]
|
||||
name = "matchit"
|
||||
version = "0.7.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94"
|
||||
|
||||
[[package]]
|
||||
name = "memchr"
|
||||
version = "2.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
|
||||
|
||||
[[package]]
|
||||
name = "mime"
|
||||
version = "0.3.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "1.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"wasi",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "once_cell"
|
||||
version = "1.21.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
|
||||
|
||||
[[package]]
|
||||
name = "percent-encoding"
|
||||
version = "2.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-lite"
|
||||
version = "0.2.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
|
||||
|
||||
[[package]]
|
||||
name = "prettyplease"
|
||||
version = "0.2.37"
|
||||
|
|
@ -416,13 +181,10 @@ name = "rt"
|
|||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
"libc",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
"tower 0.4.13",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -438,18 +200,6 @@ dependencies = [
|
|||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustversion"
|
||||
version = "1.0.22"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
|
||||
|
||||
[[package]]
|
||||
name = "ryu"
|
||||
version = "1.0.23"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
|
||||
|
||||
[[package]]
|
||||
name = "semver"
|
||||
version = "1.0.27"
|
||||
|
|
@ -499,61 +249,6 @@ dependencies = [
|
|||
"zmij",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_path_to_error"
|
||||
version = "0.1.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
|
||||
dependencies = [
|
||||
"itoa",
|
||||
"serde",
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_urlencoded"
|
||||
version = "0.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
|
||||
dependencies = [
|
||||
"form_urlencoded",
|
||||
"itoa",
|
||||
"ryu",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "signal-hook-registry"
|
||||
version = "1.4.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b"
|
||||
dependencies = [
|
||||
"errno",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "slab"
|
||||
version = "0.4.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
|
||||
|
||||
[[package]]
|
||||
name = "smallvec"
|
||||
version = "1.15.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.6.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.117"
|
||||
|
|
@ -565,12 +260,6 @@ dependencies = [
|
|||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sync_wrapper"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263"
|
||||
|
||||
[[package]]
|
||||
name = "tempfile"
|
||||
version = "3.27.0"
|
||||
|
|
@ -584,91 +273,6 @@ dependencies = [
|
|||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio"
|
||||
version = "1.52.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"mio",
|
||||
"pin-project-lite",
|
||||
"signal-hook-registry",
|
||||
"socket2",
|
||||
"tokio-macros",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-macros"
|
||||
version = "2.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower"
|
||||
version = "0.4.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
|
||||
dependencies = [
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower"
|
||||
version = "0.5.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-util",
|
||||
"pin-project-lite",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower-layer"
|
||||
version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e"
|
||||
|
||||
[[package]]
|
||||
name = "tower-service"
|
||||
version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
|
||||
|
||||
[[package]]
|
||||
name = "tracing"
|
||||
version = "0.1.44"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
|
||||
dependencies = [
|
||||
"log",
|
||||
"pin-project-lite",
|
||||
"tracing-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-core"
|
||||
version = "0.1.36"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
|
||||
dependencies = [
|
||||
"once_cell",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "unicode-ident"
|
||||
version = "1.0.24"
|
||||
|
|
@ -681,12 +285,6 @@ version = "0.2.6"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
|
||||
|
||||
[[package]]
|
||||
name = "wasi"
|
||||
version = "0.11.1+wasi-snapshot-preview1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
|
||||
|
||||
[[package]]
|
||||
name = "wasip2"
|
||||
version = "1.0.2+wasi-0.2.9"
|
||||
|
|
|
|||
|
|
@ -16,9 +16,6 @@ path = "src/bin/wo.rs"
|
|||
anyhow = "1"
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
tokio = { version = "1", features = ["rt", "macros", "net", "signal", "sync", "time"] }
|
||||
axum = "0.7"
|
||||
tower = "0.4"
|
||||
libc = "0.2" # phase 02 — direct epoll/eventfd/timerfd/signalfd syscalls
|
||||
|
||||
[dev-dependencies]
|
||||
|
|
|
|||
|
|
@ -1,15 +1,23 @@
|
|||
//! `wo` — the writeonce toolchain binary.
|
||||
//!
|
||||
//! Stage 2 scope:
|
||||
//! wo run <dir> — discover `.wo` files under <dir>, parse the type DSL,
|
||||
//! compile a catalog, and serve REST CRUD on :8080.
|
||||
//! wo --help — print usage.
|
||||
//! wo run <dir> discover `.wo` files under <dir>, parse the type DSL,
|
||||
//! compile a catalog, and serve REST CRUD on :8080.
|
||||
//! wo --help print usage.
|
||||
//!
|
||||
//! After the phase-04 cutover this binary owns one event loop on one
|
||||
//! thread. No tokio. The same pattern Redis and TigerBeetle use — see
|
||||
//! `docs/runtime/database/02-wo-language.md § Concurrency Model`.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
use std::path::PathBuf;
|
||||
use std::process::ExitCode;
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::sync::Mutex;
|
||||
use rt::http::{Connection, Listener, Router};
|
||||
use rt::runtime::{EventLoop, Interest, SignalFd, Token};
|
||||
|
||||
fn usage() {
|
||||
eprintln!(
|
||||
|
|
@ -30,38 +38,22 @@ fn main() -> ExitCode {
|
|||
let args: Vec<String> = std::env::args().skip(1).collect();
|
||||
let slice: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
|
||||
match slice.as_slice() {
|
||||
[] | ["--help"] | ["-h"] => {
|
||||
usage();
|
||||
ExitCode::from(0)
|
||||
}
|
||||
[] | ["--help"] | ["-h"] => { usage(); ExitCode::from(0) }
|
||||
["run"] => {
|
||||
eprintln!("wo run: directory argument required");
|
||||
usage();
|
||||
ExitCode::from(2)
|
||||
}
|
||||
["run", dir] => run(PathBuf::from(dir)),
|
||||
_ => {
|
||||
usage();
|
||||
ExitCode::from(2)
|
||||
}
|
||||
["run", dir] => match run(PathBuf::from(dir)) {
|
||||
Ok(c) => c,
|
||||
Err(e) => { eprintln!("error: {e}"); ExitCode::from(1) }
|
||||
},
|
||||
_ => { usage(); ExitCode::from(2) }
|
||||
}
|
||||
}
|
||||
|
||||
fn run(dir: PathBuf) -> ExitCode {
|
||||
// Tokio current-thread runtime — matches the single-threaded engine design.
|
||||
let rt = match tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
{
|
||||
Ok(r) => r,
|
||||
Err(e) => { eprintln!("error: build runtime: {e}"); return ExitCode::from(1); }
|
||||
};
|
||||
rt.block_on(async move { serve(dir).await })
|
||||
.unwrap_or_else(|e| { eprintln!("error: {e}"); ExitCode::from(1) })
|
||||
}
|
||||
|
||||
async fn serve(dir: PathBuf) -> anyhow::Result<ExitCode> {
|
||||
// 1. Discover
|
||||
fn run(dir: PathBuf) -> anyhow::Result<ExitCode> {
|
||||
// 1. Discover .wo files.
|
||||
let files = rt::discover(&dir)?;
|
||||
if files.is_empty() {
|
||||
anyhow::bail!("no .wo files found under {}", dir.display());
|
||||
|
|
@ -72,11 +64,11 @@ async fn serve(dir: PathBuf) -> anyhow::Result<ExitCode> {
|
|||
dir.display(),
|
||||
);
|
||||
|
||||
// 2. Parse each file into a Schema
|
||||
// 2. Parse each into a Schema.
|
||||
let mut schemas = Vec::new();
|
||||
for f in &files {
|
||||
match rt::parser::parse(&f.src) {
|
||||
Ok(s) => {
|
||||
Ok(s) => {
|
||||
let n = s.types.len();
|
||||
println!(" parsed {} — {} type{}", f.rel.display(), n, if n == 1 { "" } else { "s" });
|
||||
schemas.push(s);
|
||||
|
|
@ -88,39 +80,95 @@ async fn serve(dir: PathBuf) -> anyhow::Result<ExitCode> {
|
|||
}
|
||||
}
|
||||
|
||||
// 3. Compile a catalog
|
||||
// 3. Compile the catalog.
|
||||
let catalog = rt::compile::Catalog::from_schemas(schemas)?;
|
||||
println!("[wo] compiled catalog — {} type{}", catalog.order.len(),
|
||||
if catalog.order.len() == 1 { "" } else { "s" });
|
||||
|
||||
// 4. Boot the engine
|
||||
// 4. Boot the engine + the router.
|
||||
let engine = Arc::new(Mutex::new(rt::engine::Engine::new(catalog.clone())));
|
||||
println!();
|
||||
println!("[wo] routes:");
|
||||
{
|
||||
let e = engine.lock().await;
|
||||
println!();
|
||||
println!("[wo] routes:");
|
||||
let e = engine.lock().unwrap();
|
||||
print!("{}", rt::server::describe_routes(&e));
|
||||
}
|
||||
let router = rt::server::router(engine.clone(), &catalog);
|
||||
|
||||
// 5. Bind and serve
|
||||
// 5. Bind and serve.
|
||||
let addr = std::env::var("WO_LISTEN").unwrap_or_else(|_| "127.0.0.1:8080".to_string());
|
||||
let listener = match tokio::net::TcpListener::bind(&addr).await {
|
||||
Ok(l) => l,
|
||||
Err(e) => anyhow::bail!("bind {addr}: {e}"),
|
||||
};
|
||||
let listener = Listener::bind(&addr)
|
||||
.map_err(|e| anyhow::anyhow!("bind {addr}: {e}"))?;
|
||||
println!();
|
||||
println!("[wo] listening on http://{addr}");
|
||||
println!("[wo] listening on http://{}", listener.local_addr());
|
||||
println!("[wo] ctrl-C to stop");
|
||||
|
||||
let app = rt::server::router(engine.clone(), &catalog);
|
||||
|
||||
// Graceful shutdown on ctrl-C
|
||||
axum::serve(listener, app)
|
||||
.with_graceful_shutdown(async {
|
||||
let _ = tokio::signal::ctrl_c().await;
|
||||
println!("\n[wo] shutting down");
|
||||
})
|
||||
.await?;
|
||||
|
||||
serve_loop(listener, router)?;
|
||||
Ok(ExitCode::from(0))
|
||||
}
|
||||
|
||||
fn serve_loop(listener: Listener, router: Router) -> anyhow::Result<()> {
|
||||
let mut eloop = EventLoop::new()?;
|
||||
let signals = SignalFd::new()?;
|
||||
let listen_fd = listener.as_raw_fd();
|
||||
let signal_fd = signals.as_raw_fd();
|
||||
|
||||
// Tokens: connection fds carry their own raw fd as the token; the
|
||||
// listener and signalfd use their fds too — they're disjoint by
|
||||
// construction (different fds).
|
||||
eloop.register(listen_fd, Interest::READABLE, Token(listen_fd as u64))?;
|
||||
eloop.register(signal_fd, Interest::READABLE, Token(signal_fd as u64))?;
|
||||
|
||||
let mut conns: HashMap<RawFd, Connection> = HashMap::new();
|
||||
|
||||
'outer: loop {
|
||||
let events = match eloop.wait_once(Some(Duration::from_secs(60))) {
|
||||
Ok(evs) => evs,
|
||||
Err(e) => {
|
||||
eprintln!("[wo] event loop error: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
for ev in events {
|
||||
let fd = ev.token().0 as RawFd;
|
||||
|
||||
if fd == listen_fd {
|
||||
// Drain accept queue (edge-triggered).
|
||||
while let Some(cfd) = listener.accept()? {
|
||||
eloop.register(cfd, Interest::READABLE, Token(cfd as u64))?;
|
||||
conns.insert(cfd, Connection::new(cfd));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if fd == signal_fd {
|
||||
let sig = signals.read().unwrap_or(0);
|
||||
println!();
|
||||
println!("[wo] received signal {sig} — shutting down");
|
||||
break 'outer;
|
||||
}
|
||||
|
||||
// Connection event.
|
||||
let Some(conn) = conns.get_mut(&fd) else { continue };
|
||||
let want_writable = match conn.drive(ev.readable, ev.writable, ev.hangup, ev.error, &router) {
|
||||
Ok(w) => w,
|
||||
Err(_) => { conns.remove(&fd); continue; }
|
||||
};
|
||||
|
||||
if conn.is_done() {
|
||||
eloop.deregister(fd).ok();
|
||||
conns.remove(&fd); // Drop closes the fd.
|
||||
} else if want_writable {
|
||||
let _ = eloop.modify(fd, Interest::READ_WRITE, Token(fd as u64));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tear down outstanding connections cleanly. Dropping Connection closes
|
||||
// each fd; deregistering from the loop is optional (close auto-removes).
|
||||
for (fd, _) in conns.drain() {
|
||||
let _ = eloop.deregister(fd);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
247
crates/rt/src/http/connection.rs
Normal file
247
crates/rt/src/http/connection.rs
Normal file
|
|
@ -0,0 +1,247 @@
|
|||
//! Per-connection state machine driven by the phase-02 [`EventLoop`].
|
||||
//!
|
||||
//! Lifecycle (close-after-response, no keep-alive yet):
|
||||
//! Reading → drain `read(2)` to `EAGAIN`, parse, dispatch through the
|
||||
//! `Router`, queue the response.
|
||||
//! Writing → drain `write(2)` to `EAGAIN`. If a write was partial, the
|
||||
//! loop re-arms the fd as `WRITABLE` and we continue on the next event.
|
||||
//! Done → loop closes the fd.
|
||||
//!
|
||||
//! Adapted from `reference/crates/wo-http/src/connection.rs`. The owning
|
||||
//! [`EventLoop`] supplies `read`/`write` readiness via edge-triggered
|
||||
//! `epoll`; this struct is the per-fd part of the state.
|
||||
//!
|
||||
//! [`EventLoop`]: crate::runtime::EventLoop
|
||||
|
||||
use std::io;
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
|
||||
use super::request::{self, ParseResult};
|
||||
use super::response::Response;
|
||||
use super::route::Router;
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum ConnState {
|
||||
Reading,
|
||||
Writing,
|
||||
Done,
|
||||
}
|
||||
|
||||
pub struct Connection {
|
||||
fd: RawFd,
|
||||
state: ConnState,
|
||||
read_buf: Vec<u8>,
|
||||
write_buf: Vec<u8>,
|
||||
write_offset: usize,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
pub fn new(fd: RawFd) -> Self {
|
||||
Self {
|
||||
fd,
|
||||
state: ConnState::Reading,
|
||||
read_buf: Vec::with_capacity(4096),
|
||||
write_buf: Vec::new(),
|
||||
write_offset: 0,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn state(&self) -> ConnState { self.state }
|
||||
pub fn is_done(&self) -> bool { self.state == ConnState::Done }
|
||||
|
||||
/// Drain the socket into `read_buf` until `EAGAIN` or EOF.
|
||||
/// Returns `false` when the peer closed (connection should be torn down).
|
||||
fn drain_read(&mut self) -> io::Result<bool> {
|
||||
let mut tmp = [0u8; 4096];
|
||||
loop {
|
||||
let n = unsafe {
|
||||
libc::read(self.fd, tmp.as_mut_ptr() as *mut libc::c_void, tmp.len())
|
||||
};
|
||||
if n < 0 {
|
||||
let err = io::Error::last_os_error();
|
||||
if err.raw_os_error() == Some(libc::EAGAIN) {
|
||||
return Ok(true);
|
||||
}
|
||||
return Err(err);
|
||||
}
|
||||
if n == 0 {
|
||||
return Ok(false);
|
||||
}
|
||||
self.read_buf.extend_from_slice(&tmp[..n as usize]);
|
||||
// Cap at MAX_BODY_BYTES + headers — refuse pathological requests.
|
||||
if self.read_buf.len() > 32 * 1024 * 1024 {
|
||||
return Err(io::Error::new(io::ErrorKind::InvalidData, "request too large"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Drain `write_buf[write_offset..]` to the socket until `EAGAIN`.
|
||||
/// Returns `true` once everything has been flushed.
|
||||
fn drain_write(&mut self) -> io::Result<bool> {
|
||||
loop {
|
||||
let remaining = &self.write_buf[self.write_offset..];
|
||||
if remaining.is_empty() { return Ok(true); }
|
||||
let n = unsafe {
|
||||
libc::write(
|
||||
self.fd,
|
||||
remaining.as_ptr() as *const libc::c_void,
|
||||
remaining.len(),
|
||||
)
|
||||
};
|
||||
if n < 0 {
|
||||
let err = io::Error::last_os_error();
|
||||
if err.raw_os_error() == Some(libc::EAGAIN) {
|
||||
return Ok(false);
|
||||
}
|
||||
return Err(err);
|
||||
}
|
||||
if n == 0 { return Ok(false); }
|
||||
self.write_offset += n as usize;
|
||||
}
|
||||
}
|
||||
|
||||
fn try_parse(&self) -> ParseResult {
|
||||
request::parse(&self.read_buf)
|
||||
}
|
||||
|
||||
fn queue_response(&mut self, response: &Response) {
|
||||
self.write_buf = response.to_bytes();
|
||||
self.write_offset = 0;
|
||||
self.state = ConnState::Writing;
|
||||
}
|
||||
|
||||
/// One step of the state machine, given a readiness event from the
|
||||
/// loop. Returns `true` if the connection now wants `WRITABLE` (the
|
||||
/// caller should switch interest from `READABLE`); `false` otherwise.
|
||||
pub fn drive(
|
||||
&mut self,
|
||||
readable: bool,
|
||||
writable: bool,
|
||||
hangup: bool,
|
||||
error: bool,
|
||||
router: &Router,
|
||||
) -> io::Result<bool> {
|
||||
if error {
|
||||
self.state = ConnState::Done;
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
if readable && self.state == ConnState::Reading {
|
||||
let still_open = self.drain_read()?;
|
||||
match self.try_parse() {
|
||||
ParseResult::Complete(req) => {
|
||||
let resp = router.dispatch(&req);
|
||||
self.queue_response(&resp);
|
||||
}
|
||||
ParseResult::Incomplete => {
|
||||
if !still_open {
|
||||
self.state = ConnState::Done;
|
||||
return Ok(false);
|
||||
}
|
||||
}
|
||||
ParseResult::Error(msg) => {
|
||||
let resp = Response::status(super::Status::BAD_REQUEST).text(msg);
|
||||
self.queue_response(&resp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if self.state == ConnState::Writing {
|
||||
let flushed = self.drain_write()?;
|
||||
if flushed {
|
||||
self.state = ConnState::Done;
|
||||
return Ok(false);
|
||||
} else if !writable {
|
||||
// We tried, EAGAIN'd; tell the caller to wait for WRITABLE.
|
||||
return Ok(true);
|
||||
}
|
||||
}
|
||||
|
||||
if hangup && self.state != ConnState::Writing {
|
||||
self.state = ConnState::Done;
|
||||
}
|
||||
|
||||
Ok(false)
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRawFd for Connection {
|
||||
fn as_raw_fd(&self) -> RawFd { self.fd }
|
||||
}
|
||||
|
||||
impl Drop for Connection {
|
||||
fn drop(&mut self) {
|
||||
unsafe { libc::close(self.fd); }
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::http::{Method, Response, Router};
|
||||
|
||||
fn socketpair_nonblock() -> (RawFd, RawFd) {
|
||||
let mut fds = [0i32; 2];
|
||||
let r = unsafe {
|
||||
libc::socketpair(
|
||||
libc::AF_UNIX,
|
||||
libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC,
|
||||
0,
|
||||
fds.as_mut_ptr(),
|
||||
)
|
||||
};
|
||||
assert_eq!(r, 0);
|
||||
(fds[0], fds[1])
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn connection_handles_a_request() {
|
||||
let (server_fd, client_fd) = socketpair_nonblock();
|
||||
|
||||
let req = b"GET /healthz HTTP/1.1\r\nHost: localhost\r\n\r\n";
|
||||
let n = unsafe {
|
||||
libc::write(client_fd, req.as_ptr() as *const _, req.len())
|
||||
};
|
||||
assert_eq!(n, req.len() as isize);
|
||||
|
||||
let router = Router::new()
|
||||
.route(Method::Get, "/healthz", |_, _| Response::ok().text("ok"));
|
||||
|
||||
let mut conn = Connection::new(server_fd);
|
||||
let want_writable = conn.drive(true, false, false, false, &router).unwrap();
|
||||
assert!(!want_writable, "small response fits in one write");
|
||||
assert!(conn.is_done());
|
||||
|
||||
let mut buf = [0u8; 4096];
|
||||
let n = unsafe {
|
||||
libc::read(client_fd, buf.as_mut_ptr() as *mut _, buf.len())
|
||||
};
|
||||
assert!(n > 0);
|
||||
let s = std::str::from_utf8(&buf[..n as usize]).unwrap();
|
||||
assert!(s.starts_with("HTTP/1.1 200 OK\r\n"), "got: {s}");
|
||||
assert!(s.ends_with("\r\n\r\nok"));
|
||||
|
||||
unsafe { libc::close(client_fd); }
|
||||
// server_fd is closed by Connection::drop.
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn returns_404_for_unknown_path() {
|
||||
let (server_fd, client_fd) = socketpair_nonblock();
|
||||
|
||||
let req = b"GET /missing HTTP/1.1\r\n\r\n";
|
||||
unsafe { libc::write(client_fd, req.as_ptr() as *const _, req.len()); }
|
||||
|
||||
let router = Router::new()
|
||||
.route(Method::Get, "/healthz", |_, _| Response::ok().text("ok"));
|
||||
let mut conn = Connection::new(server_fd);
|
||||
conn.drive(true, false, false, false, &router).unwrap();
|
||||
|
||||
let mut buf = [0u8; 4096];
|
||||
let n = unsafe { libc::read(client_fd, buf.as_mut_ptr() as *mut _, buf.len()) };
|
||||
let s = std::str::from_utf8(&buf[..n as usize]).unwrap();
|
||||
assert!(s.starts_with("HTTP/1.1 404 Not Found\r\n"), "got: {s}");
|
||||
|
||||
unsafe { libc::close(client_fd); }
|
||||
}
|
||||
}
|
||||
150
crates/rt/src/http/listener.rs
Normal file
150
crates/rt/src/http/listener.rs
Normal file
|
|
@ -0,0 +1,150 @@
|
|||
//! Non-blocking TCP listener — `socket(2)` + `bind(2)` + `listen(2)` + `accept4(2)`.
|
||||
//!
|
||||
//! Adapted from `reference/crates/wo-http/src/listener.rs`. The v1 hand-rolled
|
||||
//! IPv4 parser had a byte-order bug for non-localhost addresses; here we
|
||||
//! defer to `std::net::SocketAddr` (stdlib, no extra crate) and convert the
|
||||
//! resulting octets to a `sockaddr_in` correctly.
|
||||
|
||||
use std::io;
|
||||
use std::net::SocketAddr;
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
|
||||
pub struct Listener {
|
||||
fd: RawFd,
|
||||
addr: SocketAddr,
|
||||
}
|
||||
|
||||
impl Listener {
|
||||
/// Bind to `addr` (IPv4 only for now) and start listening with backlog 128.
|
||||
/// Socket is created `SOCK_NONBLOCK | SOCK_CLOEXEC`.
|
||||
pub fn bind(addr: &str) -> io::Result<Self> {
|
||||
let parsed: SocketAddr = addr.parse().map_err(|e| {
|
||||
io::Error::new(io::ErrorKind::InvalidInput, format!("bad addr {addr:?}: {e}"))
|
||||
})?;
|
||||
let SocketAddr::V4(v4) = parsed else {
|
||||
return Err(io::Error::new(io::ErrorKind::InvalidInput, "IPv4 only for now"));
|
||||
};
|
||||
|
||||
let fd = unsafe {
|
||||
libc::socket(
|
||||
libc::AF_INET,
|
||||
libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC,
|
||||
0,
|
||||
)
|
||||
};
|
||||
if fd < 0 { return Err(io::Error::last_os_error()); }
|
||||
|
||||
// SO_REUSEADDR so the same port restarts cleanly between `wo run`s.
|
||||
let one: libc::c_int = 1;
|
||||
let ret = unsafe {
|
||||
libc::setsockopt(
|
||||
fd, libc::SOL_SOCKET, libc::SO_REUSEADDR,
|
||||
&one as *const _ as *const libc::c_void,
|
||||
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
|
||||
)
|
||||
};
|
||||
if ret < 0 {
|
||||
let err = io::Error::last_os_error();
|
||||
unsafe { libc::close(fd); }
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
let s_addr = u32::from_be_bytes(v4.ip().octets()).to_be();
|
||||
let sock = libc::sockaddr_in {
|
||||
sin_family: libc::AF_INET as libc::sa_family_t,
|
||||
sin_port: v4.port().to_be(),
|
||||
sin_addr: libc::in_addr { s_addr },
|
||||
sin_zero: [0; 8],
|
||||
};
|
||||
let ret = unsafe {
|
||||
libc::bind(
|
||||
fd,
|
||||
&sock as *const _ as *const libc::sockaddr,
|
||||
std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
|
||||
)
|
||||
};
|
||||
if ret < 0 {
|
||||
let err = io::Error::last_os_error();
|
||||
unsafe { libc::close(fd); }
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
if unsafe { libc::listen(fd, 128) } < 0 {
|
||||
let err = io::Error::last_os_error();
|
||||
unsafe { libc::close(fd); }
|
||||
return Err(err);
|
||||
}
|
||||
|
||||
// Resolve the actual bound address — caller may have asked for port 0.
|
||||
let local = read_local_addr(fd)?;
|
||||
Ok(Self { fd, addr: local })
|
||||
}
|
||||
|
||||
/// Accept the next pending connection. Returns `None` on `EAGAIN`.
|
||||
pub fn accept(&self) -> io::Result<Option<RawFd>> {
|
||||
let cfd = unsafe {
|
||||
libc::accept4(
|
||||
self.fd,
|
||||
std::ptr::null_mut(), std::ptr::null_mut(),
|
||||
libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC,
|
||||
)
|
||||
};
|
||||
if cfd < 0 {
|
||||
let err = io::Error::last_os_error();
|
||||
// On Linux EAGAIN == EWOULDBLOCK; one branch is enough.
|
||||
return match err.raw_os_error() {
|
||||
Some(libc::EAGAIN) => Ok(None),
|
||||
_ => Err(err),
|
||||
};
|
||||
}
|
||||
Ok(Some(cfd))
|
||||
}
|
||||
|
||||
pub fn local_addr(&self) -> SocketAddr { self.addr }
|
||||
}
|
||||
|
||||
impl AsRawFd for Listener {
|
||||
fn as_raw_fd(&self) -> RawFd { self.fd }
|
||||
}
|
||||
|
||||
impl Drop for Listener {
|
||||
fn drop(&mut self) {
|
||||
unsafe { libc::close(self.fd); }
|
||||
}
|
||||
}
|
||||
|
||||
fn read_local_addr(fd: RawFd) -> io::Result<SocketAddr> {
|
||||
let mut sock: libc::sockaddr_in = unsafe { std::mem::zeroed() };
|
||||
let mut len = std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t;
|
||||
let ret = unsafe {
|
||||
libc::getsockname(fd, &mut sock as *mut _ as *mut libc::sockaddr, &mut len)
|
||||
};
|
||||
if ret < 0 { return Err(io::Error::last_os_error()); }
|
||||
let ip = u32::from_be(sock.sin_addr.s_addr).to_be_bytes();
|
||||
let port = u16::from_be(sock.sin_port);
|
||||
Ok(SocketAddr::from(([ip[0], ip[1], ip[2], ip[3]], port)))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn binds_and_accepts_a_client() {
|
||||
let listener = Listener::bind("127.0.0.1:0").unwrap();
|
||||
let port = listener.local_addr().port();
|
||||
assert!(port > 0);
|
||||
assert!(listener.accept().unwrap().is_none(), "no clients pending yet");
|
||||
|
||||
let stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
|
||||
// Block briefly until the server-side accept sees it.
|
||||
let mut accepted = None;
|
||||
for _ in 0..50 {
|
||||
if let Some(fd) = listener.accept().unwrap() { accepted = Some(fd); break; }
|
||||
std::thread::sleep(std::time::Duration::from_millis(10));
|
||||
}
|
||||
let cfd = accepted.expect("accept produced a client fd");
|
||||
unsafe { libc::close(cfd); }
|
||||
drop(stream);
|
||||
}
|
||||
}
|
||||
21
crates/rt/src/http/mod.rs
Normal file
21
crates/rt/src/http/mod.rs
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
//! Hand-rolled HTTP/1.1 — phase 03 of the runtime plan.
|
||||
//!
|
||||
//! The transport layer for the `wo` binary after the phase-04 cutover.
|
||||
//! Drives non-blocking accept + per-connection state machines off the
|
||||
//! phase-02 [`EventLoop`](super::runtime::EventLoop). No `tokio`, no
|
||||
//! `axum`, no `hyper`. Synchronous handlers; close-after-response (the
|
||||
//! v1 model — keep-alive lands when a sample needs it).
|
||||
//!
|
||||
//! See `docs/plan/03-hand-rolled-http.md` and `docs/plan/04-cutover-remove-tokio-axum.md`.
|
||||
|
||||
mod connection;
|
||||
mod listener;
|
||||
mod request;
|
||||
mod response;
|
||||
mod route;
|
||||
|
||||
pub use connection::{Connection, ConnState};
|
||||
pub use listener::Listener;
|
||||
pub use request::{Method, Request};
|
||||
pub use response::{Response, Status};
|
||||
pub use route::{RouteParams, Router};
|
||||
179
crates/rt/src/http/request.rs
Normal file
179
crates/rt/src/http/request.rs
Normal file
|
|
@ -0,0 +1,179 @@
|
|||
//! Incremental HTTP/1.1 request parser.
|
||||
//!
|
||||
//! Adapted from `reference/crates/wo-http/src/request.rs`. v1 only parsed
|
||||
//! request headers (the v1 blog is read-only HTML). The phase-04 cutover
|
||||
//! needs JSON request bodies, so this parser also drains a
|
||||
//! `Content-Length`-delimited body. Chunked transfer encoding is not
|
||||
//! supported (no sample sends one — see `docs/plan/03-hand-rolled-http.md`).
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
pub enum Method {
|
||||
Get,
|
||||
Post,
|
||||
Patch,
|
||||
Put,
|
||||
Delete,
|
||||
Head,
|
||||
Options,
|
||||
Other,
|
||||
}
|
||||
|
||||
impl Method {
|
||||
fn parse(token: &str) -> Method {
|
||||
match token {
|
||||
"GET" => Method::Get,
|
||||
"POST" => Method::Post,
|
||||
"PATCH" => Method::Patch,
|
||||
"PUT" => Method::Put,
|
||||
"DELETE" => Method::Delete,
|
||||
"HEAD" => Method::Head,
|
||||
"OPTIONS" => Method::Options,
|
||||
_ => Method::Other,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Request {
|
||||
pub method: Method,
|
||||
pub path: String,
|
||||
pub query: Option<String>,
|
||||
pub headers: HashMap<String, String>,
|
||||
pub body: Vec<u8>,
|
||||
}
|
||||
|
||||
pub enum ParseResult {
|
||||
Complete(Request),
|
||||
Incomplete,
|
||||
Error(String),
|
||||
}
|
||||
|
||||
const MAX_HEADER_BYTES: usize = 8 * 1024;
|
||||
const MAX_BODY_BYTES: usize = 16 * 1024 * 1024;
|
||||
|
||||
/// Try to parse a complete HTTP/1.1 request out of `buf`.
|
||||
/// Returns `Complete(req, bytes_consumed)` once headers + body are present.
|
||||
pub fn parse(buf: &[u8]) -> ParseResult {
|
||||
let header_end = match find_header_end(buf) {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
if buf.len() > MAX_HEADER_BYTES {
|
||||
return ParseResult::Error("request headers too large".into());
|
||||
}
|
||||
return ParseResult::Incomplete;
|
||||
}
|
||||
};
|
||||
|
||||
let header_str = match std::str::from_utf8(&buf[..header_end]) {
|
||||
Ok(s) => s,
|
||||
Err(_) => return ParseResult::Error("non-UTF-8 in headers".into()),
|
||||
};
|
||||
|
||||
let mut lines = header_str.lines();
|
||||
|
||||
let request_line = match lines.next() {
|
||||
Some(l) => l,
|
||||
None => return ParseResult::Error("empty request".into()),
|
||||
};
|
||||
let mut parts = request_line.split_whitespace();
|
||||
let method = match parts.next() {
|
||||
Some(t) => Method::parse(t),
|
||||
None => return ParseResult::Error("missing method".into()),
|
||||
};
|
||||
let raw_path = match parts.next() {
|
||||
Some(p) => p,
|
||||
None => return ParseResult::Error("missing path".into()),
|
||||
};
|
||||
let (path, query) = match raw_path.split_once('?') {
|
||||
Some((p, q)) => (p.to_string(), Some(q.to_string())),
|
||||
None => (raw_path.to_string(), None),
|
||||
};
|
||||
|
||||
let mut headers = HashMap::new();
|
||||
for line in lines {
|
||||
if line.is_empty() { break; }
|
||||
if let Some((k, v)) = line.split_once(':') {
|
||||
headers.insert(k.trim().to_ascii_lowercase(), v.trim().to_string());
|
||||
}
|
||||
}
|
||||
|
||||
let body_len: usize = headers
|
||||
.get("content-length")
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(0);
|
||||
if body_len > MAX_BODY_BYTES {
|
||||
return ParseResult::Error("Content-Length exceeds limit".into());
|
||||
}
|
||||
|
||||
let header_bytes = header_end + 4; // include the trailing \r\n\r\n
|
||||
let total = header_bytes + body_len;
|
||||
if buf.len() < total {
|
||||
return ParseResult::Incomplete;
|
||||
}
|
||||
|
||||
let body = buf[header_bytes..total].to_vec();
|
||||
ParseResult::Complete(Request { method, path, query, headers, body })
|
||||
}
|
||||
|
||||
fn find_header_end(buf: &[u8]) -> Option<usize> {
|
||||
buf.windows(4).position(|w| w == b"\r\n\r\n")
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_simple_get() {
|
||||
let raw = b"GET /api/articles HTTP/1.1\r\nHost: localhost\r\n\r\n";
|
||||
let ParseResult::Complete(req) = parse(raw) else { panic!("expected Complete") };
|
||||
assert_eq!(req.method, Method::Get);
|
||||
assert_eq!(req.path, "/api/articles");
|
||||
assert!(req.query.is_none());
|
||||
assert!(req.body.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_query_string() {
|
||||
let raw = b"GET /tag/rust?page=2 HTTP/1.1\r\n\r\n";
|
||||
let ParseResult::Complete(req) = parse(raw) else { panic!() };
|
||||
assert_eq!(req.path, "/tag/rust");
|
||||
assert_eq!(req.query.as_deref(), Some("page=2"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_post_with_body() {
|
||||
let body = b"{\"title\":\"hi\"}";
|
||||
let mut raw = Vec::new();
|
||||
raw.extend_from_slice(b"POST /api/articles HTTP/1.1\r\n");
|
||||
raw.extend_from_slice(b"Host: localhost\r\n");
|
||||
raw.extend_from_slice(b"Content-Type: application/json\r\n");
|
||||
raw.extend_from_slice(format!("Content-Length: {}\r\n", body.len()).as_bytes());
|
||||
raw.extend_from_slice(b"\r\n");
|
||||
raw.extend_from_slice(body);
|
||||
let ParseResult::Complete(req) = parse(&raw) else { panic!() };
|
||||
assert_eq!(req.method, Method::Post);
|
||||
assert_eq!(req.body, body);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn incomplete_when_body_truncated() {
|
||||
let raw = b"POST / HTTP/1.1\r\nContent-Length: 10\r\n\r\nshort";
|
||||
assert!(matches!(parse(raw), ParseResult::Incomplete));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn incomplete_when_headers_truncated() {
|
||||
let raw = b"GET / HTTP/1.1\r\nHost: local";
|
||||
assert!(matches!(parse(raw), ParseResult::Incomplete));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_patch_method() {
|
||||
let raw = b"PATCH /api/articles/1 HTTP/1.1\r\nContent-Length: 0\r\n\r\n";
|
||||
let ParseResult::Complete(req) = parse(raw) else { panic!() };
|
||||
assert_eq!(req.method, Method::Patch);
|
||||
}
|
||||
}
|
||||
113
crates/rt/src/http/response.rs
Normal file
113
crates/rt/src/http/response.rs
Normal file
|
|
@ -0,0 +1,113 @@
|
|||
//! HTTP/1.1 response builder + serializer.
|
||||
//!
|
||||
//! Adapted from `reference/crates/wo-http/src/response.rs`. Adds:
|
||||
//! * `Status` constants for the codes the REST samples assert on
|
||||
//! (200/201/204/400/404/405/500/501).
|
||||
//! * `Response::json(&serde_json::Value)` matching the cutover-handler
|
||||
//! shape in `docs/plan/04-cutover-remove-tokio-axum.md`.
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct Status(pub u16, pub &'static str);
|
||||
|
||||
impl Status {
|
||||
pub const OK: Status = Status(200, "OK");
|
||||
pub const CREATED: Status = Status(201, "Created");
|
||||
pub const NO_CONTENT: Status = Status(204, "No Content");
|
||||
pub const BAD_REQUEST: Status = Status(400, "Bad Request");
|
||||
pub const NOT_FOUND: Status = Status(404, "Not Found");
|
||||
pub const METHOD_NOT_ALLOWED: Status = Status(405, "Method Not Allowed");
|
||||
pub const INTERNAL_SERVER_ERROR: Status = Status(500, "Internal Server Error");
|
||||
pub const NOT_IMPLEMENTED: Status = Status(501, "Not Implemented");
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Response {
|
||||
pub status: Status,
|
||||
pub headers: Vec<(String, String)>,
|
||||
pub body: Vec<u8>,
|
||||
}
|
||||
|
||||
impl Response {
|
||||
pub fn status(s: Status) -> Self {
|
||||
Self { status: s, headers: Vec::new(), body: Vec::new() }
|
||||
}
|
||||
|
||||
pub fn ok() -> Self { Self::status(Status::OK) }
|
||||
pub fn created() -> Self { Self::status(Status::CREATED) }
|
||||
pub fn no_content() -> Self { Self::status(Status::NO_CONTENT) }
|
||||
|
||||
pub fn header(mut self, k: &str, v: &str) -> Self {
|
||||
self.headers.push((k.to_string(), v.to_string()));
|
||||
self
|
||||
}
|
||||
|
||||
pub fn body(mut self, body: impl Into<Vec<u8>>) -> Self {
|
||||
self.body = body.into();
|
||||
self
|
||||
}
|
||||
|
||||
/// Plain-text body with `Content-Type: text/plain; charset=utf-8`.
|
||||
pub fn text(self, body: impl Into<String>) -> Self {
|
||||
let body: String = body.into();
|
||||
self.header("Content-Type", "text/plain; charset=utf-8")
|
||||
.body(body.into_bytes())
|
||||
}
|
||||
|
||||
/// JSON body with `Content-Type: application/json`.
|
||||
pub fn json(self, value: &Value) -> Self {
|
||||
let buf = serde_json::to_vec(value).unwrap_or_else(|_| b"null".to_vec());
|
||||
self.header("Content-Type", "application/json").body(buf)
|
||||
}
|
||||
|
||||
/// Serialize to the wire format. Auto-injects `Content-Length` and
|
||||
/// `Connection: close` (no keep-alive in Stage 2).
|
||||
pub fn to_bytes(&self) -> Vec<u8> {
|
||||
let mut buf = Vec::with_capacity(256 + self.body.len());
|
||||
buf.extend_from_slice(
|
||||
format!("HTTP/1.1 {} {}\r\n", self.status.0, self.status.1).as_bytes(),
|
||||
);
|
||||
for (k, v) in &self.headers {
|
||||
buf.extend_from_slice(format!("{k}: {v}\r\n").as_bytes());
|
||||
}
|
||||
buf.extend_from_slice(format!("Content-Length: {}\r\n", self.body.len()).as_bytes());
|
||||
buf.extend_from_slice(b"Connection: close\r\n");
|
||||
buf.extend_from_slice(b"\r\n");
|
||||
buf.extend_from_slice(&self.body);
|
||||
buf
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn ok_text_body() {
|
||||
let r = Response::ok().text("ok");
|
||||
let s = String::from_utf8(r.to_bytes()).unwrap();
|
||||
assert!(s.starts_with("HTTP/1.1 200 OK\r\n"));
|
||||
assert!(s.contains("Content-Type: text/plain"));
|
||||
assert!(s.contains("Content-Length: 2\r\n"));
|
||||
assert!(s.ends_with("\r\n\r\nok"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_body() {
|
||||
let r = Response::created().json(&json!({"id": 1, "title": "Hi"}));
|
||||
let s = String::from_utf8(r.to_bytes()).unwrap();
|
||||
assert!(s.starts_with("HTTP/1.1 201 Created\r\n"));
|
||||
assert!(s.contains("Content-Type: application/json"));
|
||||
assert!(s.contains(r#"{"id":1,"title":"Hi"}"#));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_content_status() {
|
||||
let r = Response::no_content();
|
||||
let s = String::from_utf8(r.to_bytes()).unwrap();
|
||||
assert!(s.starts_with("HTTP/1.1 204 No Content\r\n"));
|
||||
assert!(s.ends_with("\r\n\r\n"));
|
||||
}
|
||||
}
|
||||
182
crates/rt/src/http/route.rs
Normal file
182
crates/rt/src/http/route.rs
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
//! Method + URL pattern → handler dispatch.
|
||||
//!
|
||||
//! Combined adaptation of `reference/crates/wo-route/src/{router,pattern}.rs`.
|
||||
//! Handler shape is `Fn(&Request, &RouteParams) -> Response`, captured as a
|
||||
//! boxed closure so each route closes over its own state (typically an
|
||||
//! `Arc<Mutex<Engine>>` — see `crates/rt/src/server.rs`).
|
||||
//!
|
||||
//! Dispatch distinguishes 404 (no path matches any registered route) from
|
||||
//! 405 (path matches at least one route but not for the request's method),
|
||||
//! which the REST sample asserts (`expose list, get, ...` → POST → 405).
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use super::{Method, Request, Response, Status};
|
||||
|
||||
pub type HandlerFn = dyn Fn(&Request, &RouteParams) -> Response + Send + Sync + 'static;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
enum Segment {
|
||||
Literal(String),
|
||||
Param(String),
|
||||
Wildcard(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct Pattern {
|
||||
segments: Vec<Segment>,
|
||||
}
|
||||
|
||||
impl Pattern {
|
||||
fn compile(s: &str) -> Self {
|
||||
let segments = s.trim_start_matches('/')
|
||||
.split('/')
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(|seg| {
|
||||
if let Some(name) = seg.strip_prefix(':') {
|
||||
Segment::Param(name.to_string())
|
||||
} else if let Some(name) = seg.strip_prefix('*') {
|
||||
Segment::Wildcard(name.to_string())
|
||||
} else {
|
||||
Segment::Literal(seg.to_string())
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
Self { segments }
|
||||
}
|
||||
|
||||
fn matches(&self, path: &str) -> Option<Vec<(String, String)>> {
|
||||
let parts: Vec<&str> = path.trim_start_matches('/')
|
||||
.split('/')
|
||||
.filter(|s| !s.is_empty())
|
||||
.collect();
|
||||
let mut params = Vec::new();
|
||||
let mut pi = 0usize;
|
||||
for seg in &self.segments {
|
||||
match seg {
|
||||
Segment::Literal(lit) => {
|
||||
if pi >= parts.len() || parts[pi] != lit { return None; }
|
||||
pi += 1;
|
||||
}
|
||||
Segment::Param(name) => {
|
||||
if pi >= parts.len() { return None; }
|
||||
params.push((name.clone(), parts[pi].to_string()));
|
||||
pi += 1;
|
||||
}
|
||||
Segment::Wildcard(name) => {
|
||||
if pi >= parts.len() { return None; }
|
||||
params.push((name.clone(), parts[pi..].join("/")));
|
||||
return Some(params);
|
||||
}
|
||||
}
|
||||
}
|
||||
if pi == parts.len() { Some(params) } else { None }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct RouteParams {
|
||||
params: HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl RouteParams {
|
||||
pub fn get(&self, k: &str) -> Option<&str> {
|
||||
self.params.get(k).map(String::as_str)
|
||||
}
|
||||
|
||||
fn from_pairs(pairs: Vec<(String, String)>) -> Self {
|
||||
Self { params: pairs.into_iter().collect() }
|
||||
}
|
||||
}
|
||||
|
||||
struct Route {
|
||||
method: Method,
|
||||
pattern: Pattern,
|
||||
handler: Box<HandlerFn>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct Router {
|
||||
routes: Vec<Route>,
|
||||
}
|
||||
|
||||
impl Router {
|
||||
pub fn new() -> Self { Self::default() }
|
||||
|
||||
pub fn route<F>(mut self, method: Method, pattern: &str, handler: F) -> Self
|
||||
where
|
||||
F: Fn(&Request, &RouteParams) -> Response + Send + Sync + 'static,
|
||||
{
|
||||
self.routes.push(Route {
|
||||
method,
|
||||
pattern: Pattern::compile(pattern),
|
||||
handler: Box::new(handler),
|
||||
});
|
||||
self
|
||||
}
|
||||
|
||||
/// Resolve a request to a response. 404 if no path matches; 405 if the
|
||||
/// path matches a registered route under a different method.
|
||||
pub fn dispatch(&self, req: &Request) -> Response {
|
||||
let mut path_matched_any = false;
|
||||
for r in &self.routes {
|
||||
if let Some(pairs) = r.pattern.matches(&req.path) {
|
||||
if r.method == req.method {
|
||||
let params = RouteParams::from_pairs(pairs);
|
||||
return (r.handler)(req, ¶ms);
|
||||
}
|
||||
path_matched_any = true;
|
||||
}
|
||||
}
|
||||
if path_matched_any {
|
||||
Response::status(Status::METHOD_NOT_ALLOWED).text("method not allowed")
|
||||
} else {
|
||||
Response::status(Status::NOT_FOUND).text("no route")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::collections::HashMap;
|
||||
|
||||
fn req(method: Method, path: &str) -> Request {
|
||||
Request { method, path: path.into(), query: None, headers: HashMap::new(), body: vec![] }
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dispatches_static_and_param_routes() {
|
||||
let r = Router::new()
|
||||
.route(Method::Get, "/healthz", |_, _| Response::ok().text("ok"))
|
||||
.route(Method::Get, "/api/articles", |_, _| Response::ok().text("list"))
|
||||
.route(Method::Get, "/api/articles/:id", |_, p| Response::ok().text(format!("get {}", p.get("id").unwrap())))
|
||||
.route(Method::Post, "/api/articles", |_, _| Response::created().text("create"));
|
||||
|
||||
let resp = r.dispatch(&req(Method::Get, "/healthz"));
|
||||
assert_eq!(resp.status.0, 200);
|
||||
assert_eq!(resp.body, b"ok");
|
||||
|
||||
let resp = r.dispatch(&req(Method::Get, "/api/articles/42"));
|
||||
assert_eq!(resp.body, b"get 42");
|
||||
|
||||
let resp = r.dispatch(&req(Method::Post, "/api/articles"));
|
||||
assert_eq!(resp.status.0, 201);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn returns_405_when_path_matches_but_method_does_not() {
|
||||
let r = Router::new()
|
||||
.route(Method::Get, "/api/articles", |_, _| Response::ok().text("list"));
|
||||
let resp = r.dispatch(&req(Method::Post, "/api/articles"));
|
||||
assert_eq!(resp.status.0, 405);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn returns_404_when_no_path_matches() {
|
||||
let r = Router::new()
|
||||
.route(Method::Get, "/api/articles", |_, _| Response::ok().text("list"));
|
||||
let resp = r.dispatch(&req(Method::Get, "/nope"));
|
||||
assert_eq!(resp.status.0, 404);
|
||||
}
|
||||
}
|
||||
|
|
@ -7,6 +7,7 @@
|
|||
pub mod ast;
|
||||
pub mod compile;
|
||||
pub mod engine;
|
||||
pub mod http;
|
||||
pub mod lexer;
|
||||
pub mod parser;
|
||||
pub mod runtime;
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
//! Axum REST server built from `service rest` blocks in the compiled catalog.
|
||||
//! REST routing built from `service rest` blocks in the compiled catalog.
|
||||
//!
|
||||
//! For each type that declares `service rest "/path" expose ...`, we bind the
|
||||
//! exposed operations at the given path:
|
||||
|
|
@ -9,61 +9,44 @@
|
|||
//! create POST /path
|
||||
//! update PATCH /path/:id
|
||||
//! delete DELETE /path/:id
|
||||
//! subscribe stubbed in Stage 2; wires up in Stage 3
|
||||
//! me GET /path/me (stubbed 501)
|
||||
//! subscribe stubbed in Stage 2; wires up in Stage 3 (501)
|
||||
//! me GET /path/me (501)
|
||||
//! ```
|
||||
//!
|
||||
//! The engine lives behind a single `Arc<Mutex<Engine>>` — matches the
|
||||
//! single-threaded event-loop design of the runtime (see Phase 2 Concurrency
|
||||
//! Model).
|
||||
//! Phase-04 cutover: the axum + tokio backend was replaced with the
|
||||
//! hand-rolled [`http::Router`](crate::http::Router) running on the
|
||||
//! phase-02 [`EventLoop`](crate::runtime::EventLoop). Handlers are
|
||||
//! synchronous, the engine sits behind `Arc<std::sync::Mutex<_>>`, and
|
||||
//! the binary owns one event loop on one thread.
|
||||
|
||||
use crate::ast::{Operation, ServiceKind};
|
||||
use crate::compile::Catalog;
|
||||
use crate::engine::Engine;
|
||||
use crate::http::{Method, Request, Response, RouteParams, Router, Status};
|
||||
|
||||
use axum::{
|
||||
extract::{Path, State},
|
||||
http::StatusCode,
|
||||
response::IntoResponse,
|
||||
routing::{get, post},
|
||||
Json, Router,
|
||||
};
|
||||
use serde_json::{json, Value};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
pub type Shared = Arc<Mutex<Engine>>;
|
||||
|
||||
/// Per-route type context carried via axum `State`.
|
||||
#[derive(Clone)]
|
||||
struct TypeState {
|
||||
engine: Shared,
|
||||
ty: Arc<String>,
|
||||
}
|
||||
|
||||
/// Build the fully-wired axum `Router` for a running engine.
|
||||
/// The caller keeps the `Shared` handle alongside for Stage 3 subscription wiring.
|
||||
pub fn router(engine: Shared, catalog: &crate::compile::Catalog) -> Router {
|
||||
let mut app = Router::new()
|
||||
.route("/", get(root))
|
||||
.route("/healthz", get(|| async { "ok" }));
|
||||
/// Build the fully-wired [`Router`] for a running engine.
|
||||
pub fn router(engine: Shared, catalog: &Catalog) -> Router {
|
||||
let mut r = Router::new()
|
||||
.route(Method::Get, "/", |_, _| root_response())
|
||||
.route(Method::Get, "/healthz", |_, _| Response::ok().text("ok"));
|
||||
|
||||
for name in &catalog.order {
|
||||
let t = catalog.get(name).expect("type present");
|
||||
for svc in &t.services {
|
||||
if svc.kind != ServiceKind::Rest { continue; }
|
||||
let state = TypeState {
|
||||
engine: engine.clone(),
|
||||
ty: Arc::new(t.name.clone()),
|
||||
};
|
||||
app = attach_rest(app, state, svc.path.clone(), &svc.expose);
|
||||
r = attach_rest(r, engine.clone(), t.name.clone(), svc.path.clone(), &svc.expose);
|
||||
}
|
||||
}
|
||||
|
||||
app
|
||||
r
|
||||
}
|
||||
|
||||
async fn root() -> impl IntoResponse {
|
||||
Json(json!({
|
||||
fn root_response() -> Response {
|
||||
Response::ok().json(&json!({
|
||||
"runtime": "wo",
|
||||
"stage": 2,
|
||||
"notes": "REST CRUD for each `service rest` block. /healthz for liveness. LIVE subscribe in Stage 3."
|
||||
|
|
@ -71,122 +54,136 @@ async fn root() -> impl IntoResponse {
|
|||
}
|
||||
|
||||
fn attach_rest(
|
||||
mut app: Router,
|
||||
state: TypeState,
|
||||
path: String,
|
||||
ops: &[Operation],
|
||||
mut r: Router,
|
||||
engine: Shared,
|
||||
ty: String,
|
||||
path: String,
|
||||
ops: &[Operation],
|
||||
) -> Router {
|
||||
// Build a per-type sub-router with shared state, then merge.
|
||||
let mut sub = Router::new();
|
||||
let id_path = format!("{path}/:id");
|
||||
|
||||
// Collect collection-path handlers and id-path handlers separately so
|
||||
// axum's `method_routing` merges correctly.
|
||||
let mut collection = None::<axum::routing::MethodRouter<TypeState>>;
|
||||
let mut by_id = None::<axum::routing::MethodRouter<TypeState>>;
|
||||
// Register literal sub-paths (`/live`, `/me`) BEFORE the `/:id` param
|
||||
// route — the first matching pattern wins, so `:id` would otherwise
|
||||
// swallow "live" / "me" and produce a 400 invalid-id response.
|
||||
for op in ops {
|
||||
match op {
|
||||
Operation::Subscribe => {
|
||||
r = r.route(Method::Get, &format!("{path}/live"),
|
||||
|_, _| Response::status(Status::NOT_IMPLEMENTED).text("LIVE subscriptions arrive in Stage 3"));
|
||||
}
|
||||
Operation::Me => {
|
||||
r = r.route(Method::Get, &format!("{path}/me"),
|
||||
|_, _| Response::status(Status::NOT_IMPLEMENTED).text("session layer not yet implemented"));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
for op in ops {
|
||||
match op {
|
||||
Operation::List => {
|
||||
collection = Some(match collection.take() {
|
||||
Some(r) => r.get(list_h),
|
||||
None => get(list_h),
|
||||
});
|
||||
let eng = engine.clone(); let ty = ty.clone();
|
||||
r = r.route(Method::Get, &path, move |req, params| list_h(&eng, &ty, req, params));
|
||||
}
|
||||
Operation::Create => {
|
||||
collection = Some(match collection.take() {
|
||||
Some(r) => r.post(create_h),
|
||||
None => post(create_h),
|
||||
});
|
||||
let eng = engine.clone(); let ty = ty.clone();
|
||||
r = r.route(Method::Post, &path, move |req, params| create_h(&eng, &ty, req, params));
|
||||
}
|
||||
Operation::Get => {
|
||||
by_id = Some(match by_id.take() {
|
||||
Some(r) => r.get(get_h),
|
||||
None => get(get_h),
|
||||
});
|
||||
let eng = engine.clone(); let ty = ty.clone();
|
||||
r = r.route(Method::Get, &id_path, move |req, params| get_h(&eng, &ty, req, params));
|
||||
}
|
||||
Operation::Update => {
|
||||
by_id = Some(match by_id.take() {
|
||||
Some(r) => r.patch(update_h),
|
||||
None => axum::routing::patch(update_h),
|
||||
});
|
||||
let eng = engine.clone(); let ty = ty.clone();
|
||||
r = r.route(Method::Patch, &id_path, move |req, params| update_h(&eng, &ty, req, params));
|
||||
}
|
||||
Operation::Delete => {
|
||||
by_id = Some(match by_id.take() {
|
||||
Some(r) => r.delete(delete_h),
|
||||
None => axum::routing::delete(delete_h),
|
||||
});
|
||||
let eng = engine.clone(); let ty = ty.clone();
|
||||
r = r.route(Method::Delete, &id_path, move |req, params| delete_h(&eng, &ty, req, params));
|
||||
}
|
||||
Operation::Subscribe => {
|
||||
let p = format!("{path}/live");
|
||||
sub = sub.route(&p, get(subscribe_stub));
|
||||
}
|
||||
Operation::Me => {
|
||||
let p = format!("{path}/me");
|
||||
sub = sub.route(&p, get(me_stub));
|
||||
}
|
||||
Operation::Custom => { /* reserved */ }
|
||||
Operation::Subscribe | Operation::Me | Operation::Custom => {}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(r) = collection { sub = sub.route(&path, r); }
|
||||
if let Some(r) = by_id { sub = sub.route(&format!("{path}/:id"), r); }
|
||||
|
||||
let sub = sub.with_state(state);
|
||||
app = app.merge(sub);
|
||||
app
|
||||
r
|
||||
}
|
||||
|
||||
// --- handlers ---
|
||||
|
||||
async fn list_h(State(st): State<TypeState>) -> impl IntoResponse {
|
||||
let eng = st.engine.lock().await;
|
||||
match eng.list(&st.ty) {
|
||||
Ok(rows) => (StatusCode::OK, Json(json!(rows))).into_response(),
|
||||
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
|
||||
fn list_h(engine: &Shared, ty: &str, _req: &Request, _params: &RouteParams) -> Response {
|
||||
let eng = engine.lock().unwrap();
|
||||
match eng.list(ty) {
|
||||
Ok(rows) => Response::ok().json(&json!(rows)),
|
||||
Err(e) => Response::status(Status::INTERNAL_SERVER_ERROR).text(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_h(State(st): State<TypeState>, Path(id): Path<i64>) -> impl IntoResponse {
|
||||
let eng = st.engine.lock().await;
|
||||
match eng.get(&st.ty, id) {
|
||||
Ok(Some(row)) => (StatusCode::OK, Json(json!(row))).into_response(),
|
||||
Ok(None) => (StatusCode::NOT_FOUND, format!("no {} with id {id}", st.ty)).into_response(),
|
||||
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
|
||||
fn get_h(engine: &Shared, ty: &str, _req: &Request, params: &RouteParams) -> Response {
|
||||
let id = match parse_id(params) {
|
||||
Ok(id) => id,
|
||||
Err(r) => return r,
|
||||
};
|
||||
let eng = engine.lock().unwrap();
|
||||
match eng.get(ty, id) {
|
||||
Ok(Some(row)) => Response::ok().json(&json!(row)),
|
||||
Ok(None) => Response::status(Status::NOT_FOUND).text(format!("no {ty} with id {id}")),
|
||||
Err(e) => Response::status(Status::INTERNAL_SERVER_ERROR).text(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn create_h(State(st): State<TypeState>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
let mut eng = st.engine.lock().await;
|
||||
match eng.create(&st.ty, body) {
|
||||
Ok(row) => (StatusCode::CREATED, Json(json!(row))).into_response(),
|
||||
Err(e) => (StatusCode::BAD_REQUEST, e.to_string()).into_response(),
|
||||
fn create_h(engine: &Shared, ty: &str, req: &Request, _params: &RouteParams) -> Response {
|
||||
let body = match parse_json_body(req) {
|
||||
Ok(v) => v,
|
||||
Err(r) => return r,
|
||||
};
|
||||
let mut eng = engine.lock().unwrap();
|
||||
match eng.create(ty, body) {
|
||||
Ok(row) => Response::status(Status::CREATED).json(&json!(row)),
|
||||
Err(e) => Response::status(Status::BAD_REQUEST).text(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn update_h(State(st): State<TypeState>, Path(id): Path<i64>, Json(body): Json<Value>) -> impl IntoResponse {
|
||||
let mut eng = st.engine.lock().await;
|
||||
match eng.update(&st.ty, id, body) {
|
||||
Ok(Some(row)) => (StatusCode::OK, Json(json!(row))).into_response(),
|
||||
Ok(None) => (StatusCode::NOT_FOUND, format!("no {} with id {id}", st.ty)).into_response(),
|
||||
Err(e) => (StatusCode::BAD_REQUEST, e.to_string()).into_response(),
|
||||
fn update_h(engine: &Shared, ty: &str, req: &Request, params: &RouteParams) -> Response {
|
||||
let id = match parse_id(params) {
|
||||
Ok(id) => id,
|
||||
Err(r) => return r,
|
||||
};
|
||||
let body = match parse_json_body(req) {
|
||||
Ok(v) => v,
|
||||
Err(r) => return r,
|
||||
};
|
||||
let mut eng = engine.lock().unwrap();
|
||||
match eng.update(ty, id, body) {
|
||||
Ok(Some(row)) => Response::ok().json(&json!(row)),
|
||||
Ok(None) => Response::status(Status::NOT_FOUND).text(format!("no {ty} with id {id}")),
|
||||
Err(e) => Response::status(Status::BAD_REQUEST).text(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete_h(State(st): State<TypeState>, Path(id): Path<i64>) -> impl IntoResponse {
|
||||
let mut eng = st.engine.lock().await;
|
||||
match eng.delete(&st.ty, id) {
|
||||
Ok(true) => StatusCode::NO_CONTENT.into_response(),
|
||||
Ok(false) => (StatusCode::NOT_FOUND, format!("no {} with id {id}", st.ty)).into_response(),
|
||||
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
|
||||
fn delete_h(engine: &Shared, ty: &str, _req: &Request, params: &RouteParams) -> Response {
|
||||
let id = match parse_id(params) {
|
||||
Ok(id) => id,
|
||||
Err(r) => return r,
|
||||
};
|
||||
let mut eng = engine.lock().unwrap();
|
||||
match eng.delete(ty, id) {
|
||||
Ok(true) => Response::no_content(),
|
||||
Ok(false) => Response::status(Status::NOT_FOUND).text(format!("no {ty} with id {id}")),
|
||||
Err(e) => Response::status(Status::INTERNAL_SERVER_ERROR).text(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn subscribe_stub() -> impl IntoResponse {
|
||||
(StatusCode::NOT_IMPLEMENTED, "LIVE subscriptions arrive in Stage 3")
|
||||
fn parse_id(params: &RouteParams) -> Result<i64, Response> {
|
||||
params.get("id")
|
||||
.and_then(|s| s.parse::<i64>().ok())
|
||||
.ok_or_else(|| Response::status(Status::BAD_REQUEST).text("invalid id"))
|
||||
}
|
||||
|
||||
async fn me_stub() -> impl IntoResponse {
|
||||
(StatusCode::NOT_IMPLEMENTED, "session layer not yet implemented")
|
||||
fn parse_json_body(req: &Request) -> Result<Value, Response> {
|
||||
if req.body.is_empty() {
|
||||
return Ok(Value::Object(Default::default()));
|
||||
}
|
||||
serde_json::from_slice::<Value>(&req.body)
|
||||
.map_err(|e| Response::status(Status::BAD_REQUEST).text(format!("invalid JSON: {e}")))
|
||||
}
|
||||
|
||||
/// Format the endpoint banner the CLI prints on startup.
|
||||
|
|
@ -200,13 +197,13 @@ pub fn describe_routes(engine: &Engine) -> String {
|
|||
if svc.kind != ServiceKind::Rest { continue; }
|
||||
for op in &svc.expose {
|
||||
let (m, p, n) = match op {
|
||||
Operation::List => ("GET ", svc.path.clone(), format!("list {name}")),
|
||||
Operation::Get => ("GET ", format!("{}/:id", svc.path), format!("get {name}")),
|
||||
Operation::Create => ("POST ", svc.path.clone(), format!("create {name}")),
|
||||
Operation::Update => ("PATCH ", format!("{}/:id", svc.path), format!("update {name}")),
|
||||
Operation::Delete => ("DELETE", format!("{}/:id", svc.path), format!("delete {name}")),
|
||||
Operation::Subscribe => ("WS ", format!("{}/live", svc.path), format!("subscribe {name} (Stage 3)")),
|
||||
Operation::Me => ("GET ", format!("{}/me", svc.path), "(Stage 3)".into()),
|
||||
Operation::List => ("GET ", svc.path.clone(), format!("list {name}")),
|
||||
Operation::Get => ("GET ", format!("{}/:id", svc.path), format!("get {name}")),
|
||||
Operation::Create => ("POST ", svc.path.clone(), format!("create {name}")),
|
||||
Operation::Update => ("PATCH ", format!("{}/:id", svc.path), format!("update {name}")),
|
||||
Operation::Delete => ("DELETE", format!("{}/:id", svc.path), format!("delete {name}")),
|
||||
Operation::Subscribe => ("WS ", format!("{}/live", svc.path), format!("subscribe {name} (Stage 3)")),
|
||||
Operation::Me => ("GET ", format!("{}/me", svc.path), "(Stage 3)".into()),
|
||||
Operation::Custom => continue,
|
||||
};
|
||||
out.push_str(&format!(" {m} {p:<30} {n}\n"));
|
||||
|
|
@ -215,3 +212,79 @@ pub fn describe_routes(engine: &Engine) -> String {
|
|||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::compile::Catalog;
|
||||
use crate::parser::parse;
|
||||
|
||||
fn build(src: &str) -> (Shared, Router) {
|
||||
let cat = Catalog::from_schemas(vec![parse(src).unwrap()]).unwrap();
|
||||
let eng = Arc::new(Mutex::new(Engine::new(cat.clone())));
|
||||
let r = router(eng.clone(), &cat);
|
||||
(eng, r)
|
||||
}
|
||||
|
||||
fn req(method: Method, path: &str, body: &[u8]) -> Request {
|
||||
Request {
|
||||
method,
|
||||
path: path.into(),
|
||||
query: None,
|
||||
headers: Default::default(),
|
||||
body: body.to_vec(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn router_serves_crud_for_a_service_rest_block() {
|
||||
let (_eng, r) = build(r#"
|
||||
type Article { id: Id
|
||||
title: Text
|
||||
service rest "/api/articles" expose list, get, create, update, delete }
|
||||
"#);
|
||||
|
||||
// empty list
|
||||
let resp = r.dispatch(&req(Method::Get, "/api/articles", b""));
|
||||
assert_eq!(resp.status.0, 200);
|
||||
assert_eq!(resp.body, b"[]");
|
||||
|
||||
// create
|
||||
let resp = r.dispatch(&req(Method::Post, "/api/articles", br#"{"title":"hello"}"#));
|
||||
assert_eq!(resp.status.0, 201);
|
||||
|
||||
// get by id
|
||||
let resp = r.dispatch(&req(Method::Get, "/api/articles/1", b""));
|
||||
assert_eq!(resp.status.0, 200);
|
||||
|
||||
// delete
|
||||
let resp = r.dispatch(&req(Method::Delete, "/api/articles/1", b""));
|
||||
assert_eq!(resp.status.0, 204);
|
||||
|
||||
// 404 after delete
|
||||
let resp = r.dispatch(&req(Method::Get, "/api/articles/1", b""));
|
||||
assert_eq!(resp.status.0, 404);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unexposed_method_yields_405() {
|
||||
let (_eng, r) = build(r#"
|
||||
type Tag { id: Id
|
||||
label: Text
|
||||
service rest "/api/tags" expose list, get }
|
||||
"#);
|
||||
let resp = r.dispatch(&req(Method::Post, "/api/tags", br#"{"label":"rust"}"#));
|
||||
assert_eq!(resp.status.0, 405);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn live_subscribe_is_501() {
|
||||
let (_eng, r) = build(r#"
|
||||
type Article { id: Id
|
||||
title: Text
|
||||
service rest "/api/articles" expose list, subscribe }
|
||||
"#);
|
||||
let resp = r.dispatch(&req(Method::Get, "/api/articles/live", b""));
|
||||
assert_eq!(resp.status.0, 501);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue