From 998f92d2e3d524366531a1acb19204420f6eed58 Mon Sep 17 00:00:00 2001
From: Nikolay Kim <fafhrd91@gmail.com>
Date: Mon, 9 Dec 2019 06:28:09 +0600
Subject: [PATCH] fix multipart example

---
 diesel/src/main.rs       | 36 ++++++++++--------------------------
 multipart/rust-toolchain |  1 -
 multipart/src/main.rs    | 15 ++++++++-------
 3 files changed, 18 insertions(+), 34 deletions(-)
 delete mode 100644 multipart/rust-toolchain

diff --git a/diesel/src/main.rs b/diesel/src/main.rs
index 5e68345f..37603ee7 100644
--- a/diesel/src/main.rs
+++ b/diesel/src/main.rs
@@ -21,6 +21,11 @@ mod schema;
 
 type Pool = r2d2::Pool<ConnectionManager<SqliteConnection>>;
 
+#[derive(Debug, Serialize, Deserialize)]
+struct MyUser {
+    name: String,
+}
+
 /// Diesel query
 fn query(
     nm: String,
@@ -53,33 +58,11 @@ async fn add(
         .map_err(|_| HttpResponse::InternalServerError())?)
 }
 
-#[derive(Debug, Serialize, Deserialize)]
-struct MyUser {
-    name: String,
-}
-
-const MAX_SIZE: usize = 262_144; // max payload size is 256k
-
-/// This handler manually load request payload and parse json object
-async fn index_add(
-    mut pl: web::Payload,
-    pool: web::Data<Pool>,
-) -> Result<HttpResponse, Error> {
+/// This handler manually parse json object. Bytes object supports FromRequest trait (extractor)
+/// and could be loaded from request payload automatically
+async fn index_add(body: Bytes, pool: web::Data<Pool>) -> Result<HttpResponse, Error> {
     let mut body = BytesMut::new();
-    // pl.next() gets next item from asynchronous stream of byte chunks
-    // it resolves to `Option<Result<Bytes, Error>>` object
-    // `None` - indicaets end of stream
-    while let Some(chunk) = pl.next().await {
-        let chunk = chunk?;
-
-        // limit max size of in-memory payload
-        if (body.len() + chunk.len()) > MAX_SIZE {
-            return Err(error::ErrorBadRequest("overflow"));
-        }
-        body.extend_from_slice(&chunk);
-    }
-
-    // body is loaded, now we can deserialize serde-json
+    // body is loaded, now we can deserialize id with serde-json
     let r_obj = serde_json::from_slice::<MyUser>(&body);
 
     // Send to the db for create return response to peer
@@ -94,6 +77,7 @@ async fn index_add(
     }
 }
 
+/// This handler offloads json deserialization to actix-web's Json extrator
 async fn add2(
     item: web::Json<MyUser>,
     pool: web::Data<Pool>,
diff --git a/multipart/rust-toolchain b/multipart/rust-toolchain
deleted file mode 100644
index 7ff88f6e..00000000
--- a/multipart/rust-toolchain
+++ /dev/null
@@ -1 +0,0 @@
-nightly-2019-11-25
diff --git a/multipart/src/main.rs b/multipart/src/main.rs
index 13955574..dbc43c63 100644
--- a/multipart/src/main.rs
+++ b/multipart/src/main.rs
@@ -1,7 +1,8 @@
+use std::io::Write;
+
 use actix_multipart::Multipart;
 use actix_web::{middleware, web, App, Error, HttpResponse, HttpServer};
 use futures::StreamExt;
-use std::io::Write;
 
 async fn save_file(mut payload: Multipart) -> Result<HttpResponse, Error> {
     // iterate over multipart stream
@@ -10,15 +11,15 @@ async fn save_file(mut payload: Multipart) -> Result<HttpResponse, Error> {
         let content_type = field.content_disposition().unwrap();
         let filename = content_type.get_filename().unwrap();
         let filepath = format!("./tmp/{}", filename);
-        let mut f = std::fs::File::create(filepath).unwrap();
+        // File::create is blocking operation, use threadpool
+        let mut f = web::block(|| std::fs::File::create(filepath))
+            .await
+            .unwrap();
         // Field in turn is stream of *Bytes* object
         while let Some(chunk) = field.next().await {
             let data = chunk.unwrap();
-            let mut pos = 0;
-            while pos < data.len() {
-                let bytes_written = f.write(&data[pos..])?;
-                pos += bytes_written;
-            }
+            // filesystem operations are blocking, we have to use threadpool
+            f = web::block(move || f.write_all(&data).map(|_| f)).await?;
         }
     }
     Ok(HttpResponse::Ok().into())