Handle cancellation of uploads actix#834 actix#736

boxdot · May 12, 2019 · 2350a2d · 2350a2d
1 parent 36d017d
commit 2350a2d
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 5 deletions.
diff --git a/actix-multipart/CHANGES.md b/actix-multipart/CHANGES.md
@@ -1,5 +1,11 @@
 # Changes
 
+## [0.1.0-beta.4] - 2019-05-12
+
+* Handle cancellation of uploads #834 #736
+
+* Upgrade to actix-web 1.0.0-beta.4
+
 ## [0.1.0-beta.1] - 2019-04-21
 
 * Do not support nested multipart

diff --git a/actix-multipart/Cargo.toml b/actix-multipart/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix-multipart"
-version = "0.1.0-beta.1"
+version = "0.1.0-beta.4"
 authors = ["Nikolay Kim <[email protected]>"]
 description = "Multipart support for actix web framework."
 readme = "README.md"
@@ -18,7 +18,7 @@ name = "actix_multipart"
 path = "src/lib.rs"
 
 [dependencies]
-actix-web = "1.0.0-beta.1"
+actix-web = "1.0.0-beta.4"
 actix-service = "0.4.0"
 bytes = "0.4"
 derive_more = "0.14"

diff --git a/actix-multipart/src/error.rs b/actix-multipart/src/error.rs
@@ -28,6 +28,9 @@ pub enum MultipartError {
     /// Payload error
     #[display(fmt = "{}", _0)]
     Payload(PayloadError),
+    /// Not consumed
+    #[display(fmt = "Multipart stream is not consumed")]
+    NotConsumed,
 }
 
 /// Return `BadRequest` for `MultipartError`

diff --git a/actix-multipart/src/server.rs b/actix-multipart/src/server.rs
@@ -1,5 +1,5 @@
 //! Multipart payload support
-use std::cell::{RefCell, UnsafeCell};
+use std::cell::{Cell, RefCell, UnsafeCell};
 use std::marker::PhantomData;
 use std::rc::Rc;
 use std::{cmp, fmt};
@@ -116,6 +116,8 @@ impl Stream for Multipart {
                 payload.poll_stream()?;
             }
             inner.poll(&self.safety)
+        } else if !self.safety.is_clean() {
+            Err(MultipartError::NotConsumed)
         } else {
             Ok(Async::NotReady)
         }
@@ -415,6 +417,8 @@ impl Stream for Field {
             }
 
             inner.poll(&self.safety)
+        } else if !self.safety.is_clean() {
+            return Err(MultipartError::NotConsumed);
         } else {
             Ok(Async::NotReady)
         }
@@ -655,6 +659,7 @@ struct Safety {
     task: Option<Task>,
     level: usize,
     payload: Rc<PhantomData<bool>>,
+    clean: Rc<Cell<bool>>,
 }
 
 impl Safety {
@@ -663,12 +668,17 @@ impl Safety {
         Safety {
             task: None,
             level: Rc::strong_count(&payload),
+            clean: Rc::new(Cell::new(true)),
             payload,
         }
     }
 
     fn current(&self) -> bool {
-        Rc::strong_count(&self.payload) == self.level
+        Rc::strong_count(&self.payload) == self.level && self.clean.get()
+    }
+
+    fn is_clean(&self) -> bool {
+        self.clean.get()
     }
 }
 
@@ -678,6 +688,7 @@ impl Clone for Safety {
         Safety {
             task: Some(current_task()),
             level: Rc::strong_count(&payload),
+            clean: self.clean.clone(),
             payload,
         }
     }
@@ -687,7 +698,7 @@ impl Drop for Safety {
     fn drop(&mut self) {
         // parent task is dead
         if Rc::strong_count(&self.payload) != self.level {
-            panic!("Safety get dropped but it is not from top-most task");
+            self.clean.set(true);
         }
         if let Some(task) = self.task.take() {
             task.notify()