use std::{
any::{Any, TypeId},
collections::HashMap,
fs,
net::SocketAddr,
sync::{Arc, RwLock},
};
use anyhow::Result;
use tower_http::{cors::CorsLayer, trace::TraceLayer};
use tower_http::add_extension::AddExtensionLayer;
use tower::{Layer, Service};
use axum::body::Body;
use axum::http::{Request, Response};
use futures::future::BoxFuture;
use axum::response::IntoResponse;
use serde::de::DeserializeOwned;
use tracing_subscriber::{fmt, EnvFilter};
use inventory;
use std::any::TypeId as StdTypeId;
use once_cell::sync::OnceCell;
use std::time::{Duration, Instant};
use axum::response::Html;
use regex::Regex;
#[cfg(feature = "devtools")]
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
#[cfg(feature = "devtools")]
use tokio::sync::broadcast;
#[cfg(feature = "devtools")]
use tokio_stream::wrappers::BroadcastStream;
#[cfg(feature = "devtools")]
use tokio_stream::StreamExt;
#[cfg(feature = "devtools")]
use notify::{Config as NotifyConfig, RecommendedWatcher, RecursiveMode};
#[cfg(feature = "devtools")]
use notify::Watcher;
#[cfg(feature = "swagger")]
use utoipa::openapi::{OpenApiBuilder, InfoBuilder, PathsBuilder};
#[cfg(feature = "swagger")]
static OPENAPI: OnceCell<utoipa::openapi::OpenApi> = OnceCell::new();
#[cfg(feature = "swagger")]
static OPENAPI_JSON: OnceCell<serde_json::Value> = OnceCell::new();
pub use axum::Router;
pub use axum::routing::{delete, get, post, put};
pub use axum::extract::Path;
pub use axum::extract::Json;
pub use axum::extract::rejection::JsonRejection;
pub use axum::http::StatusCode;
pub use axum::body::Body as AxumBody;
pub use axum::http::{Request as AxumRequest, Response as AxumResponse};
pub use spring_axum_macros::*;
#[macro_export]
macro_rules! mybatis_init {
($dir:expr) => {{
if let Err(e) = ::spring_axum_mybatis::init_global_from_dir_once($dir) {
::tracing::error!(error = %e, "failed to init mybatis from dir");
panic!("MyBatis init failed: {}", e);
}
}};
}
#[macro_export]
macro_rules! mybatis_registry {
() => {{
::spring_axum_mybatis::global_registry()
.expect("MyBatis global registry not initialized; call mybatis_init!(...) early")
}};
}
#[macro_export]
macro_rules! mybatis_exec {
($executor:expr, $stmt_id:expr, $params_json:expr) => {{
let reg = $crate::mybatis_registry!();
let (sql, ordered) = reg
.prepare_json($stmt_id, &$params_json)
.ok_or_else(|| $crate::AppError::Internal("statement not found".into()))?;
$executor
.execute(&sql, &ordered)
.map_err(|e| $crate::AppError::Internal(format!("execute error: {}", e)))?;
}};
}
#[macro_export]
macro_rules! sql_method {
($fn_name:ident ( $($arg:ident : $typ:ty),* $(,)? ) ; namespace = $ns:literal) => {
pub fn $fn_name(&self, $($arg: $typ),*) -> ::spring_axum::AppResult<()> {
let params = ::serde_json::json!({ $(stringify!($arg): $arg),* });
let exec = ::spring_axum_mybatis::NoopExecutor::default();
::spring_axum::mybatis_exec!(exec, concat!($ns, ".", stringify!($fn_name)), params);
Ok(())
}
};
}
#[macro_export]
macro_rules! auto_discover {
() => {{
::spring_axum::SpringApp::new()
.with_discovered_components()
.with_discovered_interceptors()
.with_discovered_advices()
.with_discovered_controllers()
}};
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct ErrorResponse { pub error: String }
#[derive(Debug, Clone, serde::Serialize)]
pub struct ValidationErrorsBody {
pub errors: HashMap<String, Vec<String>>,
}
#[derive(Debug, Clone)]
pub enum AppError {
BadRequest(String),
NotFound(String),
Internal(String),
Validation(ValidationErrorsBody),
Unauthorized(String),
Forbidden(String),
}
pub type AppResult<T> = std::result::Result<T, AppError>;
impl IntoResponse for AppError {
fn into_response(self) -> AxumResponse<AxumBody> {
match self {
AppError::BadRequest(msg) => (StatusCode::BAD_REQUEST, axum::Json(ErrorResponse { error: msg })).into_response(),
AppError::NotFound(msg) => (StatusCode::NOT_FOUND, axum::Json(ErrorResponse { error: msg })).into_response(),
AppError::Internal(msg) => (StatusCode::INTERNAL_SERVER_ERROR, axum::Json(ErrorResponse { error: msg })).into_response(),
AppError::Validation(body) => (StatusCode::BAD_REQUEST, axum::Json(body)).into_response(),
AppError::Unauthorized(msg) => (StatusCode::UNAUTHORIZED, axum::Json(ErrorResponse { error: msg })).into_response(),
AppError::Forbidden(msg) => (StatusCode::FORBIDDEN, axum::Json(ErrorResponse { error: msg })).into_response(),
}
}
}
impl From<JsonRejection> for AppError {
fn from(err: JsonRejection) -> Self { AppError::BadRequest(err.to_string()) }
}
#[cfg(feature = "validator")]
impl From<validator::ValidationErrors> for AppError {
fn from(errs: validator::ValidationErrors) -> Self {
let mut map: HashMap<String, Vec<String>> = HashMap::new();
for (field, field_errs) in errs.field_errors() {
let messages: Vec<String> = field_errs
.iter()
.map(|e| {
if let Some(msg) = e.message.as_ref() {
msg.to_string()
} else {
e.code.to_string()
}
})
.collect();
map.insert(field.to_string(), messages);
}
AppError::Validation(ValidationErrorsBody { errors: map })
}
}
#[cfg(feature = "validator")]
pub struct ValidatedJson<T>(pub T);
#[cfg(feature = "validator")]
impl<T, S> axum::extract::FromRequest<S> for ValidatedJson<T>
where
T: DeserializeOwned + validator::Validate,
S: Send + Sync,
{
type Rejection = AppError;
fn from_request(req: AxumRequest<AxumBody>, state: &S) -> impl futures::Future<Output = Result<Self, Self::Rejection>> + Send {
let fut = axum::extract::Json::<T>::from_request(req, state);
async move {
match fut.await {
Ok(axum::extract::Json(v)) => {
if let Err(e) = v.validate() { return Err(AppError::from(e)); }
Ok(ValidatedJson(v))
}
Err(rej) => Err(AppError::from(rej)),
}
}
}
}
#[cfg(feature = "validator")]
pub struct ValidatedQuery<T>(pub T);
#[cfg(feature = "validator")]
impl<T, S> axum::extract::FromRequestParts<S> for ValidatedQuery<T>
where
T: DeserializeOwned + validator::Validate,
S: Send + Sync,
{
type Rejection = AppError;
fn from_request_parts(parts: &mut axum::http::request::Parts, state: &S) -> impl futures::Future<Output = Result<Self, Self::Rejection>> + Send {
let fut = axum::extract::Query::<T>::from_request_parts(parts, state);
async move {
match fut.await {
Ok(axum::extract::Query(v)) => {
if let Err(e) = v.validate() { return Err(AppError::from(e)); }
Ok(ValidatedQuery(v))
}
Err(rej) => Err(AppError::BadRequest(rej.to_string())),
}
}
}
}
#[cfg(feature = "validator")]
pub struct ValidatedForm<T>(pub T);
#[cfg(feature = "validator")]
impl<T, S> axum::extract::FromRequest<S> for ValidatedForm<T>
where
T: DeserializeOwned + validator::Validate,
S: Send + Sync,
{
type Rejection = AppError;
fn from_request(req: AxumRequest<AxumBody>, state: &S) -> impl futures::Future<Output = Result<Self, Self::Rejection>> + Send {
let fut = axum::extract::Form::<T>::from_request(req, state);
async move {
match fut.await {
Ok(axum::extract::Form(v)) => {
if let Err(e) = v.validate() { return Err(AppError::from(e)); }
Ok(ValidatedForm(v))
}
Err(rej) => Err(AppError::BadRequest(rej.to_string())),
}
}
}
}
#[cfg(all(feature = "validator", feature = "axum_extra_json"))]
pub struct ValidatedJsonStream<T>(pub T);
#[cfg(all(feature = "validator", feature = "axum_extra_json"))]
impl<T, S> axum::extract::FromRequest<S> for ValidatedJsonStream<T>
where
T: DeserializeOwned + validator::Validate + 'static,
S: Send + Sync,
{
type Rejection = AppError;
fn from_request(req: AxumRequest<AxumBody>, state: &S) -> impl futures::Future<Output = Result<Self, Self::Rejection>> + Send {
let fut = axum_extra::extract::JsonDeserializer::<T>::from_request(req, state);
async move {
match fut.await {
Ok(deserializer) => match deserializer.deserialize() {
Ok(v) => {
if let Err(e) = v.validate() { return Err(AppError::from(e)); }
Ok(ValidatedJsonStream(v))
}
Err(rej) => Err(AppError::BadRequest(rej.to_string())),
},
Err(rej) => Err(AppError::BadRequest(rej.to_string())),
}
}
}
}
#[derive(Clone, serde::Deserialize, serde::Serialize, Debug)]
pub struct ServerConfig {
pub host: String,
pub port: u16,
#[serde(default)]
pub enable_cors: bool,
}
impl Default for ServerConfig {
fn default() -> Self {
Self {
host: "127.0.0.1".into(),
port: 3000,
enable_cors: true,
}
}
}
#[derive(Clone, serde::Deserialize, serde::Serialize, Debug, Default)]
pub struct AppConfig {
#[serde(default)]
pub server: ServerConfig,
#[serde(default)]
pub devtools: DevtoolsConfig,
#[serde(default)]
pub tx_advice: TxAdviceConfig,
}
#[derive(Clone, serde::Deserialize, serde::Serialize, Debug)]
pub struct DevtoolsConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default)]
pub watch: Vec<String>,
}
impl Default for DevtoolsConfig {
fn default() -> Self {
Self { enabled: false, watch: vec!["src".into(), "resources".into(), "templates".into()] }
}
}
#[derive(Clone, serde::Deserialize, serde::Serialize, Debug, Default)]
pub struct TxAdviceConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default)]
pub pointcut: Option<String>,
}
fn load_config() -> AppConfig {
let base_path = "application.yaml";
let mut config = match fs::read_to_string(base_path) {
Ok(s) => serde_yaml::from_str::<AppConfig>(&s).unwrap_or_default(),
Err(_) => AppConfig::default(),
};
if let Ok(profile) = std::env::var("SPRING_PROFILE") {
let prof_path = format!("application-{}.yaml", profile);
if let Ok(s) = fs::read_to_string(&prof_path) {
if let Ok(overlay) = serde_yaml::from_str::<AppConfig>(&s) {
config.server = overlay.server;
}
}
}
if let Ok(host) = std::env::var("SERVER_HOST") {
config.server.host = host;
}
if let Ok(port) = std::env::var("SERVER_PORT") {
if let Ok(p) = port.parse::<u16>() {
config.server.port = p;
}
}
if let Ok(cors) = std::env::var("SERVER_ENABLE_CORS") {
config.server.enable_cors = matches!(cors.as_str(), "1" | "true" | "True" | "TRUE");
}
if let Ok(enabled) = std::env::var("DEVTOOLS_ENABLED") {
config.devtools.enabled = matches!(enabled.as_str(), "1" | "true" | "True" | "TRUE");
}
if let Ok(watch) = std::env::var("DEVTOOLS_WATCH") {
let list: Vec<String> = watch
.split([',', ';', '|'])
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
if !list.is_empty() { config.devtools.watch = list; }
}
config
}
fn validate_config(cfg: &AppConfig) {
if cfg.server.host.trim().is_empty() {
tracing::warn!(target = "spring_axum", "server.host is empty; defaulting may cause bind errors");
}
if cfg.server.port == 0 {
tracing::warn!(target = "spring_axum", port = cfg.server.port, "server.port must be in 1..=65535");
}
if cfg.server.host.starts_with("http://") || cfg.server.host.starts_with("https://") {
tracing::warn!(target = "spring_axum", host = %cfg.server.host, "server.host should be a hostname/ip, not URL");
}
#[cfg(feature = "devtools")]
if cfg.devtools.enabled {
for p in &cfg.devtools.watch {
if !std::path::Path::new(p).exists() {
tracing::info!(target = "spring_axum", path = %p, "devtools watch path not found (will still attempt watching)");
}
}
}
}
pub trait Controller {
fn routes(&self) -> Router;
}
pub struct SpringApp {
router: Router,
config: AppConfig,
ctx: Arc<ApplicationContext>,
}
impl SpringApp {
pub fn new() -> Self {
let config = load_config();
validate_config(&config);
let router = Router::new().route("/actuator/health", get(|| async { "OK" }));
let ctx = Arc::new(ApplicationContext::new());
init_defaults_once();
let router = router.merge(cache_metrics_router());
#[cfg(feature = "swagger")]
let router = router.merge(swagger_routes());
#[cfg(feature = "devtools")]
let router = if config.devtools.enabled { router.merge(devtools_router()) } else { router };
#[cfg(feature = "devtools")]
if config.devtools.enabled { start_devtools_watcher(&config); }
Self { router, config, ctx }
}
pub fn with_controller(mut self, controller: impl Controller) -> Self {
self.router = self.router.merge(controller.routes());
self
}
pub fn with_router(mut self, router: Router) -> Self {
self.router = self.router.merge(router);
self
}
pub async fn run(self) -> Result<()> {
init_tracing();
auto_init_integrations();
print_banner_if_any();
tracing::info!(target = "spring_axum", host = %self.config.server.host, port = self.config.server.port, cors = self.config.server.enable_cors, devtools = self.config.devtools.enabled, "Startup configuration");
publish_event(AppStarting { config: self.config.clone() }, &self.ctx);
let mut router = self
.router
.layer(TraceLayer::new_for_http())
.layer(AddExtensionLayer::new(self.ctx.clone()));
if self.config.server.enable_cors {
router = router.layer(CorsLayer::permissive());
}
let addr: SocketAddr = format!("{}:{}", self.config.server.host, self.config.server.port)
.parse()
.expect("invalid server address");
tracing::info!(target: "spring_axum", "Listening on http://{}", addr);
axum::serve(tokio::net::TcpListener::bind(addr).await?, router).await?;
Ok(())
}
}
fn init_tracing() {
let _ = fmt()
.with_env_filter(
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
)
.with_target(false)
.compact()
.try_init();
}
pub struct ApplicationContext {
inner: RwLock<HashMap<TypeId, Box<dyn Any + Send + Sync>>>,
}
impl ApplicationContext {
pub fn new() -> Self {
Self { inner: RwLock::new(HashMap::new()) }
}
pub fn register<T: Send + Sync + 'static>(&self, value: T) {
let arc: Arc<T> = Arc::new(value);
self.inner
.write()
.unwrap()
.insert(TypeId::of::<T>(), Box::new(arc));
}
pub fn register_dynamic(&self, type_id: StdTypeId, arc_any: Box<dyn Any + Send + Sync>) {
self.inner.write().unwrap().insert(type_id, arc_any);
}
pub fn resolve<T: Send + Sync + 'static>(&self) -> Option<Arc<T>> {
self.inner
.read()
.unwrap()
.get(&TypeId::of::<T>())
.and_then(|b| b.downcast_ref::<Arc<T>>().cloned())
}
}
pub trait TransactionManager: Send + Sync + 'static {
fn begin(&self) -> BoxFuture<'static, Result<(), AppError>>;
fn commit(&self) -> BoxFuture<'static, Result<(), AppError>>;
fn rollback(&self) -> BoxFuture<'static, Result<(), AppError>>;
}
#[derive(Clone, Default)]
pub struct NoopTransactionManager;
impl TransactionManager for NoopTransactionManager {
fn begin(&self) -> BoxFuture<'static, Result<(), AppError>> { Box::pin(async { Ok(()) }) }
fn commit(&self) -> BoxFuture<'static, Result<(), AppError>> { Box::pin(async { Ok(()) }) }
fn rollback(&self) -> BoxFuture<'static, Result<(), AppError>> { Box::pin(async { Ok(()) }) }
}
static GLOBAL_TX_MANAGER: OnceCell<Arc<dyn TransactionManager>> = OnceCell::new();
fn init_defaults_once() {
let _ = GLOBAL_TX_MANAGER.set(Arc::new(NoopTransactionManager::default()));
let _ = GLOBAL_CACHE.set(InMemoryCache::default());
}
fn auto_init_integrations() {
let mybatis_dir = std::path::Path::new("resources/mybatis");
if mybatis_dir.exists() {
if spring_axum_mybatis::global_registry().is_none() {
match spring_axum_mybatis::init_global_from_dir_once("resources/mybatis") {
Ok(_) => tracing::info!(target = "spring_axum", path = %mybatis_dir.display(), "MyBatis registry initialized"),
Err(e) => tracing::warn!(target = "spring_axum", error = %e, "Failed to init MyBatis registry"),
}
}
} else {
tracing::debug!(target = "spring_axum", "resources/mybatis not found; skipping MyBatis init");
}
#[cfg(feature = "sqlx_postgres")]
{
if GLOBAL_SQLX_POOL.get().is_none() {
match std::env::var("DATABASE_URL") {
Ok(url) => {
match sqlx::postgres::PgPoolOptions::new().connect_lazy(&url) {
Ok(pool) => {
let _ = GLOBAL_SQLX_POOL.set(pool);
tracing::info!(target = "spring_axum", "sqlx pool initialized from env DATABASE_URL");
}
Err(e) => tracing::warn!(target = "spring_axum", error = %e, "Failed to lazy connect sqlx pool"),
}
}
Err(_) => tracing::debug!(target = "spring_axum", "DATABASE_URL not set; skipping sqlx pool init"),
}
}
}
}
pub fn set_transaction_manager<T: TransactionManager>(mgr: Arc<T>) {
let _ = GLOBAL_TX_MANAGER.set(mgr);
}
pub async fn transaction<F, Fut, T>(f: F) -> AppResult<T>
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = AppResult<T>>,
{
let mgr = GLOBAL_TX_MANAGER.get().cloned().unwrap_or_else(|| Arc::new(NoopTransactionManager));
if let Err(e) = mgr.begin().await { return Err(e); }
match f().await {
Ok(val) => {
if let Err(e) = mgr.commit().await { return Err(e); }
Ok(val)
}
Err(err) => {
let _ = mgr.rollback().await;
Err(err)
}
}
}
#[derive(Clone, Default)]
pub struct InMemoryCache {
inner: Arc<RwLock<HashMap<String, (Box<dyn Any + Send + Sync>, Instant, Option<Duration>, StdTypeId)>>>,
stats: Arc<RwLock<CacheStats>>,
}
#[derive(Default, Clone, serde::Serialize)]
pub struct CacheStats {
hits: u64,
misses: u64,
puts: u64,
evicts: u64,
}
impl InMemoryCache {
pub fn get_typed<T: Clone + Send + Sync + 'static>(&self, key: &str) -> Option<T> {
let mut guard = self.inner.write().unwrap();
if let Some((boxed_any, inserted, ttl_opt, type_id)) = guard.get(key) {
let expired = match ttl_opt {
Some(ttl) => *inserted + *ttl < Instant::now(),
None => false,
};
if expired {
guard.remove(key);
self.stats.write().unwrap().misses += 1;
return None;
}
if *type_id == StdTypeId::of::<T>() {
if let Some(arc_val) = (&**boxed_any as &dyn Any).downcast_ref::<Arc<T>>() {
self.stats.write().unwrap().hits += 1;
return Some((**arc_val).clone());
}
}
self.stats.write().unwrap().misses += 1;
None
} else {
self.stats.write().unwrap().misses += 1;
None
}
}
pub fn put_typed<T: Clone + Send + Sync + 'static>(&self, key: String, value: T, ttl: Option<Duration>) {
let boxed: Box<dyn Any + Send + Sync> = Box::new(Arc::new(value));
self.inner
.write()
.unwrap()
.insert(key, (boxed, Instant::now(), ttl, StdTypeId::of::<T>()));
self.stats.write().unwrap().puts += 1;
}
pub fn evict(&self, key: &str) {
self.inner.write().unwrap().remove(key);
self.stats.write().unwrap().evicts += 1;
}
pub fn stats(&self) -> CacheStats { self.stats.read().unwrap().clone() }
}
static GLOBAL_CACHE: OnceCell<InMemoryCache> = OnceCell::new();
pub fn cache_instance() -> InMemoryCache {
GLOBAL_CACHE.get().cloned().unwrap_or_default()
}
pub fn default_cache_key(fn_name: &str, args_json: &serde_json::Value) -> String {
format!("{}:{}", fn_name, args_json)
}
fn cache_metrics_router() -> Router {
let handler = || async move {
let stats = cache_instance().stats();
axum::Json(stats)
};
Router::new().route("/actuator/metrics/cache", get(handler))
}
fn print_banner_if_any() {
if let Ok(banner) = fs::read_to_string("banner.txt") {
println!("{}", banner);
} else {
println!(":: Spring-Axum v{} ::", env!("CARGO_PKG_VERSION"));
let mut features: Vec<&str> = Vec::new();
if cfg!(feature = "validator") { features.push("validator"); }
if cfg!(feature = "swagger") { features.push("swagger"); }
if cfg!(feature = "sqlx_postgres") { features.push("sqlx_postgres"); }
if cfg!(feature = "devtools") { features.push("devtools"); }
if !features.is_empty() {
println!("Enabled features: {}", features.join(", "));
}
}
}
#[cfg(feature = "devtools")]
static DEVTOOLS_BUS: OnceCell<broadcast::Sender<String>> = OnceCell::new();
#[cfg(feature = "devtools")]
fn devtools_router() -> Router {
Router::new()
.route("/devtools/events", get(devtools_sse))
.route("/devtools/trigger", get(devtools_trigger))
}
#[cfg(feature = "devtools")]
async fn devtools_sse() -> Sse<impl futures::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
let tx = DEVTOOLS_BUS.get_or_init(|| broadcast::channel(128).0).clone();
let rx = tx.subscribe();
let stream = BroadcastStream::new(rx).filter_map(|res| match res {
Ok(msg) => Some(Ok(SseEvent::default().event("reload").data(msg))),
Err(_) => None,
});
Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(10)).text("keep-alive"))
}
#[cfg(feature = "devtools")]
async fn devtools_trigger() -> &'static str {
let _ = DEVTOOLS_BUS.get_or_init(|| broadcast::channel(128).0).send("manual".into());
"ok"
}
#[cfg(feature = "devtools")]
fn start_devtools_watcher(cfg: &AppConfig) {
if !cfg.devtools.enabled { return; }
let tx = DEVTOOLS_BUS.get_or_init(|| broadcast::channel(128).0).clone();
let paths = cfg.devtools.watch.clone();
std::thread::spawn(move || {
use std::sync::mpsc::channel;
let (evt_tx, evt_rx) = channel();
let mut watcher = RecommendedWatcher::new(move |res| {
let _ = evt_tx.send(res);
}, NotifyConfig::default()).expect("Failed to init file watcher");
for p in &paths {
let _ = watcher.watch(std::path::Path::new(p), RecursiveMode::Recursive);
}
while let Ok(res) = evt_rx.recv() {
match res {
Ok(event) => {
let _ = tx.send("reload".into());
let _ = tx.send(format!("changed: {:?}", event.paths));
}
Err(e) => {
let _ = tx.send(format!("watch_error: {}", e));
}
}
}
});
}
pub struct Component<T>(pub Arc<T>);
impl<S, T> axum::extract::FromRequestParts<S> for Component<T>
where
T: Send + Sync + 'static,
{
type Rejection = (axum::http::StatusCode, String);
fn from_request_parts(
parts: &mut axum::http::request::Parts,
_state: &S,
) -> impl futures::Future<Output = Result<Self, Self::Rejection>> + Send {
let ctx_opt = parts.extensions.get::<Arc<ApplicationContext>>().cloned();
async move {
if let Some(ctx) = ctx_opt {
if let Some(v) = ctx.resolve::<T>() {
return Ok(Component(v));
}
}
Err((axum::http::StatusCode::INTERNAL_SERVER_ERROR, "Component not found".into()))
}
}
}
impl SpringApp {
pub fn with_component<T: Send + Sync + 'static>(self, value: T) -> Self {
self.ctx.register::<T>(value);
self
}
pub fn with_discovered_components(self) -> Self {
for reg in inventory::iter::<ComponentRegistration> {
let (type_id, arc_any) = (reg.init)(&self.ctx);
self.ctx.register_dynamic(type_id, arc_any);
}
self
}
}
pub struct ComponentRegistration {
pub init: fn(&ApplicationContext) -> (StdTypeId, Box<dyn Any + Send + Sync>),
}
inventory::collect!(ComponentRegistration);
pub struct ControllerRouterRegistration {
pub init: fn() -> Router,
}
inventory::collect!(ControllerRouterRegistration);
impl SpringApp {
pub fn with_discovered_controllers(mut self) -> Self {
for reg in inventory::iter::<ControllerRouterRegistration> {
let router = (reg.init)();
self.router = self.router.merge(router);
}
self
}
}
pub trait Interceptor: Clone + Send + Sync + 'static {
fn on_request(&self, req: Request<Body>) -> Request<Body> { req }
fn on_response(&self, res: Response<Body>) -> Response<Body> { res }
}
pub struct InterceptorRegistration {
pub apply: fn(Router) -> Router,
}
inventory::collect!(InterceptorRegistration);
impl SpringApp {
pub fn with_discovered_interceptors(mut self) -> Self {
for reg in inventory::iter::<InterceptorRegistration> {
self.router = (reg.apply)(self.router);
}
self
}
}
#[derive(Clone)]
pub struct InterceptorLayer<I> {
interceptor: I,
}
impl<I> InterceptorLayer<I> {
pub fn new(interceptor: I) -> Self { Self { interceptor } }
}
impl<S, I> Layer<S> for InterceptorLayer<I>
where
I: Interceptor,
{
type Service = InterceptorService<S, I>;
fn layer(&self, inner: S) -> Self::Service { InterceptorService { inner, interceptor: self.interceptor.clone() } }
}
#[derive(Clone)]
pub struct InterceptorService<S, I> {
inner: S,
interceptor: I,
}
impl<S, I> Service<Request<Body>> for InterceptorService<S, I>
where
S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
S::Future: Send + 'static,
I: Interceptor,
{
type Response = Response<Body>;
type Error = S::Error;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, req: Request<Body>) -> Self::Future {
let interceptor = self.interceptor.clone();
let mut inner = self.inner.clone();
let req = interceptor.on_request(req);
Box::pin(async move {
let res = inner.call(req).await?;
Ok(interceptor.on_response(res))
})
}
}
impl SpringApp {
pub fn with_interceptor<I: Interceptor>(mut self, interceptor: I) -> Self {
self.router = self.router.layer(InterceptorLayer::new(interceptor));
self
}
}
#[derive(Clone)]
pub struct Pointcut {
methods: Option<Vec<axum::http::Method>>,
path_re: Option<Regex>,
}
impl Pointcut {
pub fn matches(&self, req: &Request<Body>) -> bool {
let method_ok = match &self.methods {
Some(ms) => ms.iter().any(|m| m == req.method()),
None => true,
};
let path_ok = match &self.path_re {
Some(re) => re.is_match(req.uri().path()),
None => true,
};
method_ok && path_ok
}
}
fn glob_to_regex(glob: &str) -> String {
let mut s = String::from("^");
let mut i = 0;
let bytes = glob.as_bytes();
while i < bytes.len() {
match bytes[i] as char {
'*' => {
if i + 1 < bytes.len() && bytes[i + 1] as char == '*' {
s.push_str(".*");
i += 2;
} else {
s.push_str("[^/]*");
i += 1;
}
}
'.' | '+' | '?' | '(' | ')' | '|' | '{' | '}' | '[' | ']' | '^' | '$' => {
s.push('\\');
s.push(bytes[i] as char);
i += 1;
}
c => {
s.push(c);
i += 1;
}
}
}
s.push('$');
s
}
fn parse_pointcut(expr: &str) -> Pointcut {
let parts: Vec<&str> = expr.split_whitespace().collect();
if parts.is_empty() {
return Pointcut { methods: None, path_re: None };
}
if parts.len() == 1 {
let re = Regex::new(&glob_to_regex(parts[0])).ok();
return Pointcut { methods: None, path_re: re };
}
let methods: Option<Vec<axum::http::Method>> = {
let ms = parts[0]
.split(',')
.filter_map(|m| axum::http::Method::from_bytes(m.trim().as_bytes()).ok())
.collect::<Vec<_>>();
if ms.is_empty() { None } else { Some(ms) }
};
let path_re = Regex::new(&glob_to_regex(parts[1])).ok();
Pointcut { methods, path_re }
}
pub trait Advice: Send + Sync + 'static {
fn before(&self, req: Request<Body>) -> Request<Body> { req }
fn after(&self, res: Response<Body>) -> Response<Body> { res }
fn before_async(&self, req: Request<Body>) -> futures::future::BoxFuture<'static, Request<Body>> {
Box::pin(async move { req })
}
fn after_async(&self, res: Response<Body>) -> futures::future::BoxFuture<'static, Response<Body>> {
Box::pin(async move { res })
}
fn pointcut_expr(&self) -> &'static str;
fn pointcut(&self) -> Pointcut { parse_pointcut(self.pointcut_expr()) }
}
#[derive(Clone)]
pub struct AdviceLayer {
advice: Arc<dyn Advice>,
pointcut: Pointcut,
}
impl AdviceLayer {
pub fn new(advice: Box<dyn Advice>) -> Self { Self { pointcut: advice.pointcut(), advice: Arc::from(advice) } }
}
#[derive(Clone)]
pub struct AdviceService<S> {
inner: S,
advice: Arc<dyn Advice>,
pointcut: Pointcut,
}
impl<S> Layer<S> for AdviceLayer {
type Service = AdviceService<S>;
fn layer(&self, inner: S) -> Self::Service { AdviceService { inner, advice: self.advice.clone(), pointcut: self.pointcut.clone() } }
}
impl<S> Service<Request<Body>> for AdviceService<S>
where
S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
S::Future: Send + 'static,
{
type Response = Response<Body>;
type Error = S::Error;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> { self.inner.poll_ready(cx) }
fn call(&mut self, req: Request<Body>) -> Self::Future {
let mut inner = self.inner.clone();
let advice = self.advice.clone();
let pointcut = self.pointcut.clone();
let matched = pointcut.matches(&req);
let req = if matched { (*advice).before(req) } else { req };
Box::pin(async move {
let req2 = if matched { (*advice).before_async(req).await } else { req };
let res = inner.call(req2).await?;
let res2 = if matched { (*advice).after(res) } else { res };
let res3 = if matched { (*advice).after_async(res2).await } else { res2 };
Ok(res3)
})
}
}
pub struct AnyAdviceRegistration {
pub build_boxed: fn() -> Box<dyn Advice>,
}
inventory::collect!(AnyAdviceRegistration);
impl SpringApp {
pub fn with_discovered_advices(mut self) -> Self {
for reg in inventory::iter::<AnyAdviceRegistration> {
let boxed = (reg.build_boxed)();
self.router = self.router.layer(AdviceLayer::new(boxed));
}
if self.config.tx_advice.enabled {
let expr = self
.config
.tx_advice
.pointcut
.as_deref()
.unwrap_or("method:* path:/**");
let expr_static: &'static str = Box::leak(expr.to_string().into_boxed_str());
let boxed: Box<dyn Advice> = Box::new(TransactionAdvice::new(expr_static));
self.router = self.router.layer(AdviceLayer::new(boxed));
}
self
}
}
pub struct EventListenerRegistration {
pub matches: fn(StdTypeId) -> bool,
pub handle: fn(&dyn Any, &ApplicationContext),
}
inventory::collect!(EventListenerRegistration);
pub fn publish_event<T: Send + Sync + 'static>(evt: T, ctx: &ApplicationContext) {
let type_id = StdTypeId::of::<T>();
for reg in inventory::iter::<EventListenerRegistration> {
if (reg.matches)(type_id) {
(reg.handle)(&evt, ctx);
}
}
}
#[derive(Clone, Debug)]
pub struct AppStarting {
pub config: AppConfig,
}
#[macro_export]
macro_rules! event_listener {
($ty:ty, $handler:path) => {
inventory::submit! {
::spring_axum::EventListenerRegistration {
matches: |incoming: ::std::any::TypeId| incoming == ::std::any::TypeId::of::<$ty>(),
handle: |evt: &dyn ::std::any::Any, ctx: &::spring_axum::ApplicationContext| {
if let Some(v) = evt.downcast_ref::<$ty>() {
$handler(v, ctx);
}
},
}
}
};
}
#[macro_export]
macro_rules! advice {
($ty:ty) => {
inventory::submit! {
::spring_axum::AnyAdviceRegistration {
build_boxed: || Box::new(<$ty as Default>::default()),
}
}
};
}
#[derive(Clone)]
pub struct TransactionAdvice {
expr: &'static str,
}
impl TransactionAdvice {
pub const fn new(expr: &'static str) -> Self { Self { expr } }
}
impl Advice for TransactionAdvice {
fn pointcut_expr(&self) -> &'static str { self.expr }
fn before_async(&self, req: Request<Body>) -> futures::future::BoxFuture<'static, Request<Body>> {
Box::pin(async move {
if let Some(mgr) = GLOBAL_TX_MANAGER.get() {
let _ = mgr.begin().await;
}
req
})
}
fn after_async(&self, res: Response<Body>) -> futures::future::BoxFuture<'static, Response<Body>> {
Box::pin(async move {
if let Some(mgr) = GLOBAL_TX_MANAGER.get() {
if res.status().is_success() {
let _ = mgr.commit().await;
} else {
let _ = mgr.rollback().await;
}
}
res
})
}
}
#[macro_export]
macro_rules! transaction_advice {
($expr:literal) => {
inventory::submit! {
::spring_axum::AnyAdviceRegistration {
build_boxed: || Box::new(::spring_axum::TransactionAdvice::new($expr)),
}
}
};
}
#[cfg(feature = "swagger")]
fn build_openapi() -> utoipa::openapi::OpenApi {
OpenApiBuilder::new()
.info(InfoBuilder::new().title("Spring Axum").version("0.1.0").build())
.paths(PathsBuilder::new().build())
.build()
}
#[cfg(feature = "swagger")]
fn swagger_routes() -> Router {
fn ui_html() -> String {
r#"<!DOCTYPE html>
<html>
<head>
<meta charset=\"utf-8\" />
<title>Swagger UI</title>
</head>
<body>
<h1>Swagger UI (placeholder)</h1>
<p>OpenAPI JSON at <a href=\"/api-docs/openapi.json\">/api-docs/openapi.json</a></p>
</body>
</html>"#.to_string()
}
let openapi = OPENAPI.get_or_init(|| build_openapi()).clone();
let openapi_json = serde_json::to_value(&openapi).unwrap_or_else(|_| serde_json::json!({
"openapi": "3.0.0",
"info": {"title": "Spring Axum", "version": "0.1.0"},
"paths": {}
}));
let _ = OPENAPI_JSON.set(openapi_json);
Router::new()
.route("/swagger-ui", get(|| async { Html(ui_html()) }))
.route("/api-docs/openapi.json", get(|| async {
let v = OPENAPI_JSON
.get()
.cloned()
.unwrap_or_else(|| serde_json::json!({"openapi":"3.0.0","info":{"title":"Spring Axum","version":"0.1.0"},"paths":{}}));
axum::Json(v)
}))
}
#[cfg(feature = "sqlx_postgres")]
use sqlx::{Pool, Postgres, PgConnection};
#[cfg(feature = "sqlx_postgres")]
static GLOBAL_SQLX_POOL: OnceCell<Pool<Postgres>> = OnceCell::new();
#[cfg(feature = "sqlx_postgres")]
pub fn set_sqlx_pool(pool: Pool<Postgres>) {
let _ = GLOBAL_SQLX_POOL.set(pool);
}
#[cfg(feature = "sqlx_postgres")]
pub fn sqlx_pool() -> Option<Pool<Postgres>> { GLOBAL_SQLX_POOL.get().cloned() }
#[cfg(feature = "sqlx_postgres")]
pub async fn sqlx_transaction<F, Fut, T>(f: F) -> AppResult<T>
where
F: for<'a> FnOnce(&'a mut PgConnection) -> Fut,
Fut: std::future::Future<Output = AppResult<T>> + Send,
T: Send + 'static,
{
let pool = GLOBAL_SQLX_POOL
.get()
.cloned()
.ok_or_else(|| AppError::Internal("sqlx pool not set".into()))?;
let mut conn = pool
.acquire()
.await
.map_err(|e| AppError::Internal(format!("acquire conn error: {}", e)))?;
sqlx::query("BEGIN")
.execute(conn.as_mut())
.await
.map_err(|e| AppError::Internal(format!("begin tx error: {}", e)))?;
let result = f(conn.as_mut()).await;
match result {
Ok(val) => {
sqlx::query("COMMIT")
.execute(conn.as_mut())
.await
.map_err(|e| AppError::Internal(format!("commit error: {}", e)))?;
Ok(val)
}
Err(err) => {
let _ = sqlx::query("ROLLBACK").execute(conn.as_mut()).await;
Err(err)
}
}
}