refactor filter system

1 files changed, 167 insertions, 118 deletions

diff --git a/src/filters/files.rs b/src/filters/files.rs
index ee40d70..fc3d63b 100644
--- a/src/filters/files.rs
+++ b/src/filters/files.rs
@@ -1,8 +1,7 @@
-use crate::{
-    config::{CacheConfig, FileserverConfig},
-    ServiceError,
-};
+use super::{Node, NodeContext, NodeKind, NodeRequest, NodeResponse};
+use crate::{config::return_true, ServiceError};
 use bytes::{Bytes, BytesMut};
+use futures::Future;
 use futures_util::{future, future::Either, ready, stream, FutureExt, Stream, StreamExt};
 use headers::{
     AcceptRanges, CacheControl, ContentLength, ContentRange, ContentType, HeaderMapExt,
@@ -11,154 +10,204 @@ use headers::{
 use http_body_util::{combinators::BoxBody, BodyExt, StreamBody};
 use humansize::FormatSizeOptions;
 use hyper::{
-    body::{Frame, Incoming},
+    body::Frame,
     header::{CONTENT_TYPE, LOCATION},
     http::HeaderValue,
-    Request, Response, StatusCode,
+    Response, StatusCode,
 };
 use log::debug;
 use markup::Render;
 use percent_encoding::percent_decode_str;
-use std::{fs::Metadata, io, ops::Range, path::Path, pin::Pin, task::Poll};
+use serde::Deserialize;
+use serde_yaml::Value;
+use std::{
+    fs::Metadata,
+    io,
+    ops::Range,
+    path::{Path, PathBuf},
+    pin::Pin,
+    sync::Arc,
+    task::Poll,
+};
 use tokio::{
     fs::{read_to_string, File},
     io::AsyncSeekExt,
 };
 use tokio_util::io::poll_read_buf;
 
-pub async fn serve_files(
-    req: &Request<Incoming>,
-    config: &FileserverConfig,
-) -> Result<hyper::Response<BoxBody<Bytes, ServiceError>>, ServiceError> {
-    let rpath = req.uri().path();
+pub struct FilesKind;
 
-    let mut path = config.root.clone();
-    let mut user_path_depth = 0;
-    for seg in rpath.split("/") {
-        let seg = percent_decode_str(seg).decode_utf8()?;
+#[derive(Debug, Deserialize)]
+struct Files {
+    root: PathBuf,
+    #[serde(default)]
+    index: bool,
+    #[serde(default = "return_true")]
+    last_modified: bool,
+    // #[serde(default = "return_true")]
+    // etag: bool,
+    #[serde(default)]
+    cache: CacheMode,
+}
 
-        if seg == "" || seg == "." {
-            continue;
-        }
+#[derive(Debug, Default, Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum CacheMode {
+    #[default]
+    Public,
+    Private,
+    NoStore,
+}
 
-        if seg == ".." {
-            if user_path_depth <= 0 {
-                return Err(ServiceError::BadPath);
-            }
-            path.pop();
-            user_path_depth -= 1;
-        } else {
-            path.push(seg.as_ref());
-            user_path_depth += 1;
-        }
+impl NodeKind for FilesKind {
+    fn name(&self) -> &'static str {
+        "files"
     }
-    if !path.exists() {
-        return Err(ServiceError::NotFound);
+    fn instanciate(&self, config: Value) -> anyhow::Result<Arc<dyn Node>> {
+        Ok(Arc::new(serde_yaml::from_value::<Files>(config)?))
     }
+}
 
-    let metadata = path.metadata()?;
+impl Node for Files {
+    fn handle<'a>(
+        &'a self,
+        _context: &'a mut NodeContext,
+        request: NodeRequest,
+    ) -> Pin<Box<dyn Future<Output = Result<NodeResponse, ServiceError>> + Send + Sync + 'a>> {
+        Box::pin(async move {
+            let rpath = request.uri().path();
 
-    if metadata.file_type().is_dir() {
-        debug!("sending index for {path:?}");
-        if let Ok(indexhtml) = read_to_string(path.join("index.html")).await {
-            return Ok(html_string_response(indexhtml));
-        }
+            let mut path = self.root.clone();
+            let mut user_path_depth = 0;
+            for seg in rpath.split("/") {
+                let seg = percent_decode_str(seg).decode_utf8()?;
+
+                if seg == "" || seg == "." {
+                    continue;
+                }
 
-        if config.index {
-            if !rpath.ends_with("/") {
-                let mut r = Response::new(String::new());
-                *r.status_mut() = StatusCode::FOUND;
-                r.headers_mut().insert(
-                    LOCATION,
-                    HeaderValue::from_str(&format!("{}/", rpath))
-                        .map_err(|_| ServiceError::Other)?,
-                );
-                return Ok(r.map(|b| b.map_err(|e| match e {}).boxed()));
+                if seg == ".." {
+                    if user_path_depth <= 0 {
+                        return Err(ServiceError::BadPath);
+                    }
+                    path.pop();
+                    user_path_depth -= 1;
+                } else {
+                    path.push(seg.as_ref());
+                    user_path_depth += 1;
+                }
+            }
+            if !path.exists() {
+                return Err(ServiceError::NotFound);
             }
 
-            return index(&path, rpath.to_string())
-                .await
-                .map(html_string_response);
-        } else {
-            return Err(ServiceError::NotFound);
-        }
-    }
+            let metadata = path.metadata()?;
 
-    let modified = metadata.modified()?;
+            if metadata.file_type().is_dir() {
+                debug!("sending index for {path:?}");
+                if let Ok(indexhtml) = read_to_string(path.join("index.html")).await {
+                    return Ok(html_string_response(indexhtml));
+                }
 
-    let not_modified = if config.last_modified {
-        req.headers()
-            .typed_get::<headers::IfModifiedSince>()
-            .map(|if_modified_since| {
-                Ok::<_, ServiceError>(!if_modified_since.is_modified(modified))
-            })
-            .transpose()?
-            .unwrap_or_default()
-    } else {
-        false
-    };
+                if self.index {
+                    if !rpath.ends_with("/") {
+                        let mut r = Response::new(String::new());
+                        *r.status_mut() = StatusCode::FOUND;
+                        r.headers_mut().insert(
+                            LOCATION,
+                            HeaderValue::from_str(&format!("{}/", rpath))
+                                .map_err(|_| ServiceError::Other)?,
+                        );
+                        return Ok(r.map(|b| b.map_err(|e| match e {}).boxed()));
+                    }
+
+                    return index(&path, rpath.to_string())
+                        .await
+                        .map(html_string_response);
+                } else {
+                    return Err(ServiceError::NotFound);
+                }
+            }
 
-    // let etag = ETag::from_str(&calc_etag(modified)).map_err(|_| ServiceError::Other)?;
-    // let etag_matches = if config.etag {
-    //     req.headers()
-    //         .typed_get::<headers::IfNoneMatch>()
-    //         .map(|if_none_match| if_none_match.precondition_passes(&etag))
-    //         .unwrap_or_default()
-    // } else {
-    //     false
-    // };
+            let modified = metadata.modified()?;
 
-    let range = req.headers().typed_get::<headers::Range>();
-    let range = bytes_range(range, metadata.len())?;
+            let not_modified = if self.last_modified {
+                request
+                    .headers()
+                    .typed_get::<headers::IfModifiedSince>()
+                    .map(|if_modified_since| {
+                        Ok::<_, ServiceError>(!if_modified_since.is_modified(modified))
+                    })
+                    .transpose()?
+                    .unwrap_or_default()
+            } else {
+                false
+            };
 
-    debug!("sending file {path:?}");
-    let file = File::open(path.clone()).await?;
+            // let etag = ETag::from_str(&calc_etag(modified)).map_err(|_| ServiceError::Other)?;
+            // let etag_matches = if self.etag {
+            //     request.headers()
+            //         .typed_get::<headers::IfNoneMatch>()
+            //         .map(|if_none_match| if_none_match.precondition_passes(&etag))
+            //         .unwrap_or_default()
+            // } else {
+            //     false
+            // };
 
-    // let skip_body = not_modified || etag_matches;
-    let skip_body = not_modified;
-    let mut r = if skip_body {
-        Response::new("".to_string()).map(|b| b.map_err(|e| match e {}).boxed())
-    } else {
-        Response::new(BoxBody::new(StreamBody::new(
-            StreamBody::new(file_stream(file, 4096, range.clone()))
-                .map(|e| e.map(|e| Frame::data(e)).map_err(ServiceError::Io)),
-        )))
-    };
+            let range = request.headers().typed_get::<headers::Range>();
+            let range = bytes_range(range, metadata.len())?;
 
-    if !skip_body {
-        if range.end - range.start != metadata.len() {
-            *r.status_mut() = StatusCode::PARTIAL_CONTENT;
-            r.headers_mut().typed_insert(
-                ContentRange::bytes(range.clone(), metadata.len()).expect("valid ContentRange"),
-            );
-        }
-    }
-    // if not_modified || etag_matches {
-    if not_modified {
-        *r.status_mut() = StatusCode::NOT_MODIFIED;
-    }
+            debug!("sending file {path:?}");
+            let file = File::open(path.clone()).await?;
 
-    r.headers_mut().typed_insert(AcceptRanges::bytes());
-    r.headers_mut()
-        .typed_insert(ContentLength(range.end - range.start));
+            // let skip_body = not_modified || etag_matches;
+            let skip_body = not_modified;
+            let mut r = if skip_body {
+                Response::new("".to_string()).map(|b| b.map_err(|e| match e {}).boxed())
+            } else {
+                Response::new(BoxBody::new(StreamBody::new(
+                    StreamBody::new(file_stream(file, 4096, range.clone()))
+                        .map(|e| e.map(|e| Frame::data(e)).map_err(ServiceError::Io)),
+                )))
+            };
 
-    let mime = mime_guess::from_path(path).first_or_octet_stream();
-    r.headers_mut().typed_insert(ContentType::from(mime));
+            if !skip_body {
+                if range.end - range.start != metadata.len() {
+                    *r.status_mut() = StatusCode::PARTIAL_CONTENT;
+                    r.headers_mut().typed_insert(
+                        ContentRange::bytes(range.clone(), metadata.len())
+                            .expect("valid ContentRange"),
+                    );
+                }
+            }
+            // if not_modified || etag_matches {
+            if not_modified {
+                *r.status_mut() = StatusCode::NOT_MODIFIED;
+            }
 
-    r.headers_mut().typed_insert(match config.cache {
-        CacheConfig::Public => CacheControl::new().with_public(),
-        CacheConfig::Private => CacheControl::new().with_private(),
-        CacheConfig::NoStore => CacheControl::new().with_no_store(),
-    });
+            r.headers_mut().typed_insert(AcceptRanges::bytes());
+            r.headers_mut()
+                .typed_insert(ContentLength(range.end - range.start));
 
-    // if config.etag {
-    //     r.headers_mut().typed_insert(etag);
-    // }
-    if config.last_modified {
-        r.headers_mut().typed_insert(LastModified::from(modified));
-    }
+            let mime = mime_guess::from_path(path).first_or_octet_stream();
+            r.headers_mut().typed_insert(ContentType::from(mime));
 
-    Ok(r)
+            r.headers_mut().typed_insert(match self.cache {
+                CacheMode::Public => CacheControl::new().with_public(),
+                CacheMode::Private => CacheControl::new().with_private(),
+                CacheMode::NoStore => CacheControl::new().with_no_store(),
+            });
+
+            // if self.etag {
+            //     r.headers_mut().typed_insert(etag);
+            // }
+            if self.last_modified {
+                r.headers_mut().typed_insert(LastModified::from(modified));
+            }
+
+            Ok(r)
+        })
+    }
 }
 
 // Adapted from warp (https://github.com/seanmonstar/warp/blob/master/src/filters/fs.rs). Thanks!