refactor

Author: metame

server/src/files.rs

@@ -1,5 +1,7 @@
 use std::{fmt, path::PathBuf, time::Duration};
 
+use actix_multipart::Field;
+use futures::StreamExt;
 use opendal::{services::S3, Operator};
 use tokio::fs::File;
 
@@ -21,7 +23,7 @@ pub struct S3Config {
 
 #[derive(Clone, Debug)]
 pub struct FSConfig {
-    path: PathBuf,
+    pub path: PathBuf,
 }
 
 impl FileStore {
@@ -86,6 +88,36 @@ impl fmt::Display for FileStore {
     }
 }
 
+pub async fn s3_upload(
+    file_store:&FileStore,
+    file_id: &str,
+    mut field: Field,
+) -> AtomicServerResult<i64> {
+    let mut builder = S3::default();
+
+    if let FileStore::S3(config) = file_store {
+        builder.bucket(&config.bucket);
+        builder.root(&config.path);
+        config.region.as_ref().map(|r| builder.region(&r));
+        config.endpoint.as_ref().map(|e| builder.endpoint(&e));
+    } else {
+        return Err("Uploading to S3 but no S3 config provided".into());
+    }
+
+    let op: Operator = Operator::new(builder)?.finish();
+    let mut w = op.writer(file_id).await?;
+    let mut len = 0;
+    while let Some(chunk) = field.next().await {
+        let data = chunk.map_err(|e| format!("Error while reading multipart data. {}", e))?;
+        len = len + data.len();
+        w.write(data).await?;
+    }
+
+    let byte_length: i64 = len.try_into().map_err(|_e| "Too large")?;
+    w.close().await?;
+    Ok(byte_length)
+}
+
 pub async fn s3_upload_object(
     file_store: &FileStore,
     file_id: &str,

server/src/handlers/upload.rs

@@ -1,22 +1,18 @@
-use std::{
-    ffi::OsStr,
-    fs::{self, File},
-    io::Write,
-    path::Path,
-};
+use std::{ffi::OsStr, fs::File, io::Write, path::Path};
 
-use actix_multipart::Multipart;
+use actix_multipart::{Multipart, Field};
 use actix_web::{web, HttpResponse};
 use atomic_lib::{
     commit::CommitResponse, hierarchy::check_write, urls, utils::now, Resource, Storelike, Value,
 };
+
 use futures::{StreamExt, TryStreamExt};
 use serde::Deserialize;
 
 use crate::{
     appstate::AppState,
     errors::AtomicServerResult,
-    files::{self, FileStore},
+    files::{self, FSConfig, FileStore},
     helpers::get_client_agent,
 };
 
@@ -54,43 +50,24 @@ pub async fn upload_handler(
     let mut created_resources: Vec<Resource> = Vec::new();
     let mut commit_responses: Vec<CommitResponse> = Vec::new();
 
-    while let Ok(Some(mut field)) = body.try_next().await {
+    while let Ok(Some(field)) = body.try_next().await {
         let content_type = field.content_disposition().clone();
         let filename = content_type.get_filename().ok_or("Filename is missing")?;
 
-        std::fs::create_dir_all(&appstate.config.uploads_path)?;
-
-        let fs_file_id = format!(
-            "{}-{}",
+        let file_store = &appstate.file_store;
+        let file_id = format!(
+            "{}{}-{}",
+            file_store.prefix(),
             now(),
             sanitize_filename::sanitize(filename)
                 // Spacebars lead to very annoying bugs in browsers
                 .replace(' ', "-")
         );
 
-        let mut file_path = appstate.config.uploads_path.clone();
-        file_path.push(&fs_file_id);
-        let mut file = File::create(&file_path)?;
-
-        // Field in turn is stream of *Bytes* object
-        while let Some(chunk) = field.next().await {
-            let data = chunk.map_err(|e| format!("Error while reading multipart data. {}", e))?;
-            // TODO: Update a SHA256 hash here for checksum
-            file.write_all(&data)?;
-        }
-
-        let byte_count: i64 = file
-            .metadata()?
-            .len()
-            .try_into()
-            .map_err(|_e| "Too large")?;
-
-        let file_store = &appstate.file_store;
-        let file_id = format!("{}{}", file_store.prefix(), &fs_file_id);
-        if let FileStore::S3(_) = file_store {
-            files::s3_upload_object(&file_store, &file_id, &file_path).await?;
-            fs::remove_file(&file_path)?;
-        }
+        let byte_count = match file_store {
+            FileStore::S3(_) => files::s3_upload(&file_store, &file_id, field).await?,
+            FileStore::FS(config) => fs_upload(&file_store, &config, &file_id, field).await?,
+        };
 
         let subject_path = format!("files/{}", urlencoding::encode(&file_id));
         let new_subject = format!("{}/{}", store.get_server_url(), subject_path);
@@ -132,6 +109,33 @@ pub async fn upload_handler(
     )?))
 }
 
+
+async fn fs_upload(
+    file_store: &FileStore,
+    config: &FSConfig,
+    file_id: &str,
+    mut field: Field,
+) -> AtomicServerResult<i64> {
+    std::fs::create_dir_all(config.path.clone())?;
+
+    let mut file = File::create(file_store.get_fs_file_path(file_id)?)?;
+
+    let byte_count: i64 = file
+        .metadata()?
+        .len()
+        .try_into()
+        .map_err(|_e| "Too large")?;
+
+    // Field in turn is stream of *Bytes* object
+    while let Some(chunk) = field.next().await {
+        let data = chunk.map_err(|e| format!("Error while reading multipart data. {}", e))?;
+        // TODO: Update a SHA256 hash here for checksum
+        file.write_all(&data)?;
+    }
+
+    Ok(byte_count)
+}
+
 fn guess_mime_for_filename(filename: &str) -> String {
     if let Some(ext) = get_extension_from_filename(filename) {
         actix_files::file_extension_to_mime(ext).to_string()