diff --git a/CHANGELOG.md b/CHANGELOG.md index 85397bb8..ebff37de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `tower::make::Shared` ([#229](https://github.com/tokio-rs/axum/pull/229)) - All usage of `tower::BoxError` has been replaced with `axum::BoxError` ([#229](https://github.com/tokio-rs/axum/pull/229)) - `tower::util::Either` no longer implements `IntoResponse` ([#229](https://github.com/tokio-rs/axum/pull/229)) +- `extract::BodyStream` is no longer generic over the request body ([#234](https://github.com/tokio-rs/axum/pull/234)) - `extract::Body` has been renamed to `extract::RawBody` to avoid conflicting with `body::Body` - These future types have been moved diff --git a/src/extract/mod.rs b/src/extract/mod.rs index c4aa9e6c..d2bbd9b5 100644 --- a/src/extract/mod.rs +++ b/src/extract/mod.rs @@ -259,7 +259,7 @@ //! .route( //! "/body-stream", //! // same for `extract::BodyStream` -//! get(|_: extract::BodyStream>| async {}), +//! get(|_: extract::BodyStream| async {}), //! ) //! .route( //! // and `Request<_>` diff --git a/src/extract/request_parts.rs b/src/extract/request_parts.rs index c209194d..aeafbc17 100644 --- a/src/extract/request_parts.rs +++ b/src/extract/request_parts.rs @@ -1,14 +1,17 @@ use super::{rejection::*, take_body, Extension, FromRequest, RequestParts}; -use crate::BoxError; +use crate::{BoxError, Error}; use async_trait::async_trait; use bytes::Bytes; use futures_util::stream::Stream; use http::{Extensions, HeaderMap, Method, Request, Uri, Version}; +use http_body::Body as HttpBody; use std::{ convert::Infallible, + fmt, pin::Pin, task::{Context, Poll}, }; +use sync_wrapper::SyncWrapper; #[async_trait] impl FromRequest for Request @@ -191,34 +194,48 @@ where /// ``` /// /// [`Stream`]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html -#[derive(Debug)] -pub struct BodyStream(B); +pub struct BodyStream( + SyncWrapper + Send + 'static>>>, +); -impl Stream for BodyStream -where - B: http_body::Body + Unpin, -{ - type Item = Result; +impl Stream for BodyStream { + type Item = Result; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Pin::new(&mut self.0).poll_data(cx) + Pin::new(self.0.get_mut()).poll_data(cx) } } #[async_trait] -impl FromRequest for BodyStream +impl FromRequest for BodyStream where - B: http_body::Body + Unpin + Send, + B: HttpBody + Send + 'static, + B::Data: Into, + B::Error: Into, { type Rejection = BodyAlreadyExtracted; async fn from_request(req: &mut RequestParts) -> Result { - let body = take_body(req)?; - let stream = BodyStream(body); + let body = take_body(req)? + .map_data(Into::into) + .map_err(|err| Error::new(err.into())); + let stream = BodyStream(SyncWrapper::new(Box::pin(body))); Ok(stream) } } +impl fmt::Debug for BodyStream { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_tuple("BodyStream").finish() + } +} + +#[test] +fn body_stream_traits() { + crate::tests::assert_send::(); + crate::tests::assert_sync::(); +} + /// Extractor that extracts the raw request body. /// /// # Example diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 8e38c45b..62af8a17 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -698,3 +698,6 @@ where addr } + +pub(crate) fn assert_send() {} +pub(crate) fn assert_sync() {}