refactor filter system

1 files changed, 15 insertions, 81 deletions

diff --git a/src/main.rs b/src/main.rs
index 0109a62..7c74f70 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -10,14 +10,10 @@ pub mod helper;
 #[cfg(feature = "mond")]
 pub mod reporting;
 
-use crate::{
-    config::{Config, RouteFilter},
-    filters::{files::serve_files, proxy::proxy_request},
-};
 use anyhow::{anyhow, Context, Result};
-use bytes::Bytes;
-use config::setup_file_watch;
+use config::{setup_file_watch, Config, NODE_KINDS};
 use error::ServiceError;
+use filters::{NodeContext, MODULES};
 use futures::future::try_join_all;
 use helper::TokioIo;
 use http_body_util::{combinators::BoxBody, BodyExt};
@@ -30,14 +26,11 @@ use hyper::{
     Request, Response, StatusCode,
 };
 use log::{debug, error, info, warn, LevelFilter};
-#[cfg(feature = "mond")]
-use reporting::Reporting;
 use rustls::pki_types::{CertificateDer, PrivateKeyDer};
 use std::{
     collections::HashMap,
     io::BufReader,
     net::SocketAddr,
-    ops::ControlFlow,
     path::{Path, PathBuf},
     process::exit,
     str::FromStr,
@@ -57,14 +50,7 @@ pub struct State {
     pub access_logs: RwLock<HashMap<String, BufWriter<File>>>,
     pub l_incoming: Semaphore,
     pub l_outgoing: Semaphore,
-    #[cfg(feature = "mond")]
-    pub reporting: Reporting,
 }
-pub struct HostState {}
-
-pub type FilterRequest = Request<Incoming>;
-pub type FilterResponseOut = Option<Response<BoxBody<Bytes, ServiceError>>>;
-pub type FilterResponse = Option<Response<BoxBody<Bytes, ServiceError>>>;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -73,6 +59,11 @@ async fn main() -> anyhow::Result<()> {
         .parse_env("LOG")
         .init();
 
+    NODE_KINDS
+        .write()
+        .unwrap()
+        .extend(MODULES.iter().map(|m| (m.name().to_owned(), *m)));
+
     let Some(config_path) = std::env::args().skip(1).next() else {
         eprintln!("error: first argument is expected to be the configuration file");
         exit(1)
@@ -249,70 +240,17 @@ fn load_private_key(path: &Path) -> anyhow::Result<PrivateKeyDer<'static>> {
 async fn service(
     state: Arc<State>,
     config: Arc<Config>,
-    req: Request<Incoming>,
+    request: Request<Incoming>,
     addr: SocketAddr,
 ) -> Result<hyper::Response<BoxBody<bytes::Bytes, ServiceError>>, ServiceError> {
-    debug!("{addr} ~> {:?} {}", req.headers().get(HOST), req.uri());
-    #[cfg(feature = "mond")]
-    state.reporting.request_in.inc();
-
-    let host = req
-        .headers()
-        .get(HOST)
-        .and_then(|e| e.to_str().ok())
-        .map(String::from)
-        .unwrap_or(String::from(""));
-    let host = remove_port(&host);
-    let route = config.hosts.get(host).ok_or(ServiceError::NoHost)?;
-    #[cfg(feature = "mond")]
-    state.reporting.hosts.get(host).unwrap().requests_in.inc();
-
-    // TODO this code is horrible
-    let mut req = Some(req);
-    let mut resp = None;
-    for filter in &route.0 {
-        let cf = match filter {
-            RouteFilter::Proxy { backend } => {
-                resp = Some(
-                    proxy_request(
-                        &state,
-                        req.take().ok_or(ServiceError::RequestTaken)?,
-                        addr,
-                        backend,
-                    )
-                    .await?,
-                );
-                ControlFlow::Continue(())
-            }
-            RouteFilter::Files { config } => {
-                resp = Some(
-                    serve_files(req.as_ref().ok_or(ServiceError::RequestTaken)?, config).await?,
-                );
-                ControlFlow::Continue(())
-            }
-            RouteFilter::HttpBasicAuth { config } => filters::auth::http_basic(
-                config,
-                req.as_ref().ok_or(ServiceError::RequestTaken)?,
-                &mut resp,
-            )?,
-            RouteFilter::AccessLog { config } => {
-                filters::accesslog::access_log(
-                    &state,
-                    host,
-                    addr,
-                    config,
-                    req.as_ref().ok_or(ServiceError::RequestTaken)?,
-                )
-                .await?
-            }
-        };
-        match cf {
-            ControlFlow::Continue(_) => continue,
-            ControlFlow::Break(_) => break,
-        }
-    }
+    debug!(
+        "{addr} ~> {:?} {}",
+        request.headers().get(HOST),
+        request.uri()
+    );
 
-    let mut resp = resp.ok_or(ServiceError::NoResponse)?;
+    let mut context = NodeContext { addr, state };
+    let mut resp = config.handler.handle(&mut context, request).await?;
 
     let server_header = resp.headers().get(SERVER).cloned();
     resp.headers_mut().insert(
@@ -327,7 +265,3 @@ async fn service(
 
     return Ok(resp);
 }
-
-pub fn remove_port(s: &str) -> &str {
-    s.split_once(":").map(|(s, _)| s).unwrap_or(s)
-}