asonix/actix-form-data
1
0
Fork 0
Code Issues Pull requests Actions Projects Releases 1 Wiki Activity

Prepare actix-web 3.0 version

Browse source 
This is still broken because there's no 3.0-compatible multipart release
This commit is contained in:
asonix 2020-05-15 11:59:31 -05:00
parent a2f74fa7d7
commit 95ef512a93
8 changed files with 253 additions and 26103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25549
ACKNOWLEDGEMENTS
View file

File diff suppressed because it is too large Load diff

29
Cargo.toml
View file

@ -1,7 +1,7 @@
[package]
name = "actix-form-data"
description = "Multipart Form Data for Actix Web"
version = "0.4.0"
version = "0.5.0-alpha.0"
license = "GPL-3.0"
authors = ["asonix <asonix@asonix.dog>"]
repository = "https://git.asonix.dog/Aardwolf/actix-form-data.git"
@ -13,20 +13,19 @@ edition = "2018"
name = "form_data"
[dependencies]
actix-multipart = "0.1.0"
actix-rt = "0.2.2"
actix-threadpool = "0.1.0"
actix-web = "1.0.0"
bytes = "0.4.7"
failure = "0.1"
futures = "0.1.21"
http = "0.1.5"
log = "0.4.1"
mime = "0.3.5"
actix-http = "2.0.0-alpha.3"
actix-multipart = "0.2.0"
actix-rt = "1.1.1"
actix-web = "3.0.0-alpha.2"
bytes = "0.5.0"
futures = "0.3.4"
log = "0.4.8"
mime = "0.3.16"
thiserror = "1.0"
[dev-dependencies]
actix = "0.8.1"
env_logger = "0.6.0"
serde = "1.0"
anyhow = "1.0"
env_logger = "0.7.1"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
serde_derive = "1.0"
tokio = { version = "0.2.21", features = ["fs"] }

31
examples/simple.rs
View file

@ -5,29 +5,17 @@ use actix_web::{
web::{post, resource, Data},
App, HttpResponse, HttpServer,
};
use form_data::{handle_multipart, Error, Field, FilenameGenerator, Form};
use futures::Future;
use form_data::{handle_multipart, Error, Field, Form};
struct Gen;
async fn upload(mp: Multipart, state: Data<Form>) -> Result<HttpResponse, Error> {
let uploaded_content = handle_multipart(mp, state.get_ref().clone()).await?;
impl FilenameGenerator for Gen {
fn next_filename(&self, _: &mime::Mime) -> Option<PathBuf> {
let mut p = PathBuf::new();
p.push("examples/filename.png");
Some(p)
}
println!("Uploaded Content: {:?}", uploaded_content);
Ok(HttpResponse::Created().finish())
}
fn upload((mp, state): (Multipart, Data<Form>)) -> Box<Future<Item = HttpResponse, Error = Error>> {
Box::new(
handle_multipart(mp, state.get_ref().clone()).map(|uploaded_content| {
println!("Uploaded Content: {:?}", uploaded_content);
HttpResponse::Created().finish()
}),
)
}
fn main() -> Result<(), failure::Error> {
#[actix_rt::main]
async fn main() -> Result<(), anyhow::Error> {
let form = Form::new()
.field("Hey", Field::text())
.field(
@ -37,7 +25,7 @@ fn main() -> Result<(), failure::Error> {
.field("Two", Field::float())
.finalize(),
)
.field("files", Field::array(Field::file(Gen)));
.field("files", Field::array(Field::file()));
println!("{:?}", form);
@ -47,7 +35,8 @@ fn main() -> Result<(), failure::Error> {
.service(resource("/upload").route(post().to(upload)))
})
.bind("127.0.0.1:8080")?
.run()?;
.run()
.await?;
Ok(())
}

83
src/error.rs
View file

@ -18,7 +18,6 @@
*/
use std::{
io,
num::{ParseFloatError, ParseIntError},
string::FromUtf8Error,
};
@ -28,89 +27,57 @@ use actix_web::{
error::{PayloadError, ResponseError},
HttpResponse,
};
use bytes::Bytes;
use failure::Fail;
use futures::sync::mpsc::SendError;
#[derive(Debug, Fail)]
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[fail(display = "Error saving file, {}", _0)]
FsPool(#[cause] io::Error),
#[fail(display = "Error parsing payload, {}", _0)]
Payload(PayloadError),
#[fail(display = "Error in multipart creation, {}", _0)]
#[error("Error parsing payload, {0}")]
Payload(#[from] PayloadError),
#[error("Error in multipart creation, {0}")]
Multipart(MultipartError),
#[fail(display = "Failed to parse field, {}", _0)]
ParseField(#[cause] FromUtf8Error),
#[fail(display = "Failed to parse int, {}", _0)]
ParseInt(#[cause] ParseIntError),
#[fail(display = "Failed to parse float, {}", _0)]
ParseFloat(#[cause] ParseFloatError),
#[fail(display = "Failed to generate filename")]
GenFilename,
#[fail(display = "Bad Content-Type")]
#[error("Failed to parse field, {0}")]
ParseField(#[from] FromUtf8Error),
#[error("Failed to parse int, {0}")]
ParseInt(#[from] ParseIntError),
#[error("Failed to parse float, {0}")]
ParseFloat(#[from] ParseFloatError),
#[error("Bad Content-Type")]
ContentType,
#[fail(display = "Bad Content-Disposition")]
#[error("Bad Content-Disposition")]
ContentDisposition,
#[fail(display = "Failed to make directory for upload")]
MkDir,
#[fail(display = "Failed to parse field name")]
#[error("Failed to parse field name")]
Field,
#[fail(display = "Could not write file")]
WriteFile,
#[fail(display = "Too many fields in request")]
#[error("Too many fields in request")]
FieldCount,
#[fail(display = "Field too large")]
#[error("Field too large")]
FieldSize,
#[fail(display = "Found field with unexpected name or type")]
#[error("Found field with unexpected name or type")]
FieldType,
#[fail(display = "Failed to parse filename")]
#[error("Failed to parse filename")]
Filename,
#[fail(display = "Too many files in request")]
#[error("Too many files in request")]
FileCount,
#[fail(display = "File too large")]
#[error("File too large")]
FileSize,
}
impl From<MultipartError> for Error {
fn from(e: MultipartError) -> Self {
Error::Multipart(e)
}
}
impl From<PayloadError> for Error {
fn from(e: PayloadError) -> Self {
Error::Payload(e)
}
}
impl From<io::Error> for Error {
fn from(e: io::Error) -> Self {
Error::FsPool(e)
}
}
impl From<SendError<Bytes>> for Error {
fn from(_: SendError<Bytes>) -> Self {
Error::WriteFile
fn from(m: MultipartError) -> Self {
Error::Multipart(m)
}
}
impl ResponseError for Error {
fn error_response(&self) -> HttpResponse {
match *self {
Error::FsPool(_) => HttpResponse::InternalServerError().finish(),
Error::Payload(ref e) => ResponseError::error_response(e),
Error::Multipart(ref e) => ResponseError::error_response(e),
Error::ParseField(_) | Error::ParseInt(_) | Error::ParseFloat(_) => {
HttpResponse::BadRequest().finish()
}
Error::GenFilename | Error::MkDir => HttpResponse::InternalServerError().finish(),
Error::Multipart(_)
| Error::ParseField(_)
| Error::ParseInt(_)
| Error::ParseFloat(_) => HttpResponse::BadRequest().finish(),
Error::ContentType
| Error::ContentDisposition
| Error::Field
| Error::FieldCount
| Error::WriteFile
| Error::FieldSize
| Error::FieldType
| Error::Filename

113
src/file_future.rs
View file

@ -1,113 +0,0 @@
use bytes::{Bytes, BytesMut};
use failure::Fail;
use futures::{
sync::mpsc::{channel, SendError},
task, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream,
};
use std::{
fs::File,
io::{Error, Write},
path::Path,
};
#[derive(Clone, Debug, Fail)]
#[fail(display = "Error in Channel")]
struct ChannelError;
pub fn write(
filename: impl AsRef<Path> + Clone + Send + 'static,
) -> impl Sink<SinkItem = Bytes, SinkError = SendError<Bytes>> {
let (tx, rx) = channel(50);
actix_rt::spawn(
actix_threadpool::run(move || {
CreateFuture::new(filename.clone())
.from_err()
.and_then(|file| {
rx.map_err(|_| failure::Error::from(ChannelError))
.forward(WriteSink::new(file))
})
.wait()
})
.map_err(|_| ())
.map(|_| ()),
);
tx
}
struct CreateFuture<P>(P)
where
P: AsRef<Path> + Clone;
impl<P> CreateFuture<P>
where
P: AsRef<Path> + Clone,
{
fn new(path: P) -> Self {
CreateFuture(path)
}
}
impl<P> Future for CreateFuture<P>
where
P: AsRef<Path> + Clone,
{
type Item = File;
type Error = Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
File::create(self.0.clone()).map(Async::Ready)
}
}
struct WriteSink {
buffer: BytesMut,
file: File,
}
impl WriteSink {
fn new(file: File) -> Self {
WriteSink {
buffer: BytesMut::new(),
file,
}
}
}
impl Sink for WriteSink {
type SinkItem = Bytes;
type SinkError = Error;
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
if let Async::NotReady = self.poll_complete()? {
return Ok(AsyncSink::NotReady(item));
}
self.buffer = BytesMut::new();
self.buffer.extend_from_slice(&item);
self.poll_complete()?;
Ok(AsyncSink::Ready)
}
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
if self.buffer.is_empty() {
return Ok(Async::Ready(()));
}
let written = self.file.write(&self.buffer)?;
if written == 0 {
return Err(Error::last_os_error());
}
self.buffer.advance(written);
if self.buffer.is_empty() {
Ok(Async::Ready(()))
} else {
task::current().notify();
Ok(Async::NotReady)
}
}
}

34
src/lib.rs
View file

@ -25,8 +25,6 @@
//! # Example
//!
//!```rust
//! use std::path::PathBuf;
//!
//! use actix_multipart::Multipart;
//! use actix_web::{
//! web::{post, resource, Data},
@ -34,17 +32,8 @@
//! };
//! use form_data::{handle_multipart, Error, Field, FilenameGenerator, Form};
//! use futures::Future;
//!
//! struct Gen;
//!
//! impl FilenameGenerator for Gen {
//! fn next_filename(&self, _: &mime::Mime) -> Option<PathBuf> {
//! let mut p = PathBuf::new();
//! p.push("examples/filename.png");
//! Some(p)
//! }
//! }
//!
//!
//!
//! fn upload((mp, state): (Multipart, Data<Form>)) -> Box<Future<Item = HttpResponse, Error = Error>> {
//! Box::new(
//! handle_multipart(mp, state.get_ref().clone()).map(|uploaded_content| {
@ -53,7 +42,7 @@
//! }),
//! )
//! }
//!
//!
//! fn main() -> Result<(), failure::Error> {
//! let form = Form::new()
//! .field("Hey", Field::text())
@ -65,9 +54,9 @@
//! .finalize(),
//! )
//! .field("files", Field::array(Field::file(Gen)));
//!
//!
//! println!("{:?}", form);
//!
//!
//! HttpServer::new(move || {
//! App::new()
//! .data(form.clone())
@ -75,24 +64,13 @@
//! })
//! .bind("127.0.0.1:8080")?;
//! // .run()?;
//!
//!
//! Ok(())
//! }
//!```
use std::path::PathBuf;
mod error;
mod file_future;
mod types;
mod upload;
pub use self::{error::Error, types::*, upload::handle_multipart};
/// A trait for types that produce filenames for uploade files
///
/// Currently, the mime type provided to the `next_filename` method is guessed from the uploaded
/// file's original filename, so relying on this to be 100% accurate is probably a bad idea.
pub trait FilenameGenerator: Send + Sync {
fn next_filename(&self, mime_type: &mime::Mime) -> Option<PathBuf>;
}

198
src/types.rs
View file

@ -17,17 +17,31 @@
* along with Actix Form Data. If not, see <http://www.gnu.org/licenses/>.
*/
use crate::Error;
use bytes::Bytes;
use futures::Stream;
use log::trace;
use mime::Mime;
use std::{
collections::{HashMap, VecDeque},
fmt,
path::PathBuf,
sync::Arc,
pin::Pin,
};
use bytes::Bytes;
use log::trace;
pub struct FileStream {
pub filename: String,
pub content_type: Mime,
pub stream: Pin<Box<dyn Stream<Item = Result<Bytes, Error>>>>,
}
use crate::FilenameGenerator;
impl fmt::Debug for FileStream {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("FileStream")
.field("filename", &self.filename)
.field("content_type", &self.content_type)
.finish()
}
}
/// The result of a succesfull parse through a given multipart stream.
///
@ -54,33 +68,37 @@ use crate::FilenameGenerator;
/// _ => (),
/// }
/// ```
#[derive(Clone, Debug, PartialEq)]
#[derive(Debug)]
pub enum Value {
Map(HashMap<String, Value>),
Array(Vec<Value>),
File(String, PathBuf),
File(FileStream),
Bytes(Bytes),
Text(String),
Int(i64),
Float(f64),
Bytes(Bytes),
}
impl Value {
pub(crate) fn merge(&mut self, rhs: Self) {
match (self, rhs) {
(&mut Value::Map(ref mut hm), Value::Map(ref other)) => {
other.into_iter().fold(hm, |hm, (key, value)| {
if let Some(v) = hm.get_mut(key) {
v.merge(value.clone());
} else {
hm.insert(key.to_owned(), value.clone());
}
match self {
Value::Map(ref mut hm) => {
if let Value::Map(other) = rhs {
other.into_iter().fold(hm, |hm, (key, value)| {
if let Some(v) = hm.get_mut(&key) {
v.merge(value);
} else {
hm.insert(key.to_owned(), value);
}
hm
});
hm
});
}
}
(&mut Value::Array(ref mut v), Value::Array(ref other)) => {
v.extend(other.clone());
Value::Array(ref mut v) => {
if let Value::Array(other) = rhs {
v.extend(other);
}
}
_ => (),
}
@ -100,9 +118,16 @@ impl Value {
}
}
pub fn file(self) -> Option<(String, PathBuf)> {
pub fn file(self) -> Option<FileStream> {
match self {
Value::File(name, path) => Some((name, path)),
Value::File(file_stream) => Some(file_stream),
_ => None,
}
}
pub fn bytes(self) -> Option<Bytes> {
match self {
Value::Bytes(bytes) => Some(bytes),
_ => None,
}
}
@ -127,26 +152,16 @@ impl Value {
_ => None,
}
}
pub fn bytes(self) -> Option<Bytes> {
match self {
Value::Bytes(bytes) => Some(bytes),
_ => None,
}
}
}
impl From<MultipartContent> for Value {
fn from(mc: MultipartContent) -> Self {
match mc {
MultipartContent::File {
filename,
stored_as,
} => Value::File(filename, stored_as),
MultipartContent::File(file_stream) => Value::File(file_stream),
MultipartContent::Bytes(bytes) => Value::Bytes(bytes),
MultipartContent::Text(string) => Value::Text(string),
MultipartContent::Int(i) => Value::Int(i),
MultipartContent::Float(f) => Value::Float(f),
MultipartContent::Bytes(b) => Value::Bytes(b),
}
}
}
@ -155,24 +170,24 @@ impl From<MultipartContent> for Value {
#[derive(Clone)]
pub enum Field {
Array(Array),
File(Arc<FilenameGenerator>),
Map(Map),
File,
Bytes,
Int,
Float,
Text,
Bytes,
}
impl fmt::Debug for Field {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match *self {
Field::Array(ref arr) => write!(f, "Array({:?})", arr),
Field::File(_) => write!(f, "File(filename_generator)"),
Field::Map(ref map) => write!(f, "Map({:?})", map),
Field::File => write!(f, "File"),
Field::Bytes => write!(f, "Bytes"),
Field::Int => write!(f, "Int"),
Field::Float => write!(f, "Float"),
Field::Text => write!(f, "Text"),
Field::Bytes => write!(f, "Bytes"),
}
}
}
@ -186,41 +201,30 @@ impl Field {
///
/// # Example
/// ```rust
/// # extern crate mime;
/// # extern crate form_data;
/// # use std::path::{Path, PathBuf};
/// # use form_data::{Form, Field, FilenameGenerator};
///
/// struct Gen;
///
/// impl FilenameGenerator for Gen {
/// fn next_filename(&self, _: &mime::Mime) -> Option<PathBuf> {
/// Some(AsRef::<Path>::as_ref("path.png").to_owned())
/// }
/// }
///
/// fn main() {
/// let name_generator = Gen;
/// let form = Form::new()
/// .field("file-field", Field::file(name_generator));
/// }
/// # use form_data::{Form, Field};
/// #
/// let form = Form::new().field("file-field", Field::file());
/// ```
pub fn file<T>(gen: T) -> Self
where
T: FilenameGenerator + 'static,
{
Field::File(Arc::new(gen))
pub fn file() -> Self {
Field::File
}
/// Add a Bytes field to a form
///
/// # Example
/// ```rust
/// # use form_data::{Form, Field};
/// let form = Form::new().field("text-field", Field::bytes());
pub fn bytes() -> Self {
Field::Bytes
}
/// Add a Text field to a form
///
/// # Example
/// ```rust
/// # extern crate form_data;
/// # use form_data::{Form, Field};
/// # fn main() {
/// let form = Form::new().field("text-field", Field::text());
/// # }
pub fn text() -> Self {
Field::Text
}
@ -229,11 +233,8 @@ impl Field {
///
/// # Example
/// ```rust
/// # extern crate form_data;
/// # use form_data::{Form, Field};
/// # fn main() {
/// let form = Form::new().field("int-field", Field::int());
/// # }
/// ```
pub fn int() -> Self {
Field::Int
@ -243,30 +244,13 @@ impl Field {
///
/// # Example
/// ```rust
/// # extern crate form_data;
/// # use form_data::{Form, Field};
/// # fn main() {
/// let form = Form::new().field("float-field", Field::float());
/// # }
/// ```
pub fn float() -> Self {
Field::Float
}
/// Add a Bytes field to a form
///
/// # Example
/// ```rust
/// # extern crate form_data;
/// # use form_data::{Form, Field};
/// # fn main() {
/// let form = Form::new().field("bytes-field", Field::bytes());
/// # }
/// ```
pub fn bytes() -> Self {
Field::Bytes
}
/// Add an Array to a form
///
/// # Example
@ -311,9 +295,16 @@ impl Field {
match *self {
Field::Array(ref arr) => arr.valid_field(name),
Field::Map(ref map) => map.valid_field(name),
Field::File(ref gen) => {
Field::File => {
if name.is_empty() {
Some(FieldTerminator::File(Arc::clone(gen)))
Some(FieldTerminator::File)
} else {
None
}
}
Field::Bytes => {
if name.is_empty() {
Some(FieldTerminator::Bytes)
} else {
None
}
@ -339,13 +330,6 @@ impl Field {
None
}
}
Field::Bytes => {
if name.is_empty() {
Some(FieldTerminator::Bytes)
} else {
None
}
}
}
}
}
@ -438,24 +422,12 @@ impl Map {
///
/// # Example
/// ```rust
/// # extern crate mime;
/// # extern crate form_data;
/// # use std::path::{Path, PathBuf};
/// # use form_data::{Form, Field, FilenameGenerator};
/// # struct Gen;
/// # impl FilenameGenerator for Gen {
/// # fn next_filename(&self, _: &mime::Mime) -> Option<PathBuf> {
/// # Some(AsRef::<Path>::as_ref("path.png").to_owned())
/// # }
/// # }
/// # fn main() {
/// # let name_generator = Gen;
/// # use form_data::{Form, Field};
/// let form = Form::new()
/// .field("field-name", Field::text())
/// .field("second-field", Field::int())
/// .field("third-field", Field::float())
/// .field("fourth-field", Field::bytes())
/// .field("fifth-field", Field::file(name_generator))
/// .field("fifth-field", Field::file())
/// .field(
/// "map-field",
/// Field::map()
@ -467,7 +439,6 @@ impl Map {
/// "array-field",
/// Field::array(Field::text())
/// );
/// # }
/// ```
#[derive(Clone)]
pub struct Form {
@ -577,7 +548,7 @@ impl NamePart {
#[derive(Clone)]
pub(crate) enum FieldTerminator {
File(Arc<FilenameGenerator>),
File,
Bytes,
Int,
Float,
@ -587,7 +558,7 @@ pub(crate) enum FieldTerminator {
impl fmt::Debug for FieldTerminator {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match *self {
FieldTerminator::File(_) => write!(f, "File(filename_generator)"),
FieldTerminator::File => write!(f, "File"),
FieldTerminator::Bytes => write!(f, "Bytes"),
FieldTerminator::Int => write!(f, "Int"),
FieldTerminator::Float => write!(f, "Float"),
@ -599,12 +570,9 @@ impl fmt::Debug for FieldTerminator {
pub(crate) type MultipartHash = (Vec<NamePart>, MultipartContent);
pub(crate) type MultipartForm = Vec<MultipartHash>;
#[derive(Clone, Debug, PartialEq)]
#[derive(Debug)]
pub(crate) enum MultipartContent {
File {
filename: String,
stored_as: PathBuf,
},
File(FileStream),
Bytes(Bytes),
Text(String),
Int(i64),

319
src/upload.rs
View file

@ -17,31 +17,25 @@
* along with Actix Form Data. If not, see <http://www.gnu.org/licenses/>.
*/
use crate::{
error::Error,
types::{
ContentDisposition, FieldTerminator, FileStream, Form, MultipartContent, MultipartForm,
MultipartHash, NamePart, Value,
},
};
use bytes::BytesMut;
use futures::stream::StreamExt;
use log::trace;
use std::{
collections::HashMap,
fs::DirBuilder,
path::{Path, PathBuf},
path::Path,
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
};
use bytes::BytesMut;
use futures::{
future::{result, Either},
Future, Stream,
};
use log::trace;
use crate::{
error::Error,
types::{
self, ContentDisposition, MultipartContent, MultipartForm, MultipartHash, NamePart, Value,
},
FilenameGenerator,
};
fn consolidate(mf: MultipartForm) -> Value {
mf.into_iter().fold(
Value::Map(HashMap::new()),
@ -102,221 +96,138 @@ fn parse_content_disposition(field: &actix_multipart::Field) -> ContentDispositi
}
}
#[cfg(unix)]
fn build_dir(stored_dir: PathBuf) -> Result<(), Error> {
use std::os::unix::fs::DirBuilderExt;
DirBuilder::new()
.recursive(true)
.mode(0o755)
.create(stored_dir)
.map_err(|_| Error::MkDir)
}
#[cfg(not(unix))]
fn build_dir(stored_dir: PathBuf) -> Result<(), Error> {
DirBuilder::new()
.recursive(true)
.create(stored_dir)
.map_err(|_| Error::MkDir)
}
fn handle_file_upload(
async fn handle_file_upload(
field: actix_multipart::Field,
gen: Arc<FilenameGenerator>,
filename: Option<String>,
form: types::Form,
) -> Box<Future<Item = MultipartContent, Error = Error>> {
let filename = match filename {
Some(filename) => filename,
None => return Box::new(result(Err(Error::Filename))),
};
form: Form,
) -> Result<MultipartContent, Error> {
let filename = filename.ok_or(Error::Filename)?;
let path: &Path = filename.as_ref();
let filename = path.file_name().and_then(|filename| filename.to_str());
let filename = if let Some(filename) = filename {
filename.to_owned()
} else {
return Box::new(result(Err(Error::Filename)));
};
let filename = filename.ok_or(Error::Filename)?.to_owned();
let stored_as = match gen.next_filename(field.content_type()) {
Some(file_path) => file_path,
None => return Box::new(result(Err(Error::GenFilename))),
};
let file_size = Arc::new(AtomicUsize::new(0));
let mut stored_dir = stored_as.clone();
stored_dir.pop();
let content_type = field.content_type().clone();
let mkdir_fut = actix_threadpool::run(move || build_dir(stored_dir.clone()));
Ok(MultipartContent::File(FileStream {
filename,
content_type,
stream: Box::pin(field.then(move |res| {
let form = form.clone();
let file_size = file_size.clone();
async move {
match res {
Ok(bytes) => {
let size = file_size.fetch_add(bytes.len(), Ordering::Relaxed);
let counter = Arc::new(AtomicUsize::new(0));
if size + bytes.len() > form.max_file_size {
return Err(Error::FileSize);
}
Box::new(mkdir_fut.map_err(|_| Error::MkDir).and_then(move |_| {
let write = crate::file_future::write(stored_as.clone());
field
.map_err(Error::Multipart)
.and_then(move |bytes| {
let size = counter.fetch_add(bytes.len(), Ordering::Relaxed) + bytes.len();
if size > form.max_file_size {
Err(Error::FileSize)
} else {
Ok(bytes)
Ok(bytes)
}
Err(e) => Err(Error::from(e)),
}
})
.forward(write)
.map(move |_| MultipartContent::File {
filename,
stored_as,
})
}
})),
}))
}
fn handle_form_data(
field: actix_multipart::Field,
term: types::FieldTerminator,
form: types::Form,
) -> Box<Future<Item = MultipartContent, Error = Error>> {
async fn handle_form_data(
mut field: actix_multipart::Field,
term: FieldTerminator,
form: Form,
) -> Result<MultipartContent, Error> {
trace!("In handle_form_data, term: {:?}", term);
let term2 = term.clone();
let mut bytes = BytesMut::new();
Box::new(
field
.from_err()
.fold(BytesMut::new(), move |mut acc, bytes| {
if acc.len() + bytes.len() < form.max_field_size {
acc.extend(bytes);
Ok(acc)
} else {
Err(Error::FieldSize)
}
})
.and_then(move |bytes| match term {
types::FieldTerminator::Bytes => Ok(MultipartContent::Bytes(bytes.freeze())),
_ => String::from_utf8(bytes.to_vec())
.map_err(Error::ParseField)
.map(MultipartContent::Text),
})
.and_then(move |content| {
trace!("Matching: {:?}", content);
match content {
types::MultipartContent::Text(string) => match term2 {
types::FieldTerminator::File(_) => Err(Error::FieldType),
types::FieldTerminator::Bytes => Err(Error::FieldType),
types::FieldTerminator::Float => string
.parse::<f64>()
.map(MultipartContent::Float)
.map_err(Error::ParseFloat),
types::FieldTerminator::Int => string
.parse::<i64>()
.map(MultipartContent::Int)
.map_err(Error::ParseInt),
types::FieldTerminator::Text => Ok(MultipartContent::Text(string)),
},
b @ types::MultipartContent::Bytes(_) => Ok(b),
_ => Err(Error::FieldType),
}
}),
)
while let Some(res) = field.next().await {
let b = res?;
if bytes.len() + b.len() > form.max_field_size {
return Err(Error::FieldSize);
}
bytes.extend(b);
}
if let FieldTerminator::Bytes = term {
return Ok(MultipartContent::Bytes(bytes.freeze()));
}
let s = String::from_utf8(bytes.to_vec()).map_err(Error::ParseField)?;
match term {
FieldTerminator::Bytes | FieldTerminator::File => {
return Err(Error::FieldType);
}
FieldTerminator::Text => Ok(MultipartContent::Text(s)),
FieldTerminator::Float => s
.parse()
.map_err(Error::ParseFloat)
.map(MultipartContent::Float),
FieldTerminator::Int => s
.parse()
.map_err(Error::ParseInt)
.map(MultipartContent::Int),
}
}
fn handle_stream_field(
async fn handle_stream_field(
field: actix_multipart::Field,
form: types::Form,
) -> Box<Future<Item = MultipartHash, Error = Error>> {
form: Form,
) -> Result<MultipartHash, Error> {
let content_disposition = parse_content_disposition(&field);
let name = match content_disposition.name {
Some(name) => name,
None => return Box::new(result(Err(Error::Field))),
let name = content_disposition.name.ok_or(Error::Field)?;
let name = parse_multipart_name(name)?;
let term = form
.valid_field(name.iter().cloned().collect())
.ok_or(Error::FieldType)?;
let content = match term {
FieldTerminator::File => {
handle_file_upload(field, content_disposition.filename, form).await?
}
term => handle_form_data(field, term, form.clone()).await?,
};
let name = match parse_multipart_name(name) {
Ok(name) => name,
Err(e) => return Box::new(result(Err(e))),
};
let term = match form.valid_field(name.iter().cloned().collect()) {
Some(term) => term,
None => return Box::new(result(Err(Error::FieldType))),
};
let fut = match term {
types::FieldTerminator::File(gen) => Either::A(handle_file_upload(
field,
gen,
content_disposition.filename,
form,
)),
term => Either::B(handle_form_data(field, term, form)),
};
Box::new(fut.map(|content| (name, content)))
}
fn handle_stream(
m: actix_multipart::Multipart,
form: types::Form,
) -> Box<Stream<Item = MultipartHash, Error = Error>> {
Box::new(
m.map_err(Error::from)
.map(move |field| {
handle_stream_field(field, form.clone())
.map(From::from)
.into_stream()
})
.flatten(),
)
Ok((name, content))
}
/// Handle multipart streams from Actix Web
pub fn handle_multipart(
m: actix_multipart::Multipart,
form: types::Form,
) -> Box<Future<Item = Value, Error = Error>> {
Box::new(
handle_stream(m, form.clone())
.fold(
(Vec::new(), 0, 0),
move |(mut acc, file_count, field_count), (name, content)| match content {
MultipartContent::File {
filename,
stored_as,
} => {
let file_count = file_count + 1;
pub async fn handle_multipart(
mut m: actix_multipart::Multipart,
form: Form,
) -> Result<Value, Error> {
let mut multipart_form = Vec::new();
let mut file_count: u32 = 0;
let mut field_count: u32 = 0;
if file_count < form.max_files {
acc.push((
name,
MultipartContent::File {
filename,
stored_as,
},
));
while let Some(res) = m.next().await {
let field = res?;
let (name_parts, content) = handle_stream_field(field, form.clone()).await?;
Ok((acc, file_count, field_count))
} else {
Err(Error::FileCount)
}
}
b @ MultipartContent::Bytes(_)
| b @ MultipartContent::Text(_)
| b @ MultipartContent::Float(_)
| b @ MultipartContent::Int(_) => {
let field_count = field_count + 1;
match content {
MultipartContent::File(_) => {
file_count += 1;
if file_count >= form.max_files {
return Err(Error::FileCount);
}
}
_ => {
field_count += 1;
if field_count >= form.max_fields {
return Err(Error::FieldCount);
}
}
}
if field_count < form.max_fields {
acc.push((name, b));
multipart_form.push((name_parts, content));
}
Ok((acc, file_count, field_count))
} else {
Err(Error::FieldCount)
}
}
},
)
.map(|(multipart_form, _, _)| consolidate(multipart_form)),
)
Ok(consolidate(multipart_form))
}