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>; +#[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, -) -> Result { +/// 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) -> Result { let mut body = BytesMut::new(); - // pl.next() gets next item from asynchronous stream of byte chunks - // it resolves to `Option>` 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::(&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, pool: web::Data, 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 { // iterate over multipart stream @@ -10,15 +11,15 @@ async fn save_file(mut payload: Multipart) -> Result { 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())