.forgejo/workflows/check.yaml

@@ -1,55 +0,0 @@
-on: 
-  push:
-    branches:
-      - '*'
-  pull_request:
-    branches:
-      - main
-
-jobs:
-  clippy:
-    runs-on: base-image
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Clippy
-        run: |
-          cargo clippy --no-default-features -- -D warnings
-
-  tests:
-    runs-on: base-image
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Test
-        run: cargo test
-
-  check:
-    strategy:
-      fail-fast: false
-      matrix:
-        target:
-          - x86_64-unknown-linux-musl
-          - armv7-unknown-linux-musleabihf
-          - aarch64-unknown-linux-musl
-    runs-on: base-image
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Debug builds
-        run: cargo zigbuild --target ${{ matrix.target }}

.forgejo/workflows/publish.yaml

@@ -1,78 +0,0 @@
-on:
-  push:
-    tags:
-      - 'v*.*.*'
-
-jobs:
-  clippy:
-    runs-on: base-image
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Clippy
-        run: |
-          cargo clippy --no-default-features -- -D warnings
-
-  tests:
-    runs-on: base-image
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Test
-        run: cargo test
-
-  build:
-    needs:
-      - clippy
-      - tests
-    runs-on: base-image
-    strategy:
-      fail-fast: false
-      matrix:
-        target:
-          - x86_64-unknown-linux-musl
-          - armv7-unknown-linux-musleabihf
-          - aarch64-unknown-linux-musl
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Compile actix-form-data
-        run: cargo zigbuild --target ${{ matrix.target }}
-
-  publish-forgejo:
-    needs: [build]
-    runs-on: base-image
-    steps:
-      - uses: actions/forgejo-release@v1
-        with:
-          direction: upload
-          token: ${{ secrets.GITHUB_TOKEN }}
-
-  publish-crate:
-    needs: [build]
-    runs-on: base-image
-    steps:
-      -
-        name: Checkout actix-form-data
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Publish Crate
-        run: cargo publish --token ${{ secrets.CRATES_IO_TOKEN }}

.gitignore

@@ -2,5 +2,3 @@
 **/*.rs.bk
 Cargo.lock
 /examples/filename*.png
-/.envrc
-/.direnv

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "actix-form-data"
 description = "Multipart Form Data for Actix Web"
-version = "0.7.0-beta.7"
+version = "0.7.0-beta.0"
 license = "GPL-3.0"
 authors = ["asonix <asonix@asonix.dog>"]
 repository = "https://git.asonix.dog/asonix/actix-form-data.git"
@@ -10,21 +10,19 @@ keywords = ["actix", "form-data", "multipart", "async"]
 edition = "2021"
 
 [dependencies]
-actix-multipart = { version = "0.6.0", default-features = false }
+actix-multipart = "0.4.0"
+actix-rt = "2.5.0"
 actix-web = { version = "4.0.0", default-features = false }
-futures-core = "0.3.28"
+futures-util = "0.3.17"
 mime = "0.3.16"
-streem = "0.2.0"
 thiserror = "1.0"
-tokio = { version = "1", default-features = false, features = ["macros", "sync"] }
+tokio = { version = "1", default-features = false, features = ["sync"] }
 tracing = "0.1.15"
 
 [dev-dependencies]
-actix-rt = "2.5.0"
-async-fs = "2.1.0"
+async-fs = "1.2.1"
 anyhow = "1.0"
-futures-lite = "2.1.0"
-futures-util = "0.3.17"
+futures-lite = "1.4.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 thiserror = "1.0"

examples/simple.rs

@@ -11,8 +11,8 @@ impl FormData for UploadedContent {
     type Item = ();
     type Error = Error;
 
-    fn form(_: &HttpRequest) -> Result<Form<Self::Item, Self::Error>, Self::Error> {
-        Ok(Form::new()
+    fn form(_: &HttpRequest) -> Form<Self::Item, Self::Error> {
+        Form::new()
             .field("Hey", Field::text())
             .field(
                 "Hi",
@@ -29,7 +29,7 @@ impl FormData for UploadedContent {
                     }
                     Ok(()) as Result<(), Error>
                 })),
-            ))
+            )
     }
 
     fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>

examples/upload.rs

@@ -61,11 +61,11 @@ impl FormData for UploadedContent {
     type Item = PathBuf;
     type Error = Errors;
 
-    fn form(req: &HttpRequest) -> Result<Form<Self::Item, Self::Error>, Self::Error> {
+    fn form(req: &HttpRequest) -> Form<Self::Item, Self::Error> {
         let file_count = req.app_data::<Arc<AtomicUsize>>().expect("Set config");
         let file_count = Arc::clone(file_count);
 
-        Ok(Form::new()
+        Form::new()
             .field("Hey", Field::text())
             .field(
                 "Hi",
@@ -85,7 +85,7 @@ impl FormData for UploadedContent {
                             .map_err(Errors::from)
                     }
                 })),
-            ))
+            )
     }
 
     fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>

flake.lock

@@ -1,61 +0,0 @@
-{
-  "nodes": {
-    "flake-utils": {
-      "inputs": {
-        "systems": "systems"
-      },
-      "locked": {
-        "lastModified": 1701680307,
-        "narHash": "sha256-kAuep2h5ajznlPMD9rnQyffWG8EM/C73lejGofXvdM8=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "4022d587cbbfd70fe950c1e2083a02621806a725",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
-    "nixpkgs": {
-      "locked": {
-        "lastModified": 1702151865,
-        "narHash": "sha256-9VAt19t6yQa7pHZLDbil/QctAgVsA66DLnzdRGqDisg=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "666fc80e7b2afb570462423cb0e1cf1a3a34fedd",
-        "type": "github"
-      },
-      "original": {
-        "owner": "NixOS",
-        "ref": "nixos-unstable",
-        "repo": "nixpkgs",
-        "type": "github"
-      }
-    },
-    "root": {
-      "inputs": {
-        "flake-utils": "flake-utils",
-        "nixpkgs": "nixpkgs"
-      }
-    },
-    "systems": {
-      "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-systems",
-        "repo": "default",
-        "type": "github"
-      }
-    }
-  },
-  "root": "root",
-  "version": 7
-}

flake.nix

@@ -1,27 +0,0 @@
-{
-  description = "A very basic flake";
-
-  inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
-    flake-utils.url = "github:numtide/flake-utils";
-  };
-
-  outputs = { self, nixpkgs, flake-utils }:
-    flake-utils.lib.eachDefaultSystem (system:
-      let
-        pkgs = import nixpkgs {
-          inherit system;
-        };
-      in
-      {
-        packages.default = pkgs.hello;
-
-        devShell = with pkgs; mkShell {
-          nativeBuildInputs = [ cargo cargo-outdated cargo-zigbuild clippy gcc protobuf rust-analyzer rustc rustfmt ];
-
-          RUST_SRC_PATH = "${pkgs.rust.packages.stable.rustPlatform.rustLibSrc}";
-        };
-
-        formatter = pkgs.nixpkgs-fmt;
-      });
-}

src/error.rs

@@ -19,11 +19,12 @@
 
 use std::{
     num::{ParseFloatError, ParseIntError},
-    str::Utf8Error,
+    string::FromUtf8Error,
 };
 
+use actix_multipart::MultipartError;
 use actix_web::{
-    error::{ParseError, PayloadError, ResponseError},
+    error::{PayloadError, ResponseError},
     http::StatusCode,
     HttpResponse,
 };
@@ -33,9 +34,9 @@ pub enum Error {
     #[error("Error parsing payload")]
     Payload(#[from] PayloadError),
     #[error("Error in multipart creation")]
-    Multipart(#[from] MultipartError),
+    Multipart(MultipartError),
     #[error("Failed to parse field")]
-    ParseField(#[from] Utf8Error),
+    ParseField(#[from] FromUtf8Error),
     #[error("Failed to parse int")]
     ParseInt(#[from] ParseIntError),
     #[error("Failed to parse float")]
@@ -58,85 +59,11 @@ pub enum Error {
     FileCount,
     #[error("File too large")]
     FileSize,
-    #[error("Task panicked")]
-    Panicked,
 }
 
-#[derive(Debug, thiserror::Error)]
-pub enum MultipartError {
-    #[error("No Content-Disposition `form-data` header")]
-    NoContentDisposition,
-    #[error("No Content-Type header found")]
-    NoContentType,
-    #[error("Cannot parse Content-Type header")]
-    ParseContentType,
-    #[error("Multipart boundary is not found")]
-    Boundary,
-    #[error("Nested multipart is not supported")]
-    Nested,
-    #[error("Multipart stream is incomplete")]
-    Incomplete,
-    #[error("Failed parsing")]
-    Parse(#[source] ParseError),
-    #[error("Multipart stream is not consumed")]
-    NotConsumed,
-    #[error("An error occured processing field `{field_name}`: `{zource}`")]
-    Field { field_name: String, zource: String },
-    #[error("Duplicate field found for: `{0}")]
-    DuplicateField(String),
-    #[error("Field with name `{0}` is required")]
-    MissingField(String),
-    #[error("Unsupported field `{0}`")]
-    UnsupportedField(String),
-    #[error("Unknown error occured: {0}")]
-    Unknown(String),
-}
-
-impl From<actix_multipart::MultipartError> for Error {
-    fn from(value: actix_multipart::MultipartError) -> Self {
-        match value {
-            actix_multipart::MultipartError::NoContentDisposition => {
-                Error::Multipart(MultipartError::NoContentDisposition)
-            }
-            actix_multipart::MultipartError::NoContentType => {
-                Error::Multipart(MultipartError::NoContentType)
-            }
-            actix_multipart::MultipartError::ParseContentType => {
-                Error::Multipart(MultipartError::ParseContentType)
-            }
-            actix_multipart::MultipartError::Boundary => Error::Multipart(MultipartError::Boundary),
-            actix_multipart::MultipartError::Nested => Error::Multipart(MultipartError::Nested),
-            actix_multipart::MultipartError::Incomplete => {
-                Error::Multipart(MultipartError::Incomplete)
-            }
-            actix_multipart::MultipartError::Parse(e) => Error::Multipart(MultipartError::Parse(e)),
-            actix_multipart::MultipartError::Payload(e) => Error::Payload(e),
-            actix_multipart::MultipartError::NotConsumed => {
-                Error::Multipart(MultipartError::NotConsumed)
-            }
-            actix_multipart::MultipartError::Field { field_name, source } => {
-                Error::Multipart(MultipartError::Field {
-                    field_name,
-                    zource: source.to_string(),
-                })
-            }
-            actix_multipart::MultipartError::DuplicateField(s) => {
-                Error::Multipart(MultipartError::DuplicateField(s))
-            }
-            actix_multipart::MultipartError::MissingField(s) => {
-                Error::Multipart(MultipartError::MissingField(s))
-            }
-            actix_multipart::MultipartError::UnsupportedField(s) => {
-                Error::Multipart(MultipartError::UnsupportedField(s))
-            }
-            e => Error::Multipart(MultipartError::Unknown(e.to_string())),
-        }
-    }
-}
-
-impl From<tokio::task::JoinError> for Error {
-    fn from(_: tokio::task::JoinError) -> Self {
-        Self::Panicked
+impl From<MultipartError> for Error {
+    fn from(m: MultipartError) -> Self {
+        Error::Multipart(m)
     }
 }
 
@@ -151,7 +78,6 @@ impl ResponseError for Error {
     fn error_response(&self) -> HttpResponse {
         match *self {
             Error::Payload(ref e) => e.error_response(),
-            Error::Panicked => HttpResponse::InternalServerError().finish(),
             Error::Multipart(_)
             | Error::ParseField(_)
             | Error::ParseInt(_)
@@ -168,14 +94,3 @@ impl ResponseError for Error {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::Error;
-
-    #[test]
-    fn assert_send() {
-        fn is_send<E: Send>() {}
-        is_send::<Error>();
-    }
-}

src/extractor.rs

@@ -3,13 +3,13 @@ use crate::{
     upload::handle_multipart,
 };
 use actix_web::{dev::Payload, FromRequest, HttpRequest, ResponseError};
-use std::{future::Future, pin::Pin, rc::Rc};
+use std::{future::Future, pin::Pin};
 
 pub trait FormData {
     type Item: 'static;
     type Error: ResponseError + 'static;
 
-    fn form(req: &HttpRequest) -> Result<Form<Self::Item, Self::Error>, Self::Error>;
+    fn form(req: &HttpRequest) -> Form<Self::Item, Self::Error>;
 
     fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>
     where
@@ -27,16 +27,14 @@ where
 
     fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future {
         let multipart = actix_multipart::Multipart::new(req.headers(), payload.take());
-        let res = T::form(req);
+        let form = T::form(req);
 
         Box::pin(async move {
-            let form = Rc::new(res?);
-
-            let uploaded = match handle_multipart(multipart, Rc::clone(&form)).await {
+            let uploaded = match handle_multipart(multipart, &form).await {
                 Ok(Ok(uploaded)) => uploaded,
                 Ok(Err(e)) => return Err(e.into()),
                 Err(e) => {
-                    if let Some(f) = &form.transform_error {
+                    if let Some(f) = form.transform_error {
                         return Err((f)(e));
                     } else {
                         return Err(e.into());

src/lib.rs

@@ -28,18 +28,19 @@
 //! use actix_form_data::{Error, Field, Form, FormData, Multipart, Value};
 //! use actix_web::{
 //!     web::{post, resource},
-//!     App, HttpRequest, HttpResponse, HttpServer,
+//!     App, HttpResponse, HttpServer,
 //! };
 //! use futures_util::stream::StreamExt;
 //!
 //! struct UploadedContent(Value<()>);
 //!
 //! impl FormData for UploadedContent {
+//!     type Config = ();
 //!     type Item = ();
 //!     type Error = Error;
 //!
-//!     fn form(_: &HttpRequest) -> Result<Form<Self::Item, Self::Error>, Self::Error> {
-//!         Ok(Form::new()
+//!     fn form(_: Option<&()>) -> Form<Self::Item, Self::Error> {
+//!         Form::new()
 //!             .field("Hey", Field::text())
 //!             .field(
 //!                 "Hi",
@@ -56,7 +57,7 @@
 //!                     }
 //!                     Ok(()) as Result<(), Error>
 //!                 })),
-//!             ))
+//!             )
 //!     }
 //!
 //!     fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>

src/types.rs

@@ -19,7 +19,7 @@
 
 use crate::Error;
 use actix_web::web::Bytes;
-use futures_core::Stream;
+use futures_util::Stream;
 use mime::Mime;
 use std::{
     collections::{HashMap, VecDeque},
@@ -32,7 +32,7 @@ use tracing::trace;
 #[derive(Debug)]
 pub struct FileMeta<T> {
     pub filename: String,
-    pub content_type: Option<Mime>,
+    pub content_type: Mime,
     pub result: T,
 }
 
@@ -154,10 +154,10 @@ impl<T> From<MultipartContent<T>> for Value<T> {
 
 pub type FileFn<T, E> = Box<
     dyn Fn(
-        String,
-        Option<Mime>,
-        Pin<Box<dyn Stream<Item = Result<Bytes, Error>>>>,
-    ) -> Pin<Box<dyn Future<Output = Result<T, E>>>>,
+            String,
+            Mime,
+            Pin<Box<dyn Stream<Item = Result<Bytes, Error>>>>,
+        ) -> Pin<Box<dyn Future<Output = Result<T, E>>>>
 >;
 
 /// The field type represents a field in the form-data that is allowed to be parsed.
@@ -215,7 +215,7 @@ impl<T, E> Field<T, E> {
     /// ```
     pub fn file<F, Fut>(f: F) -> Self
     where
-        F: Fn(String, Option<Mime>, Pin<Box<dyn Stream<Item = Result<Bytes, Error>>>>) -> Fut
+        F: Fn(String, Mime, Pin<Box<dyn Stream<Item = Result<Bytes, Error>>>>) -> Fut
             + Clone
             + 'static,
         Fut: Future<Output = Result<T, E>> + 'static,
@@ -437,8 +437,8 @@ impl<T, E> Map<T, E> {
             Some(NamePart::Map(name_part)) => self
                 .inner
                 .iter()
-                .find(|(ref item, _)| *item == *name_part)
-                .and_then(|(_, ref field)| field.valid_field(name)),
+                .find(|&&(ref item, _)| *item == *name_part)
+                .and_then(|&(_, ref field)| field.valid_field(name)),
             _ => None,
         }
     }

src/upload.rs

@@ -25,10 +25,19 @@ use crate::{
     },
 };
 use actix_web::web::BytesMut;
-use std::{collections::HashMap, path::Path, rc::Rc};
-use streem::IntoStreamer;
-use tokio::task::JoinSet;
-use tracing::Instrument;
+use futures_util::{
+    select,
+    stream::{FuturesUnordered, StreamExt},
+};
+use std::{
+    collections::HashMap,
+    path::Path,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
+};
+use tracing::trace;
 
 fn consolidate<T>(mf: MultipartForm<T>) -> Value<T> {
     mf.into_iter().fold(
@@ -67,13 +76,16 @@ fn parse_multipart_name(name: String) -> Result<Vec<NamePart>, Error> {
                 NamePart::Map(part.to_owned())
             }
         })
-        .try_fold(vec![], |mut v, part| {
-            if v.is_empty() && !part.is_map() {
-                return Err(Error::ContentDisposition);
-            }
+        .fold(Ok(vec![]), |acc, part| match acc {
+            Ok(mut v) => {
+                if v.is_empty() && !part.is_map() {
+                    return Err(Error::ContentDisposition);
+                }
 
-            v.push(part);
-            Ok(v)
+                v.push(part);
+                Ok(v)
+            }
+            Err(e) => Err(e),
         })
 }
 
@@ -103,42 +115,32 @@ where
 
     let filename = filename.ok_or(Error::Filename)?.to_owned();
 
-    let content_type = field.content_type().cloned();
+    let file_size = Arc::new(AtomicUsize::new(0));
+
+    let content_type = field.content_type().clone();
 
     let max_file_size = form.max_file_size;
 
-    let field_stream = streem::try_from_fn(move |yielder| async move {
-        let mut file_size = 0;
-
-        let mut stream = field.into_streamer();
-
-        while let Some(bytes) = stream.try_next().await? {
-            tracing::trace!("Bytes from field");
-
-            file_size += bytes.len();
-
-            if file_size > max_file_size {
-                drop(bytes);
-
-                while stream.try_next().await?.is_some() {
-                    tracing::trace!("Dropping oversized bytes");
-                }
-
-                return Err(Error::FileSize);
-            }
-
-            yielder.yield_ok(bytes).await;
-        }
-
-        tracing::debug!("Finished consuming field");
-
-        Ok(())
-    });
-
     let result = file_fn(
         filename.clone(),
         content_type.clone(),
-        Box::pin(field_stream),
+        Box::pin(field.then(move |res| {
+            let file_size = file_size.clone();
+            async move {
+                match res {
+                    Ok(bytes) => {
+                        let size = file_size.fetch_add(bytes.len(), Ordering::Relaxed);
+
+                        if size + bytes.len() > max_file_size {
+                            return Err(Error::FileSize);
+                        }
+
+                        Ok(bytes)
+                    }
+                    Err(e) => Err(Error::from(e)),
+                }
+            }
+        })),
     )
     .await;
 
@@ -153,7 +155,7 @@ where
 }
 
 async fn handle_form_data<'a, T, E>(
-    field: actix_multipart::Field,
+    mut field: actix_multipart::Field,
     term: FieldTerminator<'a, T, E>,
     form: &Form<T, E>,
 ) -> Result<MultipartContent<T>, Error>
@@ -161,54 +163,27 @@ where
     T: 'static,
     E: 'static,
 {
-    tracing::trace!("In handle_form_data, term: {:?}", term);
-    let mut buf = Vec::new();
-
-    let mut stream = field.into_streamer();
-
-    while let Some(bytes) = stream.try_next().await? {
-        tracing::trace!("bytes from field");
-
-        if buf.len() + bytes.len() > form.max_field_size {
-            drop(buf);
-
-            while stream.try_next().await?.is_some() {
-                tracing::trace!("Dropping oversized bytes");
-            }
+    trace!("In handle_form_data, term: {:?}", term);
+    let mut bytes = BytesMut::new();
 
+    while let Some(res) = field.next().await {
+        let b = res?;
+        if bytes.len() + b.len() > form.max_field_size {
             return Err(Error::FieldSize);
         }
 
-        buf.push(bytes);
+        bytes.extend(b);
     }
 
-    let bytes = match buf.len() {
-        0 => return Err(Error::FieldSize),
-        1 => buf.pop().expect("contains an element"),
-        _ => {
-            let total_length = buf.iter().map(|b| b.len()).sum();
-
-            let mut bytes = BytesMut::with_capacity(total_length);
-
-            for b in buf {
-                bytes.extend(b);
-            }
-
-            bytes.freeze()
-        }
-    };
-
-    tracing::debug!("Finished consuming field");
-
     if let FieldTerminator::Bytes = term {
-        return Ok(MultipartContent::Bytes(bytes));
+        return Ok(MultipartContent::Bytes(bytes.freeze()));
     }
 
-    let s = std::str::from_utf8(&bytes).map_err(Error::ParseField)?;
+    let s = String::from_utf8(bytes.to_vec()).map_err(Error::ParseField)?;
 
     match term {
         FieldTerminator::Bytes | FieldTerminator::File(_) => Err(Error::FieldType),
-        FieldTerminator::Text => Ok(MultipartContent::Text(String::from(s))),
+        FieldTerminator::Text => Ok(MultipartContent::Text(s)),
         FieldTerminator::Float => s
             .parse()
             .map_err(Error::ParseFloat)
@@ -222,7 +197,7 @@ where
 
 async fn handle_stream_field<T, E>(
     field: actix_multipart::Field,
-    form: Rc<Form<T, E>>,
+    form: &Form<T, E>,
 ) -> Result<Result<MultipartHash<T>, E>, Error>
 where
     T: 'static,
@@ -239,22 +214,21 @@ where
 
     let content = match term {
         FieldTerminator::File(file_fn) => {
-            match handle_file_upload(field, content_disposition.filename, &form, file_fn).await? {
+            match handle_file_upload(field, content_disposition.filename, form, file_fn).await? {
                 Ok(content) => content,
                 Err(e) => return Ok(Err(e)),
             }
         }
-        term => handle_form_data(field, term, &form).await?,
+        term => handle_form_data(field, term, form).await?,
     };
 
     Ok(Ok((name, content)))
 }
 
 /// Handle multipart streams from Actix Web
-#[tracing::instrument(level = "TRACE", skip_all)]
 pub async fn handle_multipart<T, E>(
     m: actix_multipart::Multipart,
-    form: Rc<Form<T, E>>,
+    form: &Form<T, E>,
 ) -> Result<Result<Value<T>, E>, Error>
 where
     T: 'static,
@@ -264,111 +238,35 @@ where
     let mut file_count: u32 = 0;
     let mut field_count: u32 = 0;
 
-    let mut set = JoinSet::new();
+    let mut unordered = FuturesUnordered::new();
 
-    let mut m = m.into_streamer();
+    let mut m = m.fuse();
 
-    let mut error: Option<Error> = None;
-    let mut provided_error: Option<E> = None;
-    let mut is_closed = false;
-    let mut stream_error = false;
-
-    'outer: loop {
-        tracing::trace!("multipart loop");
-
-        if error.is_some() || provided_error.is_some() {
-            set.abort_all();
-
-            if !stream_error {
-                while let Some(res) = m.next().await {
-                    tracing::trace!("draining multipart field");
-
-                    if let Ok(field) = res {
-                        let mut stream = field.into_streamer();
-                        while stream.next().await.is_some() {
-                            tracing::trace!("Throwing away uploaded bytes, we have an error");
-                        }
-                    } else {
-                        break;
-                    }
+    loop {
+        select! {
+            opt = m.next() => {
+                if let Some(res) = opt {
+                    unordered.push(handle_stream_field(res?, form));
                 }
             }
-
-            while set.join_next().await.is_some() {
-                tracing::trace!("Throwing away joined result");
-            }
-
-            break 'outer;
-        }
-
-        tokio::select! {
-            opt = m.next(), if !is_closed => {
-                tracing::trace!("Selected stream");
-                is_closed = opt.is_none();
-
+            opt = unordered.next() => {
                 if let Some(res) = opt {
-                    match res {
-                        Ok(field) => {
-                            set.spawn_local(handle_stream_field(field, Rc::clone(&form)).instrument(tracing::trace_span!("multipart-field")));
-                        },
-                        Err(e) => {
-                            is_closed = true;
-                            stream_error = true;
-                            error = Some(e.into());
-                            continue 'outer;
-                        }
-                    }
-                }
-            }
-            opt = set.join_next(), if !set.is_empty() => {
-                tracing::trace!("Selected set");
-                if let Some(res) = opt {
-                    let (name_parts, content) = match res {
-                        Ok(Ok(Ok(tup))) => tup,
-                        Ok(Ok(Err(e))) => {
-                            provided_error = Some(e);
-                            continue 'outer;
-                        }
-                        Ok(Err(e)) => {
-                            error = Some(e);
-                            continue 'outer;
-                        },
-                        Err(e) => {
-                            error = Some(e.into());
-                            continue 'outer;
-                        },
-                    };
-
-                    let (l, r) = match count(&content, file_count, field_count, &form) {
+                    let (name_parts, content) = match res? {
                         Ok(tup) => tup,
-                        Err(e) => {
-                            error = Some(e);
-                            continue 'outer;
-                        }
+                        Err(e) => return Ok(Err(e)),
                     };
 
+                    let (l, r) = count(&content, file_count, field_count, form)?;
                     file_count = l;
                     field_count = r;
 
                     multipart_form.push((name_parts, content));
                 }
             }
-            else => {
-                break 'outer;
-            }
+            complete => break,
         }
     }
 
-    tracing::debug!("Finished consuming multipart");
-
-    if let Some(e) = provided_error {
-        return Ok(Err(e));
-    }
-
-    if let Some(e) = error {
-        return Err(e);
-    }
-
     Ok(Ok(consolidate(multipart_form)))
 }
 
@@ -381,13 +279,13 @@ fn count<T, E>(
     match content {
         MultipartContent::File(_) => {
             file_count += 1;
-            if file_count > form.max_files {
+            if file_count >= form.max_files {
                 return Err(Error::FileCount);
             }
         }
         _ => {
             field_count += 1;
-            if field_count > form.max_fields {
+            if field_count >= form.max_fields {
                 return Err(Error::FieldCount);
             }
         }