berkus
/
actix-web
mirror of https://github.com/fafhrd91/actix-web
1
0
Fork 0
Code Issues Releases Wiki Activity

ReadBody does not require Unpin stream

This commit is contained in:
Rob Ede 2021-12-24 23:51:33 +00:00
parent 48c8f8433f
commit 745e96a185
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
1 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
awc/src/response.rs
View File

@ -415,10 +415,13 @@ where
} }
} }
struct ReadBody<S> { pin_project_lite::pin_project! {
stream: Payload<S>, struct ReadBody<S> {
buf: BytesMut, #[pin]
limit: usize, stream: Payload<S>,
buf: BytesMut,
limit: usize,
}
} }
impl<S> ReadBody<S> { impl<S> ReadBody<S> {
@ -433,15 +436,15 @@ impl<S> ReadBody<S> {
impl<S> Future for ReadBody<S> impl<S> Future for ReadBody<S>
where where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin, S: Stream<Item = Result<Bytes, PayloadError>>,
{ {
type Output = Result<Bytes, PayloadError>; type Output = Result<Bytes, PayloadError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut(); let mut this = self.project();
while let Some(chunk) = ready!(Pin::new(&mut this.stream).poll_next(cx)?) { while let Some(chunk) = ready!(this.stream.as_mut().poll_next(cx)?) {
if (this.buf.len() + chunk.len()) > this.limit { if (this.buf.len() + chunk.len()) > *this.limit {
return Poll::Ready(Err(PayloadError::Overflow)); return Poll::Ready(Err(PayloadError::Overflow));
} }
this.buf.extend_from_slice(&chunk); this.buf.extend_from_slice(&chunk);
@ -453,11 +456,14 @@ where
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use static_assertions::assert_impl_all;
use super::*;
use crate::{http::header, test::TestResponse}; use crate::{http::header, test::TestResponse};
assert_impl_all!(ClientResponse: Unpin);
#[actix_rt::test] #[actix_rt::test]
async fn test_body() { async fn test_body() {
let mut req = TestResponse::with_header((header::CONTENT_LENGTH, "xxxx")).finish(); let mut req = TestResponse::with_header((header::CONTENT_LENGTH, "xxxx")).finish();