From a7666e4881592cfd4899b3c1cec3ed6839a86018 Mon Sep 17 00:00:00 2001
From: Nikolay Kim <fafhrd91@gmail.com>
Date: Sun, 1 Dec 2024 16:40:20 +0500
Subject: [PATCH] Fix handling unconsumed payload in h1 dispatcher #477 (#478)

---
 ntex-http/CHANGES.md           |  2 +-
 ntex/CHANGES.md                |  6 ++++++
 ntex/src/http/h1/dispatcher.rs | 25 ++++++++++++++++++++++++-
 3 files changed, 31 insertions(+), 2 deletions(-)

diff --git a/ntex-http/CHANGES.md b/ntex-http/CHANGES.md
index 9d66b099..deac9c2c 100644
--- a/ntex-http/CHANGES.md
+++ b/ntex-http/CHANGES.md
@@ -1,6 +1,6 @@
 # Changes
 
-## [0.1.13] - 2024-01-xx
+## [0.1.13] - 2024-01-30
 
 * Move body related types from ntex::http
 
diff --git a/ntex/CHANGES.md b/ntex/CHANGES.md
index a1f89ac7..b5c796b0 100644
--- a/ntex/CHANGES.md
+++ b/ntex/CHANGES.md
@@ -1,5 +1,11 @@
 # Changes
 
+## [2.9.0] - 2024-11-30
+
+* Fix handling unconsumed payload in h1 dispatcher #477
+
+* Move body to ntex-http
+
 ## [2.8.0] - 2024-11-04
 
 * Use updated Service trait
diff --git a/ntex/src/http/h1/dispatcher.rs b/ntex/src/http/h1/dispatcher.rs
index 244853b8..18263583 100644
--- a/ntex/src/http/h1/dispatcher.rs
+++ b/ntex/src/http/h1/dispatcher.rs
@@ -181,7 +181,13 @@ where
                     Poll::Pending => ready!(inner.poll_request(cx)),
                 },
                 // read request and call service
-                State::ReadRequest => ready!(inner.poll_read_request(cx)),
+                State::ReadRequest => {
+                    if inner.flags.contains(Flags::SENDPAYLOAD_AND_STOP) {
+                        inner.stop()
+                    } else {
+                        ready!(inner.poll_read_request(cx))
+                    }
+                }
                 // consume request's payload
                 State::ReadPayload => {
                     let result = inner.poll_request_payload(cx);
@@ -1263,4 +1269,21 @@ mod tests {
         assert!(mark.load(Ordering::Relaxed) == 1536);
         assert!(err_mark.load(Ordering::Relaxed) == 1);
     }
+
+    #[crate::rt_test]
+    async fn test_unconsumed_payload() {
+        let (client, server) = Io::create();
+        client.remote_buffer_cap(4096);
+        client.write("GET /test HTTP/1.1\r\ncontent-length:512\r\n\r\n");
+
+        let mut h1 = h1(server, |_| {
+            Box::pin(async { Ok::<_, io::Error>(Response::Ok().body("TEST")) })
+        });
+        // required because io shutdown is async oper
+        assert!(poll_fn(|cx| Pin::new(&mut h1).poll(cx)).await.is_ok());
+
+        assert!(h1.inner.io.is_closed());
+        let buf = client.local_buffer(|buf| buf.split());
+        assert_eq!(&buf[..15], b"HTTP/1.1 200 OK");
+    }
 }