fix multipart example

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)]
+/// This handler manually parse json object. Bytes object supports FromRequest trait (extractor)
-struct MyUser {
+/// and could be loaded from request payload automatically
-    name: String,
+async fn index_add(body: Bytes, pool: web::Data<Pool>) -> Result<HttpResponse, Error> {
-}
-
-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> {
     let mut body = BytesMut::new();
-    // pl.next() gets next item from asynchronous stream of byte chunks
+    // body is loaded, now we can deserialize id with serde-json
-    // 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
     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>,

multipart/rust-toolchain

@@ -1 +0,0 @@
-nightly-2019-11-25

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;
+            // filesystem operations are blocking, we have to use threadpool
-            while pos < data.len() {
+            f = web::block(move || f.write_all(&data).map(|_| f)).await?;
-                let bytes_written = f.write(&data[pos..])?;
-                pos += bytes_written;
-            }
         }
     }
     Ok(HttpResponse::Ok().into())