finish refactor

Author: metame

server/src/files.rs

@@ -1,9 +1,8 @@
-use std::{fmt, path::PathBuf, time::Duration};
+use std::{fmt, fs, io::Write, path::PathBuf, time::Duration};
 
 use actix_multipart::Field;
 use futures::StreamExt;
 use opendal::{services::S3, Operator};
-use tokio::fs::File;
 
 use crate::{appstate::AppState, config::Config, errors::AtomicServerResult};
 
@@ -80,6 +79,13 @@ impl FileStore {
     pub fn encoded(&self) -> String {
         urlencoding::encode(self.prefix()).into_owned()
     }
+
+    pub async fn upload_file(&self, file_id: &str, field: Field) -> AtomicServerResult<i64> {
+        match self {
+            FileStore::S3(_) => s3_upload(self, &file_id, field).await,
+            FileStore::FS(config) => fs_upload(self, &config, &file_id, field).await,
+        }
+    }
 }
 
 impl fmt::Display for FileStore {
@@ -88,41 +94,37 @@ impl fmt::Display for FileStore {
     }
 }
 
-pub async fn s3_upload(
-    file_store:&FileStore,
+async fn fs_upload(
+    file_store: &FileStore,
+    config: &FSConfig,
     file_id: &str,
     mut field: Field,
 ) -> AtomicServerResult<i64> {
-    let mut builder = S3::default();
+    std::fs::create_dir_all(config.path.clone())?;
 
-    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 mut file = fs::File::create(file_store.get_fs_file_path(file_id)?)?;
 
-    let op: Operator = Operator::new(builder)?.finish();
-    let mut w = op.writer(file_id).await?;
-    let mut len = 0;
+    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))?;
-        len = len + data.len();
-        w.write(data).await?;
+        // TODO: Update a SHA256 hash here for checksum
+        file.write_all(&data)?;
     }
 
-    let byte_length: i64 = len.try_into().map_err(|_e| "Too large")?;
-    w.close().await?;
-    Ok(byte_length)
+    Ok(byte_count)
 }
 
-pub async fn s3_upload_object(
+async fn s3_upload(
     file_store: &FileStore,
     file_id: &str,
-    file_path: &PathBuf,
-) -> AtomicServerResult<()> {
+    mut field: Field,
+) -> AtomicServerResult<i64> {
     let mut builder = S3::default();
 
     if let FileStore::S3(config) = file_store {
@@ -135,14 +137,17 @@ pub async fn s3_upload_object(
     }
 
     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 mut tmp_file = File::open(file_path).await?;
-    let length = tmp_file.metadata().await?.len();
-
-    let mut w = op.writer_with(&file_id).content_length(length).await?;
-    tokio::io::copy(&mut tmp_file, &mut w).await?;
+    let byte_length: i64 = len.try_into().map_err(|_e| "Too large")?;
     w.close().await?;
-    Ok(())
+    Ok(byte_length)
 }
 
 pub async fn get_s3_signed_url(

server/src/handlers/upload.rs

@@ -1,18 +1,17 @@
-use std::{ffi::OsStr, fs::File, io::Write, path::Path};
+use std::{ffi::OsStr, path::Path};
 
-use actix_multipart::{Multipart, Field};
+use actix_multipart::Multipart;
 use actix_web::{web, HttpResponse};
 use atomic_lib::{
     commit::CommitResponse, hierarchy::check_write, urls, utils::now, Resource, Storelike, Value,
 };
 
-use futures::{StreamExt, TryStreamExt};
+use futures::TryStreamExt;
 use serde::Deserialize;
 
 use crate::{
     appstate::AppState,
     errors::AtomicServerResult,
-    files::{self, FSConfig, FileStore},
     helpers::get_client_agent,
 };
 
@@ -64,10 +63,7 @@ pub async fn upload_handler(
                 .replace(' ', "-")
         );
 
-        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 byte_count = file_store.upload_file(&file_id, field).await?;
 
         let subject_path = format!("files/{}", urlencoding::encode(&file_id));
         let new_subject = format!("{}/{}", store.get_server_url(), subject_path);
@@ -109,33 +105,6 @@ 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()