diff --git a/examples/echo.rs b/examples/echo.rs index 47f88a0fa0..c7a564e9e7 100644 --- a/examples/echo.rs +++ b/examples/echo.rs @@ -7,12 +7,12 @@ use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full}; use hyper::body::HttpBody as _; use hyper::server::conn::Http; use hyper::service::service_fn; -use hyper::{Body, Method, Request, Response, StatusCode}; +use hyper::{Method, Recv, Request, Response, StatusCode}; use tokio::net::TcpListener; /// This is our service handler. It receives a Request, routes on its /// path, and returns a Future of a Response. -async fn echo(req: Request) -> Result>, hyper::Error> { +async fn echo(req: Request) -> Result>, hyper::Error> { match (req.method(), req.uri().path()) { // Serve some instructions at / (&Method::GET, "/") => Ok(Response::new(full( diff --git a/examples/hello.rs b/examples/hello.rs index 6acf47d228..4f23172b40 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -7,10 +7,10 @@ use bytes::Bytes; use http_body_util::Full; use hyper::server::conn::Http; use hyper::service::service_fn; -use hyper::{Body, Request, Response}; +use hyper::{Recv, Request, Response}; use tokio::net::TcpListener; -async fn hello(_: Request) -> Result>, Infallible> { +async fn hello(_: Request) -> Result>, Infallible> { Ok(Response::new(Full::new(Bytes::from("Hello World!")))) } diff --git a/examples/http_proxy.rs b/examples/http_proxy.rs index bddae68484..8d262f9002 100644 --- a/examples/http_proxy.rs +++ b/examples/http_proxy.rs @@ -8,7 +8,7 @@ use hyper::client::conn::Builder; use hyper::server::conn::Http; use hyper::service::service_fn; use hyper::upgrade::Upgraded; -use hyper::{Body, Method, Request, Response}; +use hyper::{Method, Recv, Request, Response}; use tokio::net::{TcpListener, TcpStream}; @@ -43,7 +43,7 @@ async fn main() -> Result<(), Box> { } } -async fn proxy(req: Request) -> Result>, hyper::Error> { +async fn proxy(req: Request) -> Result>, hyper::Error> { println!("req: {:?}", req); if Method::CONNECT == req.method() { diff --git a/examples/multi_server.rs b/examples/multi_server.rs index be084e04e7..302525b29f 100644 --- a/examples/multi_server.rs +++ b/examples/multi_server.rs @@ -8,17 +8,17 @@ use futures_util::future::join; use http_body_util::Full; use hyper::server::conn::Http; use hyper::service::service_fn; -use hyper::{Body, Request, Response}; +use hyper::{Recv, Request, Response}; use tokio::net::TcpListener; static INDEX1: &[u8] = b"The 1st service!"; static INDEX2: &[u8] = b"The 2nd service!"; -async fn index1(_: Request) -> Result>, hyper::Error> { +async fn index1(_: Request) -> Result>, hyper::Error> { Ok(Response::new(Full::new(Bytes::from(INDEX1)))) } -async fn index2(_: Request) -> Result>, hyper::Error> { +async fn index2(_: Request) -> Result>, hyper::Error> { Ok(Response::new(Full::new(Bytes::from(INDEX2)))) } diff --git a/examples/params.rs b/examples/params.rs index 02a1b7a5ad..da5a182bb2 100644 --- a/examples/params.rs +++ b/examples/params.rs @@ -5,7 +5,7 @@ use bytes::Bytes; use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full}; use hyper::server::conn::Http; use hyper::service::service_fn; -use hyper::{Body, Method, Request, Response, StatusCode}; +use hyper::{Method, Recv, Request, Response, StatusCode}; use tokio::net::TcpListener; use std::collections::HashMap; @@ -19,7 +19,7 @@ static NOTNUMERIC: &[u8] = b"Number field is not numeric"; // Using service_fn, we can turn this function into a `Service`. async fn param_example( - req: Request, + req: Request, ) -> Result>, hyper::Error> { match (req.method(), req.uri().path()) { (&Method::GET, "/") | (&Method::GET, "/post") => Ok(Response::new(full(INDEX))), diff --git a/examples/send_file.rs b/examples/send_file.rs index 699456b3fe..1249a0d7f2 100644 --- a/examples/send_file.rs +++ b/examples/send_file.rs @@ -8,7 +8,7 @@ use tokio::net::TcpListener; use bytes::Bytes; use http_body_util::Full; use hyper::service::service_fn; -use hyper::{Body, Method, Request, Response, Result, StatusCode}; +use hyper::{Method, Recv, Request, Response, Result, StatusCode}; static INDEX: &str = "examples/send_file_index.html"; static NOTFOUND: &[u8] = b"Not Found"; @@ -36,7 +36,7 @@ async fn main() -> std::result::Result<(), Box> { } } -async fn response_examples(req: Request) -> Result>> { +async fn response_examples(req: Request) -> Result>> { match (req.method(), req.uri().path()) { (&Method::GET, "/") | (&Method::GET, "/index.html") => simple_file_send(INDEX).await, (&Method::GET, "/no_file.html") => { diff --git a/examples/service_struct_impl.rs b/examples/service_struct_impl.rs index f2a3aba8b6..781e1e7a16 100644 --- a/examples/service_struct_impl.rs +++ b/examples/service_struct_impl.rs @@ -2,7 +2,7 @@ use bytes::Bytes; use http_body_util::Full; use hyper::server::conn::Http; use hyper::service::Service; -use hyper::{Body, Request, Response}; +use hyper::{Recv, Request, Response}; use tokio::net::TcpListener; use std::future::Future; @@ -37,7 +37,7 @@ struct Svc { counter: Counter, } -impl Service> for Svc { +impl Service> for Svc { type Response = Response>; type Error = hyper::Error; type Future = Pin> + Send>>; @@ -46,7 +46,7 @@ impl Service> for Svc { Poll::Ready(Ok(())) } - fn call(&mut self, req: Request) -> Self::Future { + fn call(&mut self, req: Request) -> Self::Future { fn mk_response(s: String) -> Result>, hyper::Error> { Ok(Response::builder().body(Full::new(Bytes::from(s))).unwrap()) } diff --git a/examples/upgrades.rs b/examples/upgrades.rs index 08082af89a..31980cb9b8 100644 --- a/examples/upgrades.rs +++ b/examples/upgrades.rs @@ -14,7 +14,7 @@ use hyper::header::{HeaderValue, UPGRADE}; use hyper::server::conn::Http; use hyper::service::service_fn; use hyper::upgrade::Upgraded; -use hyper::{Body, Request, Response, StatusCode}; +use hyper::{Recv, Request, Response, StatusCode}; // A simple type alias so as to DRY. type Result = std::result::Result>; @@ -38,7 +38,7 @@ async fn server_upgraded_io(mut upgraded: Upgraded) -> Result<()> { } /// Our server HTTP handler to initiate HTTP upgrades. -async fn server_upgrade(mut req: Request) -> Result>> { +async fn server_upgrade(mut req: Request) -> Result>> { let mut res = Response::new(Empty::new()); // Send a 400 to any request that doesn't have diff --git a/examples/web_api.rs b/examples/web_api.rs index d2ac47dbf4..220c5d822e 100644 --- a/examples/web_api.rs +++ b/examples/web_api.rs @@ -6,7 +6,7 @@ use bytes::{Buf, Bytes}; use http_body_util::{BodyExt, Full}; use hyper::server::conn::Http; use hyper::service::service_fn; -use hyper::{header, Body, Method, Request, Response, StatusCode}; +use hyper::{header, Method, Recv, Request, Response, StatusCode}; use tokio::net::{TcpListener, TcpStream}; type GenericError = Box; @@ -46,7 +46,7 @@ async fn client_request_response() -> Result> { Ok(Response::new(res_body)) } -async fn api_post_response(req: Request) -> Result> { +async fn api_post_response(req: Request) -> Result> { // Aggregate the body... let whole_body = hyper::body::aggregate(req).await?; // Decode as JSON... @@ -77,7 +77,7 @@ async fn api_get_response() -> Result> { Ok(res) } -async fn response_examples(req: Request) -> Result> { +async fn response_examples(req: Request) -> Result> { match (req.method(), req.uri().path()) { (&Method::GET, "/") | (&Method::GET, "/index.html") => Ok(Response::new(full(INDEX))), (&Method::GET, "/test.html") => client_request_response().await, diff --git a/src/body/body.rs b/src/body/body.rs index ac6318a3d3..abba23c327 100644 --- a/src/body/body.rs +++ b/src/body/body.rs @@ -24,7 +24,7 @@ type TrailersSender = oneshot::Sender; /// Note: To read the full body, use [`body::to_bytes`](crate::body::to_bytes) /// or [`body::aggregate`](crate::body::aggregate). #[must_use = "streams do nothing unless polled"] -pub struct Body { +pub struct Recv { kind: Kind, } @@ -70,16 +70,16 @@ pub struct Sender { const WANT_PENDING: usize = 1; const WANT_READY: usize = 2; -impl Body { +impl Recv { /// Create a `Body` stream with an associated sender half. /// /// Useful when wanting to stream chunks from another thread. #[inline] - pub fn channel() -> (Sender, Body) { + pub fn channel() -> (Sender, Recv) { Self::new_channel(DecodedLength::CHUNKED, /*wanter =*/ false) } - pub(crate) fn new_channel(content_length: DecodedLength, wanter: bool) -> (Sender, Body) { + pub(crate) fn new_channel(content_length: DecodedLength, wanter: bool) -> (Sender, Recv) { let (data_tx, data_rx) = mpsc::channel(0); let (trailers_tx, trailers_rx) = oneshot::channel(); @@ -94,7 +94,7 @@ impl Body { data_tx, trailers_tx: Some(trailers_tx), }; - let rx = Body::new(Kind::Chan { + let rx = Recv::new(Kind::Chan { content_length, want_tx, data_rx, @@ -104,18 +104,18 @@ impl Body { (tx, rx) } - fn new(kind: Kind) -> Body { - Body { kind } + fn new(kind: Kind) -> Recv { + Recv { kind } } #[allow(dead_code)] - pub(crate) fn empty() -> Body { - Body::new(Kind::Empty) + pub(crate) fn empty() -> Recv { + Recv::new(Kind::Empty) } #[cfg(feature = "ffi")] - pub(crate) fn ffi() -> Body { - Body::new(Kind::Ffi(crate::ffi::UserBody::new())) + pub(crate) fn ffi() -> Recv { + Recv::new(Kind::Ffi(crate::ffi::UserBody::new())) } #[cfg(all(feature = "http2", any(feature = "client", feature = "server")))] @@ -129,7 +129,7 @@ impl Body { if !content_length.is_exact() && recv.is_end_stream() { content_length = DecodedLength::ZERO; } - let body = Body::new(Kind::H2 { + let body = Recv::new(Kind::H2 { ping, content_length, recv, @@ -194,7 +194,7 @@ impl Body { } } -impl HttpBody for Body { +impl HttpBody for Recv { type Data = Bytes; type Error = crate::Error; @@ -270,7 +270,7 @@ impl HttpBody for Body { } } -impl fmt::Debug for Body { +impl fmt::Debug for Recv { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { #[derive(Debug)] struct Streaming; @@ -382,14 +382,14 @@ mod tests { use std::mem; use std::task::Poll; - use super::{Body, DecodedLength, HttpBody, Sender, SizeHint}; + use super::{DecodedLength, HttpBody, Recv, Sender, SizeHint}; #[test] fn test_size_of() { // These are mostly to help catch *accidentally* increasing // the size by too much. - let body_size = mem::size_of::(); + let body_size = mem::size_of::(); let body_expected_size = mem::size_of::() * 6; assert!( body_size <= body_expected_size, @@ -398,7 +398,7 @@ mod tests { body_expected_size, ); - assert_eq!(body_size, mem::size_of::>(), "Option"); + assert_eq!(body_size, mem::size_of::>(), "Option"); assert_eq!( mem::size_of::(), @@ -415,18 +415,18 @@ mod tests { #[test] fn size_hint() { - fn eq(body: Body, b: SizeHint, note: &str) { + fn eq(body: Recv, b: SizeHint, note: &str) { let a = body.size_hint(); assert_eq!(a.lower(), b.lower(), "lower for {:?}", note); assert_eq!(a.upper(), b.upper(), "upper for {:?}", note); } - eq(Body::empty(), SizeHint::with_exact(0), "empty"); + eq(Recv::empty(), SizeHint::with_exact(0), "empty"); - eq(Body::channel().1, SizeHint::new(), "channel"); + eq(Recv::channel().1, SizeHint::new(), "channel"); eq( - Body::new_channel(DecodedLength::new(4), /*wanter =*/ false).1, + Recv::new_channel(DecodedLength::new(4), /*wanter =*/ false).1, SizeHint::with_exact(4), "channel with length", ); @@ -435,7 +435,7 @@ mod tests { #[cfg(not(miri))] #[tokio::test] async fn channel_abort() { - let (tx, mut rx) = Body::channel(); + let (tx, mut rx) = Recv::channel(); tx.abort(); @@ -446,7 +446,7 @@ mod tests { #[cfg(not(miri))] #[tokio::test] async fn channel_abort_when_buffer_is_full() { - let (mut tx, mut rx) = Body::channel(); + let (mut tx, mut rx) = Recv::channel(); tx.try_send_data("chunk 1".into()).expect("send 1"); // buffer is full, but can still send abort @@ -461,7 +461,7 @@ mod tests { #[test] fn channel_buffers_one() { - let (mut tx, _rx) = Body::channel(); + let (mut tx, _rx) = Recv::channel(); tx.try_send_data("chunk 1".into()).expect("send 1"); @@ -473,14 +473,14 @@ mod tests { #[cfg(not(miri))] #[tokio::test] async fn channel_empty() { - let (_, mut rx) = Body::channel(); + let (_, mut rx) = Recv::channel(); assert!(rx.data().await.is_none()); } #[test] fn channel_ready() { - let (mut tx, _rx) = Body::new_channel(DecodedLength::CHUNKED, /*wanter = */ false); + let (mut tx, _rx) = Recv::new_channel(DecodedLength::CHUNKED, /*wanter = */ false); let mut tx_ready = tokio_test::task::spawn(tx.ready()); @@ -489,7 +489,7 @@ mod tests { #[test] fn channel_wanter() { - let (mut tx, mut rx) = Body::new_channel(DecodedLength::CHUNKED, /*wanter = */ true); + let (mut tx, mut rx) = Recv::new_channel(DecodedLength::CHUNKED, /*wanter = */ true); let mut tx_ready = tokio_test::task::spawn(tx.ready()); let mut rx_data = tokio_test::task::spawn(rx.data()); @@ -510,7 +510,7 @@ mod tests { #[test] fn channel_notices_closure() { - let (mut tx, rx) = Body::new_channel(DecodedLength::CHUNKED, /*wanter = */ true); + let (mut tx, rx) = Recv::new_channel(DecodedLength::CHUNKED, /*wanter = */ true); let mut tx_ready = tokio_test::task::spawn(tx.ready()); diff --git a/src/body/mod.rs b/src/body/mod.rs index 8c6789daa5..0348a00efe 100644 --- a/src/body/mod.rs +++ b/src/body/mod.rs @@ -10,7 +10,7 @@ //! - **The [`HttpBody`](HttpBody) trait** describes all possible bodies. //! hyper allows any body type that implements `HttpBody`, allowing //! applications to have fine-grained control over their streaming. -//! - **The [`Body`](Body) concrete type**, which is an implementation of +//! - **The [`Recv`](Recv) concrete type**, which is an implementation of //! `HttpBody`, and returned by hyper as a "receive stream" (so, for server //! requests and client responses). It is also a decent default implementation //! if you don't have very custom needs of your send streams. @@ -20,7 +20,7 @@ pub use http_body::Body as HttpBody; pub use http_body::SizeHint; pub use self::aggregate::aggregate; -pub use self::body::{Body, Sender}; +pub use self::body::{Recv, Sender}; pub(crate) use self::length::DecodedLength; pub use self::to_bytes::to_bytes; @@ -33,6 +33,6 @@ fn _assert_send_sync() { fn _assert_send() {} fn _assert_sync() {} - _assert_send::(); - _assert_sync::(); + _assert_send::(); + _assert_sync::(); } diff --git a/src/body/to_bytes.rs b/src/body/to_bytes.rs index b56c4716df..cfb6f1f56c 100644 --- a/src/body/to_bytes.rs +++ b/src/body/to_bytes.rs @@ -17,8 +17,8 @@ use super::HttpBody; /// # Example /// /// ``` -/// # use hyper::{Body, Response}; -/// # async fn doc(response: Response) -> hyper::Result<()> { +/// # use hyper::{Recv, Response}; +/// # async fn doc(response: Response) -> hyper::Result<()> { /// # use hyper::body::HttpBody; /// // let response: Response ... /// diff --git a/src/client/conn/http1.rs b/src/client/conn/http1.rs index a1b0cc9858..7b8372de69 100644 --- a/src/client/conn/http1.rs +++ b/src/client/conn/http1.rs @@ -8,7 +8,7 @@ use http::{Request, Response}; use httparse::ParserConfig; use tokio::io::{AsyncRead, AsyncWrite}; -use crate::Body; +use crate::Recv; use crate::body::HttpBody; use crate::common::{ exec::{BoxSendFuture, Exec}, @@ -24,7 +24,7 @@ type Dispatcher = /// The sender side of an established connection. pub struct SendRequest { - dispatch: dispatch::Sender, Response>, + dispatch: dispatch::Sender, Response>, } /// A future that processes all HTTP state for the IO object. @@ -63,7 +63,7 @@ pub struct Builder { /// See [`client::conn`](crate::client::conn) for more. pub async fn handshake( io: T, -) -> crate::Result<(SendRequest, Connection)> +) -> crate::Result<(SendRequest, Connection)> where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -120,7 +120,7 @@ where /// before calling this method. /// - Since absolute-form `Uri`s are not required, if received, they will /// be serialized as-is. - pub fn send_request(&mut self, req: Request) -> impl Future>> { + pub fn send_request(&mut self, req: Request) -> impl Future>> { let sent = self.dispatch.send(req); async move { diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs index 98ba5311a0..d0a3e58a0b 100644 --- a/src/client/conn/http2.rs +++ b/src/client/conn/http2.rs @@ -10,7 +10,7 @@ use std::time::Duration; use http::{Request, Response}; use tokio::io::{AsyncRead, AsyncWrite}; -use crate::Body; +use crate::Recv; use crate::body::HttpBody; use crate::common::{ exec::{BoxSendFuture, Exec}, @@ -22,7 +22,7 @@ use super::super::dispatch; /// The sender side of an established connection. pub struct SendRequest { - dispatch: dispatch::UnboundedSender, Response>, + dispatch: dispatch::UnboundedSender, Response>, } /// A future that processes all HTTP state for the IO object. @@ -53,7 +53,7 @@ pub struct Builder { /// See [`client::conn`](crate::client::conn) for more. pub async fn handshake( io: T, -) -> crate::Result<(SendRequest, Connection)> +) -> crate::Result<(SendRequest, Connection)> where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -114,7 +114,7 @@ where /// before calling this method. /// - Since absolute-form `Uri`s are not required, if received, they will /// be serialized as-is. - pub fn send_request(&mut self, req: Request) -> impl Future>> { + pub fn send_request(&mut self, req: Request) -> impl Future>> { let sent = self.dispatch.send(req); async move { diff --git a/src/client/conn/mod.rs b/src/client/conn/mod.rs index 19ec8f7130..ad83f4e89e 100644 --- a/src/client/conn/mod.rs +++ b/src/client/conn/mod.rs @@ -16,7 +16,7 @@ //! use bytes::Bytes; //! use http::{Request, StatusCode}; //! use http_body_util::Empty; -//! use hyper::{client::conn, Body}; +//! use hyper::client::conn; //! use tokio::net::TcpStream; //! use tower::ServiceExt; //! @@ -84,7 +84,7 @@ use crate::proto; use crate::rt::Executor; #[cfg(feature = "http1")] use crate::upgrade::Upgraded; -use crate::{Body, Request, Response}; +use crate::{Recv, Request, Response}; #[cfg(feature = "http1")] pub mod http1; @@ -139,7 +139,7 @@ where /// The sender side of an established connection. pub struct SendRequest { - dispatch: dispatch::Sender, Response>, + dispatch: dispatch::Sender, Response>, } /// A future that processes all HTTP state for the IO object. @@ -194,7 +194,7 @@ pub struct ResponseFuture { } enum ResponseFutureState { - Waiting(dispatch::Promise>), + Waiting(dispatch::Promise>), // Option is to be able to `take()` it in `poll` Error(Option), } @@ -268,7 +268,7 @@ impl Service> for SendRequest where B: HttpBody + 'static, { - type Response = Response; + type Response = Response; type Error = crate::Error; type Future = ResponseFuture; @@ -875,7 +875,7 @@ impl Builder { // ===== impl ResponseFuture impl Future for ResponseFuture { - type Output = crate::Result>; + type Output = crate::Result>; fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll { match self.inner { diff --git a/src/client/dispatch.rs b/src/client/dispatch.rs index 5e896a9ffe..8806c638b0 100644 --- a/src/client/dispatch.rs +++ b/src/client/dispatch.rs @@ -379,15 +379,15 @@ mod tests { #[cfg(feature = "nightly")] #[bench] fn giver_queue_throughput(b: &mut test::Bencher) { - use crate::{Body, Request, Response}; + use crate::{Recv, Request, Response}; let rt = tokio::runtime::Builder::new_current_thread() .build() .unwrap(); - let (mut tx, mut rx) = channel::, Response>(); + let (mut tx, mut rx) = channel::, Response>(); b.iter(move || { - let _ = tx.send(Request::new(Body::empty())).unwrap(); + let _ = tx.send(Request::new(Recv::empty())).unwrap(); rt.block_on(async { loop { let poll_once = PollOnce(&mut rx); diff --git a/src/ffi/body.rs b/src/ffi/body.rs index a455dae0e3..ae2c62eebe 100644 --- a/src/ffi/body.rs +++ b/src/ffi/body.rs @@ -8,10 +8,10 @@ use libc::{c_int, size_t}; use super::task::{hyper_context, hyper_task, hyper_task_return_type, AsTaskType}; use super::{UserDataPointer, HYPER_ITER_CONTINUE}; -use crate::body::{Body, Bytes, HttpBody as _}; +use crate::body::{Bytes, HttpBody as _, Recv}; /// A streaming HTTP body. -pub struct hyper_body(pub(super) Body); +pub struct hyper_body(pub(super) Recv); /// A buffer of bytes that is sent or received on a `hyper_body`. pub struct hyper_buf(pub(crate) Bytes); @@ -33,7 +33,7 @@ ffi_fn! { /// /// If not configured, this body acts as an empty payload. fn hyper_body_new() -> *mut hyper_body { - Box::into_raw(Box::new(hyper_body(Body::ffi()))) + Box::into_raw(Box::new(hyper_body(Recv::ffi()))) } ?= ptr::null_mut() } diff --git a/src/ffi/client.rs b/src/ffi/client.rs index 580718f4fc..22cae7fbd4 100644 --- a/src/ffi/client.rs +++ b/src/ffi/client.rs @@ -24,7 +24,7 @@ pub struct hyper_clientconn_options { /// send multiple requests on a single connection, such as when HTTP/1 /// keep-alive or HTTP/2 is used. pub struct hyper_clientconn { - tx: conn::SendRequest, + tx: conn::SendRequest, } // ===== impl hyper_clientconn ===== @@ -42,7 +42,7 @@ ffi_fn! { let io = non_null! { Box::from_raw(io) ?= ptr::null_mut() }; Box::into_raw(hyper_task::boxed(async move { - options.builder.handshake::<_, crate::Body>(io) + options.builder.handshake::<_, crate::Recv>(io) .await .map(|(tx, conn)| { options.exec.execute(Box::pin(async move { diff --git a/src/ffi/http_types.rs b/src/ffi/http_types.rs index f96a58d165..885b31c73b 100644 --- a/src/ffi/http_types.rs +++ b/src/ffi/http_types.rs @@ -8,13 +8,13 @@ use super::task::{hyper_task_return_type, AsTaskType}; use super::{UserDataPointer, HYPER_ITER_CONTINUE}; use crate::ext::{HeaderCaseMap, OriginalHeaderOrder, ReasonPhrase}; use crate::header::{HeaderName, HeaderValue}; -use crate::{Body, HeaderMap, Method, Request, Response, Uri}; +use crate::{HeaderMap, Method, Recv, Request, Response, Uri}; /// An HTTP request. -pub struct hyper_request(pub(super) Request); +pub struct hyper_request(pub(super) Request); /// An HTTP response. -pub struct hyper_response(pub(super) Response); +pub struct hyper_response(pub(super) Response); /// An HTTP header map. /// @@ -39,7 +39,7 @@ type hyper_request_on_informational_callback = extern "C" fn(*mut c_void, *mut h ffi_fn! { /// Construct a new HTTP request. fn hyper_request_new() -> *mut hyper_request { - Box::into_raw(Box::new(hyper_request(Request::new(Body::empty())))) + Box::into_raw(Box::new(hyper_request(Request::new(Recv::empty())))) } ?= std::ptr::null_mut() } @@ -335,13 +335,13 @@ ffi_fn! { /// /// It is safe to free the response even after taking ownership of its body. fn hyper_response_body(resp: *mut hyper_response) -> *mut hyper_body { - let body = std::mem::replace(non_null!(&mut *resp ?= std::ptr::null_mut()).0.body_mut(), crate::Body::empty()); + let body = std::mem::replace(non_null!(&mut *resp ?= std::ptr::null_mut()).0.body_mut(), crate::Recv::empty()); Box::into_raw(Box::new(hyper_body(body))) } ?= std::ptr::null_mut() } impl hyper_response { - pub(super) fn wrap(mut resp: Response) -> hyper_response { + pub(super) fn wrap(mut resp: Response) -> hyper_response { let headers = std::mem::take(resp.headers_mut()); let orig_casing = resp .extensions_mut() @@ -532,7 +532,7 @@ unsafe fn raw_name_value( // ===== impl OnInformational ===== impl OnInformational { - pub(crate) fn call(&mut self, resp: Response) { + pub(crate) fn call(&mut self, resp: Response) { let mut resp = hyper_response::wrap(resp); (self.func)(self.data.0, &mut resp); } diff --git a/src/lib.rs b/src/lib.rs index 865027459c..3c56836a2f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -65,7 +65,7 @@ pub use crate::http::{header, Method, Request, Response, StatusCode, Uri, Versio #[doc(no_inline)] pub use crate::http::HeaderMap; -pub use crate::body::Body; +pub use crate::body::Recv; pub use crate::error::{Error, Result}; #[macro_use] diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs index 5a5daf6d43..c744db8648 100644 --- a/src/proto/h1/dispatch.rs +++ b/src/proto/h1/dispatch.rs @@ -6,7 +6,7 @@ use tokio::io::{AsyncRead, AsyncWrite}; use tracing::{debug, trace}; use super::{Http1Transaction, Wants}; -use crate::body::{Body, DecodedLength, HttpBody}; +use crate::body::{Recv, DecodedLength, HttpBody}; use crate::common::{task, Future, Pin, Poll, Unpin}; use crate::proto::{BodyLength, Conn, Dispatched, MessageHead, RequestHead}; use crate::upgrade::OnUpgrade; @@ -28,7 +28,7 @@ pub(crate) trait Dispatch { self: Pin<&mut Self>, cx: &mut task::Context<'_>, ) -> Poll>>; - fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()>; + fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Recv)>) -> crate::Result<()>; fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll>; fn should_poll(&self) -> bool; } @@ -45,14 +45,14 @@ cfg_server! { cfg_client! { pin_project_lite::pin_project! { pub(crate) struct Client { - callback: Option, http::Response>>, + callback: Option, http::Response>>, #[pin] rx: ClientRx, rx_closed: bool, } } - type ClientRx = crate::client::dispatch::Receiver, http::Response>; + type ClientRx = crate::client::dispatch::Receiver, http::Response>; } impl Dispatcher @@ -246,9 +246,9 @@ where match ready!(self.conn.poll_read_head(cx)) { Some(Ok((mut head, body_len, wants))) => { let body = match body_len { - DecodedLength::ZERO => Body::empty(), + DecodedLength::ZERO => Recv::empty(), other => { - let (tx, rx) = Body::new_channel(other, wants.contains(Wants::EXPECT)); + let (tx, rx) = Recv::new_channel(other, wants.contains(Wants::EXPECT)); self.body_tx = Some(tx); rx } @@ -460,9 +460,9 @@ cfg_server! { // Service is never pinned impl, B> Unpin for Server {} - impl Dispatch for Server + impl Dispatch for Server where - S: HttpService, + S: HttpService, S::Error: Into>, Bs: HttpBody, { @@ -495,7 +495,7 @@ cfg_server! { ret } - fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> { + fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Recv)>) -> crate::Result<()> { let (msg, body) = msg?; let mut req = Request::new(body); *req.method_mut() = msg.subject.0; @@ -584,7 +584,7 @@ cfg_client! { } } - fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> { + fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Recv)>) -> crate::Result<()> { match msg { Ok((msg, body)) => { if let Some(cb) = self.callback.take() { @@ -666,7 +666,7 @@ mod tests { handle.read(b"HTTP/1.1 200 OK\r\n\r\n"); let mut res_rx = tx - .try_send(crate::Request::new(crate::Body::empty())) + .try_send(crate::Request::new(crate::Recv::empty())) .unwrap(); tokio_test::assert_ready_ok!(Pin::new(&mut dispatcher).poll(cx)); @@ -698,7 +698,7 @@ mod tests { let _dispatcher = tokio::spawn(async move { dispatcher.await }); let body = { - let (mut tx, body) = crate::Body::new_channel(DecodedLength::new(4), false); + let (mut tx, body) = crate::Recv::new_channel(DecodedLength::new(4), false); tx.try_send_data("reee".into()).unwrap(); body }; @@ -731,7 +731,7 @@ mod tests { assert!(dispatcher.poll().is_pending()); let body = { - let (mut tx, body) = crate::Body::channel(); + let (mut tx, body) = crate::Recv::channel(); tx.try_send_data("".into()).unwrap(); body }; diff --git a/src/proto/h1/role.rs b/src/proto/h1/role.rs index 983ef76ef7..dd8b6b5e86 100644 --- a/src/proto/h1/role.rs +++ b/src/proto/h1/role.rs @@ -1091,7 +1091,7 @@ impl Http1Transaction for Client { #[cfg(feature = "ffi")] if head.subject.is_informational() { if let Some(callback) = ctx.on_informational { - callback.call(head.into_response(crate::Body::empty())); + callback.call(head.into_response(crate::Recv::empty())); } } diff --git a/src/proto/h2/client.rs b/src/proto/h2/client.rs index 013f6fb5a8..9933a84a05 100644 --- a/src/proto/h2/client.rs +++ b/src/proto/h2/client.rs @@ -19,9 +19,9 @@ use crate::headers; use crate::proto::h2::UpgradedSendStream; use crate::proto::Dispatched; use crate::upgrade::Upgraded; -use crate::{Body, Request, Response}; +use crate::{Recv, Request, Response}; -type ClientRx = crate::client::dispatch::Receiver, Response>; +type ClientRx = crate::client::dispatch::Receiver, Response>; ///// An mpsc channel is used to help notify the `Connection` task when *all* ///// other handles to it have been dropped, so that it can shutdown. @@ -336,7 +336,7 @@ where )); } let (parts, recv_stream) = res.into_parts(); - let mut res = Response::from_parts(parts, Body::empty()); + let mut res = Response::from_parts(parts, Recv::empty()); let (pending, on_upgrade) = crate::upgrade::pending(); let io = H2Upgraded { @@ -354,7 +354,7 @@ where } else { let res = res.map(|stream| { let ping = ping.for_stream(&stream); - crate::Body::h2(stream, content_length.into(), ping) + crate::Recv::h2(stream, content_length.into(), ping) }); Ok(res) } diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs index 0a539692de..47c6912746 100644 --- a/src/proto/h2/server.rs +++ b/src/proto/h2/server.rs @@ -23,7 +23,7 @@ use crate::proto::Dispatched; use crate::service::HttpService; use crate::upgrade::{OnUpgrade, Pending, Upgraded}; -use crate::{Body, Response}; +use crate::{Recv, Response}; // Our defaults are chosen for the "majority" case, which usually are not // resource constrained, and so the spec default of 64kb can be too limiting @@ -76,7 +76,7 @@ impl Default for Config { pin_project! { pub(crate) struct Server where - S: HttpService, + S: HttpService, B: HttpBody, { exec: E, @@ -109,7 +109,7 @@ where impl Server where T: AsyncRead + AsyncWrite + Unpin, - S: HttpService, + S: HttpService, S::Error: Into>, B: HttpBody + 'static, E: ConnStreamExec, @@ -181,7 +181,7 @@ where impl Future for Server where T: AsyncRead + AsyncWrite + Unpin, - S: HttpService, + S: HttpService, S::Error: Into>, B: HttpBody + 'static, E: ConnStreamExec, @@ -236,7 +236,7 @@ where exec: &mut E, ) -> Poll> where - S: HttpService, + S: HttpService, S::Error: Into>, E: ConnStreamExec, { @@ -295,7 +295,7 @@ where ( Request::from_parts( parts, - crate::Body::h2(stream, content_length.into(), ping), + crate::Recv::h2(stream, content_length.into(), ping), ), None, ) @@ -309,7 +309,7 @@ where debug_assert!(parts.extensions.get::().is_none()); parts.extensions.insert(upgrade); ( - Request::from_parts(parts, crate::Body::empty()), + Request::from_parts(parts, crate::Recv::empty()), Some(ConnectParts { pending, ping, diff --git a/src/server/conn.rs b/src/server/conn.rs index a447f2bcc0..ceb7b61a5f 100644 --- a/src/server/conn.rs +++ b/src/server/conn.rs @@ -35,7 +35,7 @@ //! } //! } //! -//! async fn hello(_req: Request) -> Result>, Infallible> { +//! async fn hello(_req: Request) -> Result>, Infallible> { //! Ok(Response::new(Full::new(Bytes::from("Hello World!")))) //! } //! # } @@ -67,7 +67,7 @@ cfg_feature! { use tokio::io::{AsyncRead, AsyncWrite}; use tracing::trace; - use crate::body::{Body, HttpBody}; + use crate::body::{Recv, HttpBody}; use crate::common::{task, Future, Pin, Poll, Unpin}; #[cfg(not(all(feature = "http1", feature = "http2")))] use crate::common::Never; @@ -124,7 +124,7 @@ pin_project! { #[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))] pub struct Connection where - S: HttpService, + S: HttpService, { pub(super) conn: Option>, fallback: Fallback, @@ -133,7 +133,7 @@ pin_project! { #[cfg(feature = "http1")] type Http1Dispatcher = - proto::h1::Dispatcher, B, T, proto::ServerTransaction>; + proto::h1::Dispatcher, B, T, proto::ServerTransaction>; #[cfg(all(not(feature = "http1"), feature = "http2"))] type Http1Dispatcher = (Never, PhantomData<(T, Box>, Box>)>); @@ -152,7 +152,7 @@ pin_project! { #[project = ProtoServerProj] pub(super) enum ProtoServer where - S: HttpService, + S: HttpService, B: HttpBody, { H1 { @@ -577,14 +577,14 @@ impl Http { /// # Example /// /// ``` - /// # use hyper::{Body, Request, Response}; + /// # use hyper::{Recv, Request, Response}; /// # use hyper::service::Service; /// # use hyper::server::conn::Http; /// # use tokio::io::{AsyncRead, AsyncWrite}; /// # async fn run(some_io: I, some_service: S) /// # where /// # I: AsyncRead + AsyncWrite + Unpin + Send + 'static, - /// # S: Service, Response=hyper::Response> + Send + 'static, + /// # S: Service, Response=hyper::Response> + Send + 'static, /// # S::Error: Into>, /// # S::Future: Send, /// # { @@ -599,7 +599,7 @@ impl Http { /// ``` pub fn serve_connection(&self, io: I, service: S) -> Connection where - S: HttpService, + S: HttpService, S::Error: Into>, Bd: HttpBody + 'static, Bd::Error: Into>, @@ -679,7 +679,7 @@ impl Http { #[cfg(any(feature = "http1", feature = "http2"))] impl Connection where - S: HttpService, + S: HttpService, S::Error: Into>, I: AsyncRead + AsyncWrite + Unpin, B: HttpBody + 'static, @@ -849,7 +849,7 @@ where #[cfg(any(feature = "http1", feature = "http2"))] impl Future for Connection where - S: HttpService, + S: HttpService, S::Error: Into>, I: AsyncRead + AsyncWrite + Unpin + 'static, B: HttpBody + 'static, @@ -896,7 +896,7 @@ where #[cfg(any(feature = "http1", feature = "http2"))] impl fmt::Debug for Connection where - S: HttpService, + S: HttpService, { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Connection").finish() @@ -929,7 +929,7 @@ impl Default for ConnectionMode { impl Future for ProtoServer where T: AsyncRead + AsyncWrite + Unpin, - S: HttpService, + S: HttpService, S::Error: Into>, B: HttpBody + 'static, B::Error: Into>, @@ -964,14 +964,14 @@ mod upgrades { #[allow(missing_debug_implementations)] pub struct UpgradeableConnection where - S: HttpService, + S: HttpService, { pub(super) inner: Connection, } impl UpgradeableConnection where - S: HttpService, + S: HttpService, S::Error: Into>, I: AsyncRead + AsyncWrite + Unpin, B: HttpBody + 'static, @@ -989,7 +989,7 @@ mod upgrades { impl Future for UpgradeableConnection where - S: HttpService, + S: HttpService, S::Error: Into>, I: AsyncRead + AsyncWrite + Unpin + Send + 'static, B: HttpBody + 'static, diff --git a/src/service/util.rs b/src/service/util.rs index 241d685b11..363133e3d4 100644 --- a/src/service/util.rs +++ b/src/service/util.rs @@ -12,11 +12,11 @@ use crate::{Request, Response}; /// /// ``` /// use bytes::Bytes; -/// use hyper::{Body, Request, Response, Version}; +/// use hyper::{Recv, Request, Response, Version}; /// use http_body_util::Full; /// use hyper::service::service_fn; /// -/// let service = service_fn(|req: Request| async move { +/// let service = service_fn(|req: Request| async move { /// if req.version() == Version::HTTP_11 { /// Ok(Response::new(Full::::from("Hello World"))) /// } else { diff --git a/tests/client.rs b/tests/client.rs index 6540a04459..2bf0eedd14 100644 --- a/tests/client.rs +++ b/tests/client.rs @@ -1342,7 +1342,7 @@ mod conn { use hyper::body::HttpBody; use hyper::client::conn; - use hyper::{self, Body, Method, Request, Response, StatusCode}; + use hyper::{self, Method, Recv, Request, Response, StatusCode}; use super::{concat, s, support, tcp_connect, FutureHyperExt}; @@ -1524,7 +1524,7 @@ mod conn { rt.spawn(conn.map_err(|e| panic!("conn error: {}", e)).map(|_| ())); - let (mut sender, body) = Body::channel(); + let (mut sender, body) = Recv::channel(); let sender = thread::spawn(move || { sender.try_send_data("hello".into()).expect("try_send_data"); support::runtime().block_on(rx).unwrap(); @@ -1887,7 +1887,7 @@ mod conn { res = listener.accept() => { let (stream, _) = res.unwrap(); - let service = service_fn(|_:Request| future::ok::<_, hyper::Error>(Response::new(Empty::::new()))); + let service = service_fn(|_:Request| future::ok::<_, hyper::Error>(Response::new(Empty::::new()))); let mut shdn_rx = shdn_rx.clone(); tokio::task::spawn(async move { @@ -1980,7 +1980,7 @@ mod conn { .http2_keep_alive_timeout(Duration::from_secs(1)) // enable while idle since we aren't sending requests .http2_keep_alive_while_idle(true) - .handshake::<_, Body>(io) + .handshake::<_, Recv>(io) .await .expect("http handshake"); @@ -2011,7 +2011,7 @@ mod conn { .http2_only(true) .http2_keep_alive_interval(Duration::from_secs(1)) .http2_keep_alive_timeout(Duration::from_secs(1)) - .handshake::<_, Body>(io) + .handshake::<_, Recv>(io) .await .expect("http handshake"); @@ -2111,7 +2111,7 @@ mod conn { .http2_only(true) .http2_keep_alive_interval(Duration::from_secs(1)) .http2_keep_alive_timeout(Duration::from_secs(1)) - .handshake::<_, Body>(io) + .handshake::<_, Recv>(io) .await .expect("http handshake"); @@ -2120,7 +2120,7 @@ mod conn { }); // Use a channel to keep request stream open - let (_tx, body) = hyper::Body::channel(); + let (_tx, body) = hyper::Recv::channel(); let req1 = http::Request::new(body); let _resp = client.send_request(req1).await.expect("send_request"); diff --git a/tests/server.rs b/tests/server.rs index 72581d806c..c520d21775 100644 --- a/tests/server.rs +++ b/tests/server.rs @@ -28,7 +28,7 @@ use tokio::net::{TcpListener as TkTcpListener, TcpListener, TcpStream as TkTcpSt use hyper::body::HttpBody; use hyper::server::conn::Http; use hyper::service::service_fn; -use hyper::{Body, Method, Request, Response, StatusCode, Uri, Version}; +use hyper::{Method, Recv, Request, Response, StatusCode, Uri, Version}; mod support; @@ -1284,7 +1284,7 @@ async fn disconnect_after_reading_request_before_responding() { socket, service_fn(|_| { tokio::time::sleep(Duration::from_secs(2)).map( - |_| -> Result, hyper::Error> { + |_| -> Result, hyper::Error> { panic!("response future should have been dropped"); }, ) @@ -1611,7 +1611,7 @@ async fn upgrades_new() { }); let (upgrades_tx, upgrades_rx) = mpsc::channel(); - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { let on_upgrade = hyper::upgrade::on(req); let _ = upgrades_tx.send(on_upgrade); future::ok::<_, hyper::Error>( @@ -1653,7 +1653,7 @@ async fn upgrades_ignored() { let addr = listener.local_addr().unwrap(); tokio::spawn(async move { - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { assert_eq!(req.headers()["upgrade"], "yolo"); future::ok::<_, hyper::Error>(Response::new(Empty::::new())) }); @@ -1720,7 +1720,7 @@ async fn http_connect_new() { }); let (upgrades_tx, upgrades_rx) = mpsc::channel(); - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { let on_upgrade = hyper::upgrade::on(req); let _ = upgrades_tx.send(on_upgrade); future::ok::<_, hyper::Error>( @@ -1791,7 +1791,7 @@ async fn h2_connect() { assert!(recv_stream.data().await.unwrap().unwrap().is_empty()); }); - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { let on_upgrade = hyper::upgrade::on(req); tokio::spawn(async move { @@ -1880,7 +1880,7 @@ async fn h2_connect_multiplex() { futures.for_each(future::ready).await; }); - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { let authority = req.uri().authority().unwrap().to_string(); let on_upgrade = hyper::upgrade::on(req); @@ -1976,7 +1976,7 @@ async fn h2_connect_large_body() { assert!(recv_stream.data().await.unwrap().unwrap().is_empty()); }); - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { let on_upgrade = hyper::upgrade::on(req); tokio::spawn(async move { @@ -2050,7 +2050,7 @@ async fn h2_connect_empty_frames() { assert!(recv_stream.data().await.unwrap().unwrap().is_empty()); }); - let svc = service_fn(move |req: Request| { + let svc = service_fn(move |req: Request| { let on_upgrade = hyper::upgrade::on(req); tokio::spawn(async move { @@ -2307,8 +2307,8 @@ fn http2_body_user_error_sends_reset_reason() { struct Http2ReadyErrorSvc; -impl tower_service::Service> for Http2ReadyErrorSvc { - type Response = Response; +impl tower_service::Service> for Http2ReadyErrorSvc { + type Response = Response; type Error = h2::Error; type Future = Box< dyn futures_core::Future> @@ -2323,7 +2323,7 @@ impl tower_service::Service> for Http2ReadyErrorSvc { ))) } - fn call(&mut self, _: hyper::Request) -> Self::Future { + fn call(&mut self, _: hyper::Request) -> Self::Future { unreachable!("poll_ready error should have shutdown conn"); } } @@ -2781,7 +2781,7 @@ enum Msg { End, } -impl tower_service::Service> for TestService { +impl tower_service::Service> for TestService { type Response = Response; type Error = BoxError; type Future = BoxFuture; @@ -2790,7 +2790,7 @@ impl tower_service::Service> for TestService { Ok(()).into() } - fn call(&mut self, mut req: Request) -> Self::Future { + fn call(&mut self, mut req: Request) -> Self::Future { let tx = self.tx.clone(); let replies = self.reply.clone(); @@ -2848,7 +2848,7 @@ const HELLO: &str = "hello"; struct HelloWorld; -impl tower_service::Service> for HelloWorld { +impl tower_service::Service> for HelloWorld { type Response = Response>; type Error = hyper::Error; type Future = future::Ready>; @@ -2857,14 +2857,14 @@ impl tower_service::Service> for HelloWorld { Ok(()).into() } - fn call(&mut self, _req: Request) -> Self::Future { + fn call(&mut self, _req: Request) -> Self::Future { let response = Response::new(Full::new(HELLO.into())); future::ok(response) } } fn unreachable_service() -> impl tower_service::Service< - http::Request, + http::Request, Response = http::Response, Error = BoxError, Future = BoxFuture, @@ -3127,7 +3127,7 @@ impl TestClient { self } - async fn get(&self, uri: Uri) -> Result, hyper::Error> { + async fn get(&self, uri: Uri) -> Result, hyper::Error> { self.request( Request::builder() .uri(uri) @@ -3138,7 +3138,7 @@ impl TestClient { .await } - async fn request(&self, req: Request>) -> Result, hyper::Error> { + async fn request(&self, req: Request>) -> Result, hyper::Error> { let host = req.uri().host().expect("uri has no host"); let port = req.uri().port_u16().expect("uri has no port"); diff --git a/tests/support/mod.rs b/tests/support/mod.rs index 3028105433..95bd576c73 100644 --- a/tests/support/mod.rs +++ b/tests/support/mod.rs @@ -13,7 +13,7 @@ use hyper::server::conn::Http; use tokio::net::{TcpListener, TcpStream}; use hyper::service::service_fn; -use hyper::{Body, Request, Response, Version}; +use hyper::{Recv, Request, Response, Version}; pub use futures_util::{ future, FutureExt as _, StreamExt as _, TryFutureExt as _, TryStreamExt as _, @@ -358,7 +358,7 @@ async fn async_test(cfg: __TestConfig) { // Move a clone into the service_fn let serve_handles = serve_handles.clone(); - let service = service_fn(move |req: Request| { + let service = service_fn(move |req: Request| { let (sreq, sres) = serve_handles.lock().unwrap().remove(0); assert_eq!(req.uri().path(), sreq.uri, "client path"); @@ -525,7 +525,7 @@ async fn naive_proxy(cfg: ProxyConfig) -> (SocketAddr, impl Future) let mut builder = Response::builder().status(parts.status); *builder.headers_mut().unwrap() = parts.headers; - Result::, hyper::Error>::Ok(builder.body(body).unwrap()) + Result::, hyper::Error>::Ok(builder.body(body).unwrap()) } });