Switch to streaming the uploaded file to disk

Cargo.lock

@@ -297,6 +297,7 @@ dependencies = [
  "diesel",
  "diesel-async",
  "diesel_async_migrations",
+ "futures",
  "futures-util",
  "image",
  "memmap2",
@@ -307,6 +308,7 @@ dependencies = [
  "steam-openid",
  "time",
  "tokio",
+ "tokio-util",
  "tower-http",
  "tower-sessions",
  "tracing",

backend/Cargo.toml

@@ -10,11 +10,13 @@ serde = { version = "1.0.210", features = ["derive"] }
 steam-openid = "0.2.0"
 time = { version = "0.3.36", features = ["formatting", "parsing"] }
 tokio = { version = "1.40.0", features = ["rt", "macros", "net", "mio", "rt-multi-thread"] }
+tokio-util = { version = "0.7", features = ["io"]}
 tower-sessions = "0.13.0"
 tower-http = { version = "0.6", features = ["fs"] }
 tracing = { version = "0.1.40", features = ["async-await"] }
 tracing-subscriber = "0.3.18"
 futures-util = "0.3"
+futures = "0.3"
 
 diesel = { version = "2.2", features = ["serde_json", "chrono"] }
 diesel-async = { version = "0.5", features = ["postgres"] }

backend/src/api.rs

@@ -169,3 +169,31 @@ pub fn router(config: RouterConfig) -> axum::Router {
         .nest("/demos/", demos::router(config.upload_dir))
         .nest("/user/", user::router())
 }
+
+// Save a `Stream` to a file
+async fn stream_to_file<S, E>(path: impl Into<std::path::PathBuf>, stream: S) -> Result<(), (axum::http::StatusCode, String)>
+where
+    S: futures::Stream<Item = Result<axum::body::Bytes, E>>,
+    E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
+{
+    use futures::{Stream, TryStreamExt};
+    
+    let path: std::path::PathBuf = path.into();
+
+    async {
+        // Convert the stream into an `AsyncRead`.
+        let body_with_io_error = stream.map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err));
+        let body_reader = tokio_util::io::StreamReader::new(body_with_io_error);
+        futures::pin_mut!(body_reader);
+
+        // Create the file. `File` implements `AsyncWrite`.
+        let mut file = tokio::io::BufWriter::new(tokio::fs::File::create(path).await?);
+
+        // Copy the body into the file.
+        tokio::io::copy(&mut body_reader, &mut file).await?;
+
+        Ok::<_,std::io::Error>(())
+    }
+    .await
+    .map_err(|err| (axum::http::StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))
+}

backend/src/api/demos.rs

@@ -93,7 +93,7 @@ async fn list(
 async fn upload(
     State(state): State<Arc<DemoState>>,
     session: crate::UserSession,
-    form: axum::extract::Multipart,
+    mut form: axum::extract::Multipart,
 ) -> Result<axum::response::Redirect, (axum::http::StatusCode, &'static str)> {
     let steam_id = session
         .data()
@@ -102,22 +102,27 @@ async fn upload(
 
     tracing::info!("Upload for Session: {:?}", steam_id);
 
-    let file_content = match crate::get_demo_from_upload("demo", form).await {
+    let file_field = loop {
-        Some(c) => c,
+        let field = match form.next_field().await {
-        None => {
+            Ok(Some(f)) => f,
-            tracing::error!("Getting File content from request");
+            Ok(None) => {
-            return Err((
+                tracing::error!("");
-                axum::http::StatusCode::BAD_REQUEST,
+                return Err((axum::http::StatusCode::BAD_REQUEST, "Missing Data"));
-                "Failed to get file-content from upload",
+            }
-            ));
+            Err(e) => {
+                tracing::error!("");
+                return Err((axum::http::StatusCode::BAD_REQUEST, ""));
+            }
+        };
+
+        if field.name().map(|n| n == "demo").unwrap_or(false) {
+            break field;
         }
     };
 
     let raw_demo_id = uuid::Uuid::now_v7();
     let demo_id = raw_demo_id.to_string();
 
-    tracing::debug!(?demo_id, "Upload Size: {:?}", file_content.len());
-
     let user_folder = std::path::Path::new(&state.upload_folder).join(format!("{}/", steam_id));
     if !tokio::fs::try_exists(&user_folder).await.unwrap_or(false) {
         tokio::fs::create_dir_all(&user_folder).await.unwrap();
@@ -125,9 +130,7 @@ async fn upload(
 
     let demo_file_path = user_folder.join(format!("{}.dem", demo_id));
 
-    tokio::fs::write(&demo_file_path, file_content)
+    super::stream_to_file(demo_file_path, file_field).await.unwrap();
-        .await
-        .unwrap();
 
     let mut db_con = crate::db_connection().await;
 

backend/src/lib.rs

@@ -18,28 +18,6 @@ pub async fn db_connection() -> diesel_async::AsyncPgConnection {
         .unwrap_or_else(|e| panic!("Error connecting to {} - {:?}", database_url, e))
 }
 
-pub async fn get_demo_from_upload(
-    name: &str,
-    mut form: axum::extract::Multipart,
-) -> Option<axum::body::Bytes> {
-    while let Ok(field) = form.next_field().await {
-        let field = match field {
-            Some(f) => f,
-            None => continue,
-        };
-
-        if field.name().map(|n| n != name).unwrap_or(false) {
-            continue;
-        }
-
-        if let Ok(data) = field.bytes().await {
-            return Some(data);
-        }
-    }
-
-    None
-}
-
 pub mod api;
 pub mod steam_api;