Merge pull request #1358 from dunnock/msg-body

Adapt workspace crates to MessageBody::poll with Pin

Cargo.toml

@@ -108,7 +108,7 @@ codegen-units = 1
 
 [patch.crates-io]
 actix-web = { path = "." }
-#actix-http = { path = "actix-http" }
+actix-http = { path = "actix-http" }
 actix-http-test = { path = "test-server" }
 actix-web-codegen = { path = "actix-web-codegen" }
 actix-cors = { path = "actix-cors" }

actix-http/tests/test_ws.rs

@@ -81,9 +81,6 @@ async fn service(msg: ws::Frame) -> Result<ws::Message, Error> {
     Ok(msg)
 }
 
-/*
-Temporarily commented out due to dependency on actix-http-test
-
 #[actix_rt::test]
 async fn test_simple() {
     let ws_service = WsService::new();
@@ -195,5 +192,3 @@ async fn test_simple() {
 
     assert!(ws_service.was_polled());
 }
-
-*/

src/middleware/logger.rs

@@ -238,15 +238,20 @@ where
     }
 }
 
+use pin_project::{pin_project, pinned_drop};
+
+#[pin_project(PinnedDrop)]
 pub struct StreamLog<B> {
+    #[pin]
     body: ResponseBody<B>,
     format: Option<Format>,
     size: usize,
     time: OffsetDateTime,
 }
 
-impl<B> Drop for StreamLog<B> {
-    fn drop(&mut self) {
+#[pinned_drop]
+impl<B> PinnedDrop for StreamLog<B> {
+    fn drop(self: Pin<&mut Self>) {
         if let Some(ref format) = self.format {
             let render = |fmt: &mut Formatter<'_>| {
                 for unit in &format.0 {
@@ -259,15 +264,17 @@ impl<B> Drop for StreamLog<B> {
     }
 }
 
+
 impl<B: MessageBody> MessageBody for StreamLog<B> {
     fn size(&self) -> BodySize {
         self.body.size()
     }
 
-    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, Error>>> {
-        match self.body.poll_next(cx) {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, Error>>> {
+        let this = self.project();
+        match this.body.poll_next(cx) {
             Poll::Ready(Some(Ok(chunk))) => {
-                self.size += chunk.len();
+                *this.size += chunk.len();
                 Poll::Ready(Some(Ok(chunk)))
             }
             val => val,

src/test.rs

@@ -150,7 +150,7 @@ where
 pub async fn read_response<S, B>(app: &mut S, req: Request) -> Bytes
 where
     S: Service<Request = Request, Response = ServiceResponse<B>, Error = Error>,
-    B: MessageBody,
+    B: MessageBody + Unpin,
 {
     let mut resp = app
         .call(req)
@@ -193,7 +193,7 @@ where
 /// ```
 pub async fn read_body<B>(mut res: ServiceResponse<B>) -> Bytes
 where
-    B: MessageBody,
+    B: MessageBody + Unpin,
 {
     let mut body = res.take_body();
     let mut bytes = BytesMut::new();
@@ -251,7 +251,7 @@ where
 pub async fn read_response_json<S, B, T>(app: &mut S, req: Request) -> T
 where
     S: Service<Request = Request, Response = ServiceResponse<B>, Error = Error>,
-    B: MessageBody,
+    B: MessageBody + Unpin,
     T: DeserializeOwned,
 {
     let body = read_response(app, req).await;