fix upload bug

Author: timzaak

Cargo.lock

@@ -29,6 +29,9 @@
     "dist",
     "!dist/test"
   ],
+  "engines": {
+    "node": ">=18.0.0"
+  },
   "dependencies": {
     "@pushcorn/hocon-parser": "^1.2.9",
     "axios": "^1.6.8",

jsclient/package.json

@@ -183,11 +183,10 @@ pub mod service {
     };
     use crate::domain_storage::{DomainStorage, URI_REGEX};
     use crate::{AdminConfig, HotReloadManager};
-    use anyhow::anyhow;
-    use bytes::{Buf, Bytes};
-    use futures_util::TryStreamExt;
+    use anyhow::{anyhow, Context};
+    use bytes::Buf;
+    use futures_util::{StreamExt, TryStreamExt};
     use hyper::Body;
-    use if_chain::if_chain;
     use std::convert::Infallible;
     use std::sync::Arc;
     use warp::http::StatusCode;
@@ -282,43 +281,27 @@ pub mod service {
         form: FormData,
         storage: Arc<DomainStorage>,
     ) -> anyhow::Result<Response> {
-        let file: Vec<Option<Bytes>> = form
-            .and_then(|mut field| async move {
-                let name = field.name();
-                tracing::debug!("field name:{}", name);
-                let bytes = field
-                    .data()
-                    .await
-                    .map(|x| {
-                        x.map(|mut x| {
-                            let i = x.remaining();
-                            (&mut x).copy_to_bytes(i)
-                        })
-                        .ok()
-                    })
-                    .flatten();
-                Ok(bytes)
-            })
-            .try_collect()
-            .await?;
-        let file = file.into_iter().nth(0).flatten();
-
-        tracing::debug!(
-            "uploading file: {:?}, {:?}, {:?}",
-            &query.domain,
-            &query.version,
-            &query.path
-        );
-
-        if_chain! {
-            if let Some(file_buf) = file;
-            then {
-                storage.save_file(query.domain, query.version, query.path, file_buf)?;
-                Ok(Response::default())
-            } else {
-                Err(anyhow!("bad params, please check the api doc: https://github.com/fornetcode/spa-server/blob/master/docs/guide/sap-server-api.md"))
-            }
+        let mut parts = form.into_stream();
+        while let Some(Ok(part)) = parts.next().await {
+            // let name = part.name();
+            let file = part
+                .stream()
+                .try_fold(Vec::new(), |mut acc, buf| async move {
+                    acc.extend_from_slice(buf.chunk());
+                    Ok(acc)
+                })
+                .await
+                .with_context(|| "get form failure")?;
+            tracing::debug!(
+                "uploading file: {:?}, {:?}, {:?}",
+                &query.domain,
+                &query.version,
+                &query.path
+            );
+            storage.save_file(query.domain, query.version, query.path, file)?;
+            return Ok(Response::default());
         }
+        return Err(anyhow!("bad params, please check the api doc: https://github.com/fornetcode/spa-server/blob/master/docs/guide/sap-server-api.md"));
     }
 
     pub(super) fn get_files_metadata(

server/src/admin_server.rs

@@ -1,9 +1,10 @@
 use anyhow::Result;
 use tracing_subscriber::EnvFilter;
 
-
 #[tokio::main]
 async fn main() -> Result<()> {
-    tracing_subscriber::fmt().with_env_filter(EnvFilter::from_default_env()).init();
+    tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .init();
     spa_server::run_server().await
 }

server/src/bin/main.rs

@@ -30,7 +30,7 @@ impl Config {
 
         let load_file = hocon::HoconLoader::new()
             .load_file(&config_path)
-            .with_context(|| "can not read config file")?;
+            .with_context(|| format!("can not read config file: {config_path}"))?;
 
         load_file
             .resolve::<Config>()

server/src/config.rs

@@ -1,6 +1,5 @@
 use crate::file_cache::{CacheItem, FileCache};
 use anyhow::anyhow;
-use bytes::Bytes;
 use dashmap::DashMap;
 use lazy_static::lazy_static;
 use md5::{Digest, Md5};
@@ -76,7 +75,7 @@ impl DomainStorage {
                         let path_buf = path_prefix_buf
                             .join(domain_dir_name)
                             .join(max_version.to_string());
-                        let data = cache.cache_dir(&domain_dir_name,&path_buf)?;
+                        let data = cache.cache_dir(&domain_dir_name, &path_buf)?;
                         cache.update(domain_dir_name.to_string(), data);
                         domain_version.insert(domain_dir_name.to_owned(), (path_buf, max_version));
                     }
@@ -186,50 +185,63 @@ impl DomainStorage {
         }
     }
 
-    pub fn get_domain_serving_version(&self, domain:&str) -> Option<u32> {
-        self.meta.get(domain).map(|x|x.1)
+    pub fn get_domain_serving_version(&self, domain: &str) -> Option<u32> {
+        self.meta.get(domain).map(|x| x.1)
     }
 
     pub fn get_domain_info_by_domain(&self, domain: &str) -> Option<DomainInfo> {
-        let versions:Vec<u32> = WalkDir::new(&self.prefix.join(domain)).max_depth(1).into_iter().filter_map(|version_entity|{
-            let version_entity =version_entity.ok()?;
-            let version = version_entity.file_name().to_str()?.parse::<u32>().ok()?;
-            Some(version)
-        }).collect();
+        let versions: Vec<u32> = WalkDir::new(&self.prefix.join(domain))
+            .max_depth(1)
+            .into_iter()
+            .filter_map(|version_entity| {
+                let version_entity = version_entity.ok()?;
+                let version = version_entity.file_name().to_str()?.parse::<u32>().ok()?;
+                Some(version)
+            })
+            .collect();
         if versions.is_empty() {
             None
         } else {
             let domain = domain.to_string();
-            let current_version = self.meta.get(&domain).map(|x|x.1);
+            let current_version = self.meta.get(&domain).map(|x| x.1);
             Some(DomainInfo {
                 domain,
                 current_version,
-                versions
+                versions,
             })
         }
     }
 
     pub fn get_domain_info(&self) -> Vec<DomainInfo> {
-        let ret:Vec<DomainInfo> = walkdir::WalkDir::new(&self.prefix).max_depth(1).into_iter().filter_map(|dir_entity|{
-            let dir_entity = dir_entity.ok()?;
-            let domain_dir_name = dir_entity.file_name().to_str()?;
-            if dir_entity.metadata().ok()?.is_dir() && URI_REGEX.is_match(domain_dir_name) {
-                let domain = domain_dir_name.to_string();
-                let current_version = self.meta.get(&domain).map(|x|x.1);
-                let versions = walkdir::WalkDir::new(dir_entity.path()).max_depth(1).into_iter().filter_map(|version_entity|{
-                    let version_entity =version_entity.ok()?;
-                    let version = version_entity.file_name().to_str()?.parse::<u32>().ok()?;
-                    Some(version)
-                }).collect();
-                Some(DomainInfo {
-                    domain,
-                    current_version,
-                    versions
-                })
-            } else {
-                None
-            }
-        }).collect();
+        let ret: Vec<DomainInfo> = walkdir::WalkDir::new(&self.prefix)
+            .max_depth(1)
+            .into_iter()
+            .filter_map(|dir_entity| {
+                let dir_entity = dir_entity.ok()?;
+                let domain_dir_name = dir_entity.file_name().to_str()?;
+                if dir_entity.metadata().ok()?.is_dir() && URI_REGEX.is_match(domain_dir_name) {
+                    let domain = domain_dir_name.to_string();
+                    let current_version = self.meta.get(&domain).map(|x| x.1);
+                    let versions = walkdir::WalkDir::new(dir_entity.path())
+                        .max_depth(1)
+                        .into_iter()
+                        .filter_map(|version_entity| {
+                            let version_entity = version_entity.ok()?;
+                            let version =
+                                version_entity.file_name().to_str()?.parse::<u32>().ok()?;
+                            Some(version)
+                        })
+                        .collect();
+                    Some(DomainInfo {
+                        domain,
+                        current_version,
+                        versions,
+                    })
+                } else {
+                    None
+                }
+            })
+            .collect();
         ret
     }
     fn check_is_in_upload_process(&self, domain: &str, version: &u32) -> bool {
@@ -290,7 +302,7 @@ impl DomainStorage {
         domain: String,
         version: u32,
         path: String,
-        data: Bytes,
+        data: Vec<u8>,
     ) -> anyhow::Result<()> {
         if self.check_is_in_upload_process(&domain, &version) {
             let file_path = sanitize_path(self.get_version_path(&domain, version), &path)
@@ -379,21 +391,25 @@ impl DomainStorage {
         Ok(())
     }
 
-    pub fn remove_domain_version(&self, domain:&str, version:Option<u32>) ->anyhow::Result<bool> {
+    pub fn remove_domain_version(
+        &self,
+        domain: &str,
+        version: Option<u32>,
+    ) -> anyhow::Result<bool> {
         let mut path = self.prefix.join(domain);
         if let Some(version) = version {
             path = path.join(version.to_string());
             if path.exists() {
                 fs::remove_dir_all(path)?;
                 return Ok(true);
             }
-        }else {
+        } else {
             if path.exists() {
                 fs::remove_dir_all(path)?;
                 return Ok(true);
             }
         }
-        return Ok(false)
+        return Ok(false);
     }
 }