Basic modifications..

Author: artech-git

Cargo.lock

@@ -9,6 +9,7 @@ edition = "2021"
 axum = { version = "0.7.2", features = ["http2", "multipart", "http1"] }
 axum-core = "0.4.1"
 bytes = "1.5.0"
+config = "0.13.4"
 dashmap = "5.5.3"
 futures = "0.3.29"
 futures-stream = "0.0.0"

Cargo.toml

@@ -0,0 +1,4 @@
+
+
+data = "./data"
+file_size = 1_000_0000

settings.toml

@@ -0,0 +1,36 @@
+use axum::http::{HeaderMap, HeaderValue, header::*};
+
+use crate::errors::{FragmentError, HeaderErrors};
+
+
+pub async fn ensure_headers(mut headers: &HeaderMap) -> Result<(), FragmentError> { 
+    
+    let header = { 
+        let fields = [
+            ACCESS_CONTROL_ALLOW_ORIGIN, 
+            CONTENT_LENGTH, 
+            ACCEPT_ENCODING];
+
+        for assumed_header in fields.iter() { 
+            if let Some(val) = headers.get(assumed_header){
+                continue; 
+            } 
+        }
+    };
+
+
+    Ok(header_value)
+
+}
+
+
+// return the value of the associated field in the headermap 
+pub async fn extract_header_fields(headers: &HeaderMap, header_field: &HeaderName) -> Result<HeaderValue, FragmentError> { 
+
+    if let Some(val) = headers.get(header_field) { 
+        return Ok(val.to_owned());
+    }
+    else { 
+        return Err(HeaderErrors::HeaderFieldMissing(header_field));
+    }
+}

src/authorization.rs

@@ -0,0 +1,46 @@
+use std::path::PathBuf;
+
+use config::Config;
+
+
+#[derive(Debug)]
+pub struct LoadConfig { 
+    path: PathBuf, 
+    config_data: Config
+}
+
+impl Default for LoadConfig {
+    fn default() -> Self { 
+    
+        let path_name = "./settings.toml";
+        let path = PathBuf::from(path_name);
+
+        let config = Config::builder()
+            .add_source(config::File::with_name(path_name))
+            .build()
+            .expect("builder unable to parse the config file");
+
+        Self { 
+            path: path,
+            config_data: config
+        }
+    }
+}
+
+impl LoadConfig { 
+    fn new(source: impl AsRef<str>) -> Self {
+        
+        let path_name = source.as_ref();
+        let path = PathBuf::from(path_name);
+
+        let config = Config::builder()
+            .add_source(config::File::with_name(path_name))
+            .build()
+            .expect("builder unable to parse the config file");
+
+        Self { 
+            path: path, 
+            config_data: config
+        }
+    }
+}

src/config.rs

@@ -0,0 +1,5 @@
+
+
+
+
+pub const File_SIZE_LIMIT: u64 = (78 * 1024 * 1024) * 1024 ; // 78 gigs 

src/constants.rs

@@ -1,4 +1,4 @@
-use std::{panic::Location, backtrace::Backtrace};
+use std::{panic::Location, backtrace::Backtrace, borrow::Cow};
 
 use axum::{http::{StatusCode, HeaderMap, Response, self, HeaderValue},response::IntoResponse, body::Body};
 use thiserror::Error;
@@ -18,6 +18,24 @@ use std::convert::From;
 //     }
 // }
 
+pub trait OptionExt<V, E, T>
+{ 
+    fn to_result(self, method: T) -> Result<V, E>; 
+}
+
+impl<V, E, T> OptionExt<V, E, T> for Option<V> 
+
+where  
+        E: Into<ErrorStates>,
+        T: FnOnce() -> E
+{ 
+    fn to_result(self, method: T) -> Result<V, E>
+    {
+        let err = method(); 
+        Err(err)
+    }   
+}
+
 #[derive(Debug)]
 pub struct ErrorReport { 
     reason: ErrorStates, 
@@ -125,6 +143,11 @@ pub enum ErrorStates {
     InternalError(#[from] axum::Error),
 
 
+    #[http(code = 404, message = "Missing Header Field")]
+    #[error("Missing header field")]
+    RequestError(#[from] HeaderErrors<'static>),
+
+    
     // #[http(code = 500, message = "server went into undesired mode")]
     // #[error("internal socket Error")]
     // SocketError(#[from] ),
@@ -138,3 +161,32 @@ pub enum ErrorStates {
 
 
 }
+
+
+#[derive(Debug, thiserror::Error, HttpError)]
+pub enum HeaderErrors<'a> {
+
+    #[error("Missing header field: {0:?}")]
+    HeaderFieldMissing(Cow<'a, str>),
+
+    #[error("Field mismatch: {0:?}")]
+    FieldMismatch(Cow<'a, str>),
+
+    #[error("Invalid field input: {0:?}")]
+    InvalidField(Cow<'a, str>),
+
+    #[error("Invalid field input: {0:?}")]
+    HeaderUnwrapError(#[from] http::header::ToStrError),
+
+    // #[error("Invalid field input: {0:?}")]
+    // InvalidField(Cow<'a, str>)
+}
+
+
+// impl<'a, T> std::convert::From<T> for HeaderErrors<'a>
+// where T: std::error::Error { 
+
+//     fn from(val: T) -> Self { 
+//         val.into()
+//     }
+// }

src/errors.rs

@@ -1,4 +1,4 @@
-use std::{path::PathBuf, usize};
+use std::{path::PathBuf, usize, borrow::Cow};
 use serde::{Serialize, Deserialize}; 
 use tokio::sync::oneshot;
 use uuid::Uuid;
@@ -8,33 +8,40 @@ use uuid::Uuid;
 
 #[derive(Debug, Serialize, Deserialize )]
 pub struct FileObject { 
-    path: PathBuf, 
+    pub path: PathBuf, 
     state: UploadState, 
     file_size: usize, 
     name: String, 
     uuid: Uuid, 
     // hash: [u8; 256],
-    // sender: oneshot::Sender<usize> 
 }
 
 impl FileObject { 
 
-    pub fn new(path: impl Into<PathBuf>, size: usize, ) -> Self { 
+    pub fn new(path: impl Into<PathBuf>, size: usize, name: impl ToString) -> Self { 
         Self { 
-            path: path.into(), 
+            path: path.into(),  
             state: UploadState::UnInit, 
             file_size: size, 
-            name: "".to_string(),
+            name: name.to_string(),
             uuid: Uuid::new_v4(), 
             // hash: [0_u8; 256]
         }
     }
 
-
-    pub fn update_state(&mut self, state: UploadState) {
+    #[inline(always)]
+    pub fn set_state(&mut self, state: UploadState) {
         self.state = state; 
     }
 
+    #[inline(always)]
+    pub fn get_uuid(&self) -> Cow<'_, Uuid> { 
+        Cow::Borrowed(&(self.uuid))
+    }
+    
+    pub fn output_file_path(&self) -> PathBuf {
+        self.path.join(self.name)
+    }
 
 }
 
@@ -43,10 +50,10 @@ impl FileObject {
 #[derive(Debug, Serialize, Deserialize)]
 pub enum UploadState {
     UnInit, 
-    Failed, 
     Broken(usize),
     Resume(usize), 
     Complete, 
+    Failed, 
 }
 
 

src/file.rs

@@ -1,89 +1,147 @@
-use std::{sync::Arc, path::PathBuf, str::FromStr};
+use std::{sync::Arc, path::PathBuf, str::FromStr, borrow::Cow};
 
 use dashmap::DashMap; 
-use axum::{http::{Request, HeaderValue}, body::Body, response::IntoResponse, Extension};
-use tokio::task::JoinHandle;
-use uuid::{Uuid, uuid};
+use axum::{http::{Request, HeaderValue, HeaderMap, header::{*}}, body::Body, response::IntoResponse, Extension};
+use uuid::Uuid;
 use futures::stream::StreamExt;
-use tokio::io::{AsyncWrite, AsyncWriteExt};
+use tokio::io::AsyncWriteExt;
+use crate::{errors::{OptionExt, ErrorStates, HeaderErrors}, authorization::extract_header_fields}; 
 
 use crate::{file::FileObject, utils::generate_file, FragmentError};
 
 
 // Task handle for keeping track of all the spawned instance on the runtime
-pub type JobHandle<E> = Arc<DashMap<Uuid, JoinHandle<Result<FileObject, E>>>>; 
+pub type JobHandle<E> = Arc< DashMap<Uuid, Result<FileObject, E>>>; 
 
+/*
 
-pub async fn init_upload_process(
-    mut ext: Extension<JobHandle<FragmentError>>,
-    mut req: Request<Body>, 
-) -> Result<impl IntoResponse, FragmentError> {
+*/
+async fn streamer_uploader(
+    mut body: Body, 
+    mut file: FileObject
+) -> Result<(), FragmentError> {
+
+    let streamer = body.into_data_stream();
 
-    let (header, body) = req.into_parts(); 
+    let mut buf_size = 1_000_0000;  //REMOVE: allocated buffer_size
 
-    let uuid = uuid::Uuid::new_v4();
-    let mut path = PathBuf::from("./data/");
 
-    //******************************** */
-    let file_name = header.headers.get("FileName").unwrap().to_str().unwrap().to_string(); 
-    // let file_size = header.headers
-    //     .get("FileSize")
-    //     .unwrap()
-    //     .as_ref()
-    //     .parse::<usize>()
-    //     .unwrap(); 
+    let file_path = file.
+
+    let mut buf_writer = generate_file( streamer, buf_size).await?; 
+
+    let mut _chunk_counter = 0; 
+    let mut _byte_counter = 0; 
+
+    println!("we entered the stream");
+
+    while let Some(chunk) = stream.next().await { 
+        let bytes = chunk?;
+        
+        tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+        _byte_counter += buf_writer.write(&bytes).await?; 
+        _chunk_counter += 1;             
+    }
+    
+    drop(stream);
 
-    let file_size = 1000000; 
+    let _ = buf_writer.shutdown().await?;
+}
 
-    let mut file_obj = FileObject::new(path.clone(), file_size); 
 
-    println!("task called");
+//validating the header data of the 
+fn validate_headers(header: HeaderValue) -> Result<(), HeaderErrors<'static>> { 
+
+    if header.is_empty() {
+        return Err(HeaderErrors::HeaderFieldMissing(Cow::Borrowed("FileName")));
+    }
 
-    // let handle = tokio::task::spawn(async move {
-        let mut stream = body.into_data_stream(); 
-        //=========================================================
-        // let mut file_name = PathBuf::from("xxxxx-yyyyy.txt");
-        let mut buf_size = 1_000_0000;  //allocated buffer_size
-        //=========================================================
-        let mut buf_writer = generate_file(path.join(PathBuf::from_str(file_name.as_str()).unwrap()), buf_size).await?; 
+    let str_header = header.to_str()?; 
+    
+    if str_header.chars().all(|e| !e.is_ascii() || !e.is_alphanumeric() || e.is_ascii_hexdigit()) {
+        return Err(HeaderErrors::InvalidField(Cow::Borrowed(str_header)));
+    }
 
-        let mut _chunk_counter = 0; 
-        let mut _byte_counter = 0; 
+    Ok(())
+}
 
-        println!("we entered the stream");
 
-        while let Some(chunk) = stream.next().await{ 
-            let bytes = chunk?;
-            // println!("chunk accepted: {} len: {}", _chunk_counter, bytes.len()); 
-            tokio::time::sleep(std::time::Duration::from_millis(1)).await;
-            _byte_counter += buf_writer.write(&bytes).await?; 
-            _chunk_counter += 1;             
-        }
 
-        println!("stream closed: bytes rem");
 
-        // drop(stream);
-        
-        let _ = buf_writer.shutdown().await?;
+pub async fn init_upload_process(
+    Extension(ext): Extension<JobHandle<FragmentError>>,
+    req: Request<Body>, 
+) -> Result<impl IntoResponse, FragmentError> {
+
+    let (parts, body) = req.into_parts(); 
+    let headers = parts.headers; 
+
+    //=========================================================    
+    let headers_names = [
+        "FileName",
+        "Content-Length",
+    ];
+    
+    let extracted_headers = futures::future::join_all(headers_names
+        .iter()
+        .map(|field_name| {
+            let field = HeaderName::from_str(field_name).expect("Invalid header name");
+            extract_header_fields(&headers, &field)
+        })).await;
+    
+    //=========================================================
+    let (path, file_size, file_name) = { 
+        let file_name = extracted_headers[0]?.to_owned(); 
+        let file_size = extracted_headers[1]?.to_owned();
+
+        let _ = validate_headers(file_name)?;
+        let _ = validate_headers(file_size)?;
+
+        let path = "./data"; // TODO: change path to const. 
+
+        let file_size = file_size.to_str()
+            .map_err(|e| HeaderErrors::HeaderUnwrapError(e))?
+            .parse::<usize>()
+            .unwrap();  
+
+        let file_name = file_name.to_str()
+            .map_err(|e| HeaderErrors::HeaderUnwrapError(e))?; 
+
+        (path, file_size, file_name)
+    };
 
-    //     println!("stream closed");
-    //     Ok(file_obj)
-    // });
+    let mut file_obj = FileObject::new(path, file_size, file_name); 
+    //=========================================================
 
-    // let _ = ext.entry(uuid).or_insert(handle); //return error if stream is already present
+    let _ = streamer_uploader(body, file_obj).await?;
+
+
+        
+    let _ = ext.entry(uuid).or_insert(handle); //return error if stream is already present
 
-    println!("task exited");
+    // println!("task exited");
 
-    Ok(uuid.as_hyphenated().to_string())
+    // Ok(uuid.as_hyphenated().to_string())
+
+    Ok()
 
 }
 
 
+
+//Resume the interrupted upload process
+pub async fn resume_upload(
+    Extension(ext): Extension<JobHandle<FragmentError>>,
+    req: Request<Body>
+) -> Result<impl IntoResponse, FragmentError> {
+
+}
+
+// Handle for acquring the status of the In_progress, discarded or cancelled upload process 
 pub async fn task_progress(
     mut ext: Extension<JobHandle<FragmentError>>,
     mut req: Request<Body>, 
-) -> impl IntoResponse 
-    // where E: Into<FragmentError>
+) -> Result<impl IntoResponse, FragmentError> 
 { 
 
     let (mut headers, body) = req.into_parts(); 

src/handlers.rs

@@ -9,9 +9,10 @@ mod file;
 mod errors;
 mod utils; 
 mod handlers;
+mod authorization;
+mod config;
 
-#[tokio::main]
-async fn main() -> Result<(), FragmentError> { 
+async fn tokio_main() -> Result<(), FragmentError> { 
 
     let addr = [0_u8; 4].into(); 
     let port = 2053; 
@@ -38,5 +39,14 @@ async fn main() -> Result<(), FragmentError> {
     Ok(())
 }
 
+fn main() {
+    let mut runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .global_queue_interval(40)
+        .build()
+        .unwrap();
 
-
+    let tokio_main_process = tokio_main(); 
+    
+    runtime.block_on(tokio_main_process);
+}

src/main.rs

@@ -6,7 +6,8 @@ use tokio::fs::File;
 use tokio::io::BufWriter;  
 
 use crate::FragmentError;
-use crate::handlers::{JobHandle, init_upload_process, task_progress};
+use crate::config::LoadConfig;
+use crate::handlers::{JobHandle, init_upload_process, task_progress, resume_upload};
 
 
 pub async fn create_router( 
@@ -16,7 +17,8 @@ pub async fn create_router(
     let router = Router::new()
         .route("/upload_file", get(init_upload_process))
         .route("/status", get(task_progress))
-        .layer(Extension(ext)); 
+        .route("/resume_upload", get(resume_upload))
+        .layer(Extension(ext));  
 
     Ok(router)
 }
@@ -31,4 +33,8 @@ pub async fn generate_file(path: impl AsRef<Path>, size: usize) -> Result<BufWri
     let file =  File::create(path).await?;
     let buffered_file = BufWriter::with_capacity(size, file); 
     Ok(buffered_file)
+}
+
+pub async fn load_config(path: impl AsRef<str>) -> Result<LoadConfig, FragmentError>  {
+    Ok()
 }