diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 55af5a380..7d0520d52 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,13 +38,19 @@ jobs: with: command: generate-lockfile - name: Cache Dependencies - uses: Swatinem/rust-cache@v1.0.1 + uses: Swatinem/rust-cache@v1.2.0 + + - name: Install cargo-hack + uses: actions-rs/cargo@v1 + with: + command: install + args: cargo-hack - name: check minimal uses: actions-rs/cargo@v1 with: - command: check - args: --workspace --no-default-features --tests + command: hack + args: --clean-per-run check --workspace --no-default-features --tests - name: check full uses: actions-rs/cargo@v1 @@ -81,7 +87,7 @@ jobs: && github.ref == 'refs/heads/master' run: | cargo install cargo-tarpaulin --vers "^0.13" - cargo tarpaulin --out Xml + cargo tarpaulin --out Xml --verbose - name: Upload to Codecov if: > matrix.target.os == 'ubuntu-latest' diff --git a/CHANGES.md b/CHANGES.md index 954410b30..743f5066b 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,6 +1,10 @@ # Changes ## Unreleased - 2021-xx-xx +### Changed +* Feature `cookies` is now optional and enabled by default. [#1981] + +[#1981]: https://github.com/actix/actix-web/pull/1981 ## 4.0.0-beta.3 - 2021-02-10 diff --git a/Cargo.toml b/Cargo.toml index b0302b352..1a1b8645c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,7 @@ license = "MIT OR Apache-2.0" edition = "2018" [package.metadata.docs.rs] +# features that docs.rs will build with features = ["openssl", "rustls", "compress", "secure-cookies"] [badges] @@ -38,12 +39,15 @@ members = [ ] [features] -default = ["compress"] +default = ["compress", "cookies"] # content-encoding support compress = ["actix-http/compress", "awc/compress"] -# sessions feature +# support for cookies +cookies = ["actix-http/cookies", "awc/cookies"] + +# secure cookies feature secure-cookies = ["actix-http/secure-cookies"] # openssl @@ -95,17 +99,17 @@ futures-core = { version = "0.3.7", default-features = false } futures-util = { version = "0.3.7", default-features = false } log = "0.4" mime = "0.3" -socket2 = "0.3.16" pin-project = "1.0.0" regex = "1.4" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" serde_urlencoded = "0.7" +smallvec = "1.6" +socket2 = "0.3.16" time = { version = "0.2.23", default-features = false, features = ["std"] } -url = "2.1" tls-openssl = { package = "openssl", version = "0.10.9", optional = true } tls-rustls = { package = "rustls", version = "0.19.0", optional = true } -smallvec = "1.6" +url = "2.1" [target.'cfg(windows)'.dependencies.tls-openssl] version = "0.10.9" @@ -114,16 +118,13 @@ features = ["vendored"] optional = true [dev-dependencies] -actix = { version = "0.11.0-beta.2", default-features = false } -rand = "0.8" -env_logger = "0.8" -serde_derive = "1.0" brotli2 = "0.3.2" -flate2 = "1.0.13" criterion = "0.3" - -[profile.dev] -debug = false +env_logger = "0.8" +flate2 = "1.0.13" +rand = "0.8" +rcgen = "0.8" +serde_derive = "1.0" [profile.release] lto = true diff --git a/actix-files/Cargo.toml b/actix-files/Cargo.toml index 45fa18a69..08b7b36fc 100644 --- a/actix-files/Cargo.toml +++ b/actix-files/Cargo.toml @@ -25,6 +25,7 @@ bitflags = "1" bytes = "1" futures-core = { version = "0.3.7", default-features = false } futures-util = { version = "0.3.7", default-features = false } +http-range = "0.1.4" derive_more = "0.99.5" log = "0.4" mime = "0.3" diff --git a/actix-files/src/range.rs b/actix-files/src/range.rs index 6718980cb..8d9fe9445 100644 --- a/actix-files/src/range.rs +++ b/actix-files/src/range.rs @@ -10,9 +10,6 @@ pub struct HttpRange { pub length: u64, } -const PREFIX: &str = "bytes="; -const PREFIX_LEN: usize = 6; - #[derive(Debug, Clone, Display, Error)] #[display(fmt = "Parse HTTP Range failed")] pub struct ParseRangeErr(#[error(not(source))] ()); @@ -23,82 +20,16 @@ impl HttpRange { /// `header` is HTTP Range header (e.g. `bytes=bytes=0-9`). /// `size` is full size of response (file). pub fn parse(header: &str, size: u64) -> Result, ParseRangeErr> { - if header.is_empty() { - return Ok(Vec::new()); + match http_range::HttpRange::parse(header, size) { + Ok(ranges) => Ok(ranges + .iter() + .map(|range| HttpRange { + start: range.start, + length: range.length, + }) + .collect()), + Err(_) => Err(ParseRangeErr(())), } - if !header.starts_with(PREFIX) { - return Err(ParseRangeErr(())); - } - - let size_sig = size as i64; - let mut no_overlap = false; - - let all_ranges: Vec> = header[PREFIX_LEN..] - .split(',') - .map(|x| x.trim()) - .filter(|x| !x.is_empty()) - .map(|ra| { - let mut start_end_iter = ra.split('-'); - - let start_str = start_end_iter.next().ok_or(ParseRangeErr(()))?.trim(); - let end_str = start_end_iter.next().ok_or(ParseRangeErr(()))?.trim(); - - if start_str.is_empty() { - // If no start is specified, end specifies the - // range start relative to the end of the file. - let mut length: i64 = end_str.parse().map_err(|_| ParseRangeErr(()))?; - - if length > size_sig { - length = size_sig; - } - - Ok(Some(HttpRange { - start: (size_sig - length) as u64, - length: length as u64, - })) - } else { - let start: i64 = start_str.parse().map_err(|_| ParseRangeErr(()))?; - - if start < 0 { - return Err(ParseRangeErr(())); - } - if start >= size_sig { - no_overlap = true; - return Ok(None); - } - - let length = if end_str.is_empty() { - // If no end is specified, range extends to end of the file. - size_sig - start - } else { - let mut end: i64 = end_str.parse().map_err(|_| ParseRangeErr(()))?; - - if start > end { - return Err(ParseRangeErr(())); - } - - if end >= size_sig { - end = size_sig - 1; - } - - end - start + 1 - }; - - Ok(Some(HttpRange { - start: start as u64, - length: length as u64, - })) - } - }) - .collect::>()?; - - let ranges: Vec = all_ranges.into_iter().filter_map(|x| x).collect(); - - if no_overlap && ranges.is_empty() { - return Err(ParseRangeErr(())); - } - - Ok(ranges) } } diff --git a/actix-http/CHANGES.md b/actix-http/CHANGES.md index b781fe50e..6ba111eb3 100644 --- a/actix-http/CHANGES.md +++ b/actix-http/CHANGES.md @@ -1,6 +1,15 @@ # Changes ## Unreleased - 2021-xx-xx +### Changed +* Feature `cookies` is now optional and disabled by default. [#1981] + +### Removed +* re-export of `futures_channel::oneshot::Canceled` is removed from `error` mod. [#1994] +* `ResponseError` impl for `futures_channel::oneshot::Canceled` is removed. [#1994] + +[#1981]: https://github.com/actix/actix-web/pull/1981 +[#1994]: https://github.com/actix/actix-web/pull/1994 ## 3.0.0-beta.3 - 2021-02-10 diff --git a/actix-http/Cargo.toml b/actix-http/Cargo.toml index 2c4775898..78fb55079 100644 --- a/actix-http/Cargo.toml +++ b/actix-http/Cargo.toml @@ -15,7 +15,8 @@ license = "MIT OR Apache-2.0" edition = "2018" [package.metadata.docs.rs] -features = ["openssl", "rustls", "compress", "secure-cookies"] +# features that docs.rs will build with +features = ["openssl", "rustls", "compress", "cookies", "secure-cookies"] [lib] name = "actix_http" @@ -30,11 +31,14 @@ openssl = ["actix-tls/openssl"] # rustls support rustls = ["actix-tls/rustls"] -# enable compressison support +# enable compression support compress = ["flate2", "brotli2"] +# support for cookies +cookies = ["cookie"] + # support for secure cookies -secure-cookies = ["cookie/secure"] +secure-cookies = ["cookies", "cookie/secure"] # trust-dns as client dns resolver trust-dns = ["trust-dns-resolver"] @@ -46,24 +50,23 @@ actix-utils = "3.0.0-beta.2" actix-rt = "2" actix-tls = "3.0.0-beta.2" +ahash = "0.7" base64 = "0.13" bitflags = "1.2" bytes = "1" bytestring = "1" -cookie = { version = "0.14.1", features = ["percent-encode"] } +cfg-if = "1" +cookie = { version = "0.14.1", features = ["percent-encode"], optional = true } derive_more = "0.99.5" encoding_rs = "0.8" -futures-channel = { version = "0.3.7", default-features = false, features = ["alloc"] } futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] } futures-util = { version = "0.3.7", default-features = false, features = ["alloc", "sink"] } -ahash = "0.7" h2 = "0.3.0" http = "0.2.2" httparse = "1.3" -indexmap = "1.3" itoa = "0.4" -lazy_static = "1.4" language-tags = "0.2" +lazy_static = "1.4" log = "0.4" mime = "0.3" percent-encoding = "2.1" @@ -72,11 +75,11 @@ rand = "0.8" regex = "1.3" serde = "1.0" serde_json = "1.0" +serde_urlencoded = "0.7" sha-1 = "0.9" smallvec = "1.6" -slab = "0.4" -serde_urlencoded = "0.7" time = { version = "0.2.23", default-features = false, features = ["std"] } +tokio = { version = "1.2", features = ["sync"] } # compression brotli2 = { version="0.3.2", optional = true } @@ -90,6 +93,7 @@ actix-http-test = { version = "3.0.0-beta.2", features = ["openssl"] } actix-tls = { version = "3.0.0-beta.2", features = ["openssl"] } criterion = "0.3" env_logger = "0.8" +rcgen = "0.8" serde_derive = "1.0" tls-openssl = { version = "0.10", package = "openssl" } tls-rustls = { version = "0.19", package = "rustls" } diff --git a/actix-http/src/body.rs b/actix-http/src/body.rs deleted file mode 100644 index 0dbe93a4a..000000000 --- a/actix-http/src/body.rs +++ /dev/null @@ -1,714 +0,0 @@ -//! Traits and structures to aid consuming and writing HTTP payloads. - -use std::{ - fmt, mem, - pin::Pin, - task::{Context, Poll}, -}; - -use bytes::{Bytes, BytesMut}; -use futures_core::{ready, Stream}; -use pin_project::pin_project; - -use crate::error::Error; - -/// Body size hint. -#[derive(Debug, PartialEq, Copy, Clone)] -pub enum BodySize { - None, - Empty, - Sized(u64), - Stream, -} - -impl BodySize { - pub fn is_eof(&self) -> bool { - matches!(self, BodySize::None | BodySize::Empty | BodySize::Sized(0)) - } -} - -/// Type that implement this trait can be streamed to a peer. -pub trait MessageBody { - fn size(&self) -> BodySize; - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>>; - - downcast_get_type_id!(); -} - -downcast!(MessageBody); - -impl MessageBody for () { - fn size(&self) -> BodySize { - BodySize::Empty - } - - fn poll_next( - self: Pin<&mut Self>, - _: &mut Context<'_>, - ) -> Poll>> { - Poll::Ready(None) - } -} - -impl MessageBody for Box { - fn size(&self) -> BodySize { - self.as_ref().size() - } - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>> { - Pin::new(self.get_mut().as_mut()).poll_next(cx) - } -} - -#[pin_project(project = ResponseBodyProj)] -pub enum ResponseBody { - Body(#[pin] B), - Other(Body), -} - -impl ResponseBody { - pub fn into_body(self) -> ResponseBody { - match self { - ResponseBody::Body(b) => ResponseBody::Other(b), - ResponseBody::Other(b) => ResponseBody::Other(b), - } - } -} - -impl ResponseBody { - pub fn take_body(&mut self) -> ResponseBody { - mem::replace(self, ResponseBody::Other(Body::None)) - } -} - -impl ResponseBody { - pub fn as_ref(&self) -> Option<&B> { - if let ResponseBody::Body(ref b) = self { - Some(b) - } else { - None - } - } -} - -impl MessageBody for ResponseBody { - fn size(&self) -> BodySize { - match self { - ResponseBody::Body(ref body) => body.size(), - ResponseBody::Other(ref body) => body.size(), - } - } - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>> { - match self.project() { - ResponseBodyProj::Body(body) => body.poll_next(cx), - ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx), - } - } -} - -impl Stream for ResponseBody { - type Item = Result; - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { - match self.project() { - ResponseBodyProj::Body(body) => body.poll_next(cx), - ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx), - } - } -} - -/// Represents various types of HTTP message body. -pub enum Body { - /// Empty response. `Content-Length` header is not set. - None, - /// Zero sized response body. `Content-Length` header is set to `0`. - Empty, - /// Specific response body. - Bytes(Bytes), - /// Generic message body. - Message(Box), -} - -impl Body { - /// Create body from slice (copy) - pub fn from_slice(s: &[u8]) -> Body { - Body::Bytes(Bytes::copy_from_slice(s)) - } - - /// Create body from generic message body. - pub fn from_message(body: B) -> Body { - Body::Message(Box::new(body)) - } -} - -impl MessageBody for Body { - fn size(&self) -> BodySize { - match self { - Body::None => BodySize::None, - Body::Empty => BodySize::Empty, - Body::Bytes(ref bin) => BodySize::Sized(bin.len() as u64), - Body::Message(ref body) => body.size(), - } - } - - fn poll_next( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>> { - match self.get_mut() { - Body::None => Poll::Ready(None), - Body::Empty => Poll::Ready(None), - Body::Bytes(ref mut bin) => { - let len = bin.len(); - if len == 0 { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(mem::take(bin)))) - } - } - Body::Message(body) => Pin::new(&mut **body).poll_next(cx), - } - } -} - -impl PartialEq for Body { - fn eq(&self, other: &Body) -> bool { - match *self { - Body::None => matches!(*other, Body::None), - Body::Empty => matches!(*other, Body::Empty), - Body::Bytes(ref b) => match *other { - Body::Bytes(ref b2) => b == b2, - _ => false, - }, - Body::Message(_) => false, - } - } -} - -impl fmt::Debug for Body { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match *self { - Body::None => write!(f, "Body::None"), - Body::Empty => write!(f, "Body::Empty"), - Body::Bytes(ref b) => write!(f, "Body::Bytes({:?})", b), - Body::Message(_) => write!(f, "Body::Message(_)"), - } - } -} - -impl From<&'static str> for Body { - fn from(s: &'static str) -> Body { - Body::Bytes(Bytes::from_static(s.as_ref())) - } -} - -impl From<&'static [u8]> for Body { - fn from(s: &'static [u8]) -> Body { - Body::Bytes(Bytes::from_static(s)) - } -} - -impl From> for Body { - fn from(vec: Vec) -> Body { - Body::Bytes(Bytes::from(vec)) - } -} - -impl From for Body { - fn from(s: String) -> Body { - s.into_bytes().into() - } -} - -impl<'a> From<&'a String> for Body { - fn from(s: &'a String) -> Body { - Body::Bytes(Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(&s))) - } -} - -impl From for Body { - fn from(s: Bytes) -> Body { - Body::Bytes(s) - } -} - -impl From for Body { - fn from(s: BytesMut) -> Body { - Body::Bytes(s.freeze()) - } -} - -impl From for Body { - fn from(v: serde_json::Value) -> Body { - Body::Bytes(v.to_string().into()) - } -} - -impl From> for Body -where - S: Stream> + Unpin + 'static, -{ - fn from(s: SizedStream) -> Body { - Body::from_message(s) - } -} - -impl From> for Body -where - S: Stream> + Unpin + 'static, - E: Into + 'static, -{ - fn from(s: BodyStream) -> Body { - Body::from_message(s) - } -} - -impl MessageBody for Bytes { - fn size(&self) -> BodySize { - BodySize::Sized(self.len() as u64) - } - - fn poll_next( - self: Pin<&mut Self>, - _: &mut Context<'_>, - ) -> Poll>> { - if self.is_empty() { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(mem::take(self.get_mut())))) - } - } -} - -impl MessageBody for BytesMut { - fn size(&self) -> BodySize { - BodySize::Sized(self.len() as u64) - } - - fn poll_next( - self: Pin<&mut Self>, - _: &mut Context<'_>, - ) -> Poll>> { - if self.is_empty() { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(mem::take(self.get_mut()).freeze()))) - } - } -} - -impl MessageBody for &'static str { - fn size(&self) -> BodySize { - BodySize::Sized(self.len() as u64) - } - - fn poll_next( - self: Pin<&mut Self>, - _: &mut Context<'_>, - ) -> Poll>> { - if self.is_empty() { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(Bytes::from_static( - mem::take(self.get_mut()).as_ref(), - )))) - } - } -} - -impl MessageBody for Vec { - fn size(&self) -> BodySize { - BodySize::Sized(self.len() as u64) - } - - fn poll_next( - self: Pin<&mut Self>, - _: &mut Context<'_>, - ) -> Poll>> { - if self.is_empty() { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(Bytes::from(mem::take(self.get_mut()))))) - } - } -} - -impl MessageBody for String { - fn size(&self) -> BodySize { - BodySize::Sized(self.len() as u64) - } - - fn poll_next( - self: Pin<&mut Self>, - _: &mut Context<'_>, - ) -> Poll>> { - if self.is_empty() { - Poll::Ready(None) - } else { - Poll::Ready(Some(Ok(Bytes::from( - mem::take(self.get_mut()).into_bytes(), - )))) - } - } -} - -/// Type represent streaming body. -/// Response does not contain `content-length` header and appropriate transfer encoding is used. -pub struct BodyStream { - stream: S, -} - -impl BodyStream -where - S: Stream> + Unpin, - E: Into, -{ - pub fn new(stream: S) -> Self { - BodyStream { stream } - } -} - -impl MessageBody for BodyStream -where - S: Stream> + Unpin, - E: Into, -{ - fn size(&self) -> BodySize { - BodySize::Stream - } - - /// Attempts to pull out the next value of the underlying [`Stream`]. - /// - /// Empty values are skipped to prevent [`BodyStream`]'s transmission being - /// ended on a zero-length chunk, but rather proceed until the underlying - /// [`Stream`] ends. - fn poll_next( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>> { - loop { - let stream = &mut self.as_mut().stream; - return Poll::Ready(match ready!(Pin::new(stream).poll_next(cx)) { - Some(Ok(ref bytes)) if bytes.is_empty() => continue, - opt => opt.map(|res| res.map_err(Into::into)), - }); - } - } -} - -/// Type represent streaming body. This body implementation should be used -/// if total size of stream is known. Data get sent as is without using transfer encoding. -pub struct SizedStream { - size: u64, - stream: S, -} - -impl SizedStream -where - S: Stream> + Unpin, -{ - pub fn new(size: u64, stream: S) -> Self { - SizedStream { size, stream } - } -} - -impl MessageBody for SizedStream -where - S: Stream> + Unpin, -{ - fn size(&self) -> BodySize { - BodySize::Sized(self.size as u64) - } - - /// Attempts to pull out the next value of the underlying [`Stream`]. - /// - /// Empty values are skipped to prevent [`SizedStream`]'s transmission being - /// ended on a zero-length chunk, but rather proceed until the underlying - /// [`Stream`] ends. - fn poll_next( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll>> { - loop { - let stream = &mut self.as_mut().stream; - return Poll::Ready(match ready!(Pin::new(stream).poll_next(cx)) { - Some(Ok(ref bytes)) if bytes.is_empty() => continue, - val => val, - }); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use futures_util::future::poll_fn; - use futures_util::pin_mut; - use futures_util::stream; - - impl Body { - pub(crate) fn get_ref(&self) -> &[u8] { - match *self { - Body::Bytes(ref bin) => &bin, - _ => panic!(), - } - } - } - - impl ResponseBody { - pub(crate) fn get_ref(&self) -> &[u8] { - match *self { - ResponseBody::Body(ref b) => b.get_ref(), - ResponseBody::Other(ref b) => b.get_ref(), - } - } - } - - #[actix_rt::test] - async fn test_static_str() { - assert_eq!(Body::from("").size(), BodySize::Sized(0)); - assert_eq!(Body::from("test").size(), BodySize::Sized(4)); - assert_eq!(Body::from("test").get_ref(), b"test"); - - assert_eq!("test".size(), BodySize::Sized(4)); - assert_eq!( - poll_fn(|cx| Pin::new(&mut "test").poll_next(cx)) - .await - .unwrap() - .ok(), - Some(Bytes::from("test")) - ); - } - - #[actix_rt::test] - async fn test_static_bytes() { - assert_eq!(Body::from(b"test".as_ref()).size(), BodySize::Sized(4)); - assert_eq!(Body::from(b"test".as_ref()).get_ref(), b"test"); - assert_eq!( - Body::from_slice(b"test".as_ref()).size(), - BodySize::Sized(4) - ); - assert_eq!(Body::from_slice(b"test".as_ref()).get_ref(), b"test"); - let sb = Bytes::from(&b"test"[..]); - pin_mut!(sb); - - assert_eq!(sb.size(), BodySize::Sized(4)); - assert_eq!( - poll_fn(|cx| sb.as_mut().poll_next(cx)).await.unwrap().ok(), - Some(Bytes::from("test")) - ); - } - - #[actix_rt::test] - async fn test_vec() { - assert_eq!(Body::from(Vec::from("test")).size(), BodySize::Sized(4)); - assert_eq!(Body::from(Vec::from("test")).get_ref(), b"test"); - let test_vec = Vec::from("test"); - pin_mut!(test_vec); - - assert_eq!(test_vec.size(), BodySize::Sized(4)); - assert_eq!( - poll_fn(|cx| test_vec.as_mut().poll_next(cx)) - .await - .unwrap() - .ok(), - Some(Bytes::from("test")) - ); - } - - #[actix_rt::test] - async fn test_bytes() { - let b = Bytes::from("test"); - assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4)); - assert_eq!(Body::from(b.clone()).get_ref(), b"test"); - pin_mut!(b); - - assert_eq!(b.size(), BodySize::Sized(4)); - assert_eq!( - poll_fn(|cx| b.as_mut().poll_next(cx)).await.unwrap().ok(), - Some(Bytes::from("test")) - ); - } - - #[actix_rt::test] - async fn test_bytes_mut() { - let b = BytesMut::from("test"); - assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4)); - assert_eq!(Body::from(b.clone()).get_ref(), b"test"); - pin_mut!(b); - - assert_eq!(b.size(), BodySize::Sized(4)); - assert_eq!( - poll_fn(|cx| b.as_mut().poll_next(cx)).await.unwrap().ok(), - Some(Bytes::from("test")) - ); - } - - #[actix_rt::test] - async fn test_string() { - let b = "test".to_owned(); - assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4)); - assert_eq!(Body::from(b.clone()).get_ref(), b"test"); - assert_eq!(Body::from(&b).size(), BodySize::Sized(4)); - assert_eq!(Body::from(&b).get_ref(), b"test"); - pin_mut!(b); - - assert_eq!(b.size(), BodySize::Sized(4)); - assert_eq!( - poll_fn(|cx| b.as_mut().poll_next(cx)).await.unwrap().ok(), - Some(Bytes::from("test")) - ); - } - - #[actix_rt::test] - async fn test_unit() { - assert_eq!(().size(), BodySize::Empty); - assert!(poll_fn(|cx| Pin::new(&mut ()).poll_next(cx)) - .await - .is_none()); - } - - #[actix_rt::test] - async fn test_box() { - let val = Box::new(()); - pin_mut!(val); - assert_eq!(val.size(), BodySize::Empty); - assert!(poll_fn(|cx| val.as_mut().poll_next(cx)).await.is_none()); - } - - #[actix_rt::test] - async fn test_body_eq() { - assert!( - Body::Bytes(Bytes::from_static(b"1")) - == Body::Bytes(Bytes::from_static(b"1")) - ); - assert!(Body::Bytes(Bytes::from_static(b"1")) != Body::None); - } - - #[actix_rt::test] - async fn test_body_debug() { - assert!(format!("{:?}", Body::None).contains("Body::None")); - assert!(format!("{:?}", Body::Empty).contains("Body::Empty")); - assert!(format!("{:?}", Body::Bytes(Bytes::from_static(b"1"))).contains('1')); - } - - #[actix_rt::test] - async fn test_serde_json() { - use serde_json::json; - assert_eq!( - Body::from(serde_json::Value::String("test".into())).size(), - BodySize::Sized(6) - ); - assert_eq!( - Body::from(json!({"test-key":"test-value"})).size(), - BodySize::Sized(25) - ); - } - - mod body_stream { - use super::*; - //use futures::task::noop_waker; - //use futures::stream::once; - - #[actix_rt::test] - async fn skips_empty_chunks() { - let body = BodyStream::new(stream::iter( - ["1", "", "2"] - .iter() - .map(|&v| Ok(Bytes::from(v)) as Result), - )); - pin_mut!(body); - - assert_eq!( - poll_fn(|cx| body.as_mut().poll_next(cx)) - .await - .unwrap() - .ok(), - Some(Bytes::from("1")), - ); - assert_eq!( - poll_fn(|cx| body.as_mut().poll_next(cx)) - .await - .unwrap() - .ok(), - Some(Bytes::from("2")), - ); - } - - /* Now it does not compile as it should - #[actix_rt::test] - async fn move_pinned_pointer() { - let (sender, receiver) = futures::channel::oneshot::channel(); - let mut body_stream = Ok(BodyStream::new(once(async { - let x = Box::new(0i32); - let y = &x; - receiver.await.unwrap(); - let _z = **y; - Ok::<_, ()>(Bytes::new()) - }))); - - let waker = noop_waker(); - let mut context = Context::from_waker(&waker); - pin_mut!(body_stream); - - let _ = body_stream.as_mut().unwrap().poll_next(&mut context); - sender.send(()).unwrap(); - let _ = std::mem::replace(&mut body_stream, Err([0; 32])).unwrap().poll_next(&mut context); - }*/ - } - - mod sized_stream { - use super::*; - - #[actix_rt::test] - async fn skips_empty_chunks() { - let body = SizedStream::new( - 2, - stream::iter(["1", "", "2"].iter().map(|&v| Ok(Bytes::from(v)))), - ); - pin_mut!(body); - assert_eq!( - poll_fn(|cx| body.as_mut().poll_next(cx)) - .await - .unwrap() - .ok(), - Some(Bytes::from("1")), - ); - assert_eq!( - poll_fn(|cx| body.as_mut().poll_next(cx)) - .await - .unwrap() - .ok(), - Some(Bytes::from("2")), - ); - } - } - - #[actix_rt::test] - async fn test_body_casting() { - let mut body = String::from("hello cast"); - let resp_body: &mut dyn MessageBody = &mut body; - let body = resp_body.downcast_ref::().unwrap(); - assert_eq!(body, "hello cast"); - let body = &mut resp_body.downcast_mut::().unwrap(); - body.push('!'); - let body = resp_body.downcast_ref::().unwrap(); - assert_eq!(body, "hello cast!"); - let not_body = resp_body.downcast_ref::<()>(); - assert!(not_body.is_none()); - } -} diff --git a/actix-http/src/body/body.rs b/actix-http/src/body/body.rs new file mode 100644 index 000000000..a3fd7d41c --- /dev/null +++ b/actix-http/src/body/body.rs @@ -0,0 +1,158 @@ +use std::{ + fmt, mem, + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::{Bytes, BytesMut}; +use futures_core::Stream; + +use crate::error::Error; + +use super::{BodySize, BodyStream, MessageBody, SizedStream}; + +/// Represents various types of HTTP message body. +pub enum Body { + /// Empty response. `Content-Length` header is not set. + None, + /// Zero sized response body. `Content-Length` header is set to `0`. + Empty, + /// Specific response body. + Bytes(Bytes), + /// Generic message body. + Message(Box), +} + +impl Body { + /// Create body from slice (copy) + pub fn from_slice(s: &[u8]) -> Body { + Body::Bytes(Bytes::copy_from_slice(s)) + } + + /// Create body from generic message body. + pub fn from_message(body: B) -> Body { + Body::Message(Box::new(body)) + } +} + +impl MessageBody for Body { + fn size(&self) -> BodySize { + match self { + Body::None => BodySize::None, + Body::Empty => BodySize::Empty, + Body::Bytes(ref bin) => BodySize::Sized(bin.len() as u64), + Body::Message(ref body) => body.size(), + } + } + + fn poll_next( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + match self.get_mut() { + Body::None => Poll::Ready(None), + Body::Empty => Poll::Ready(None), + Body::Bytes(ref mut bin) => { + let len = bin.len(); + if len == 0 { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(mem::take(bin)))) + } + } + Body::Message(body) => Pin::new(&mut **body).poll_next(cx), + } + } +} + +impl PartialEq for Body { + fn eq(&self, other: &Body) -> bool { + match *self { + Body::None => matches!(*other, Body::None), + Body::Empty => matches!(*other, Body::Empty), + Body::Bytes(ref b) => match *other { + Body::Bytes(ref b2) => b == b2, + _ => false, + }, + Body::Message(_) => false, + } + } +} + +impl fmt::Debug for Body { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + Body::None => write!(f, "Body::None"), + Body::Empty => write!(f, "Body::Empty"), + Body::Bytes(ref b) => write!(f, "Body::Bytes({:?})", b), + Body::Message(_) => write!(f, "Body::Message(_)"), + } + } +} + +impl From<&'static str> for Body { + fn from(s: &'static str) -> Body { + Body::Bytes(Bytes::from_static(s.as_ref())) + } +} + +impl From<&'static [u8]> for Body { + fn from(s: &'static [u8]) -> Body { + Body::Bytes(Bytes::from_static(s)) + } +} + +impl From> for Body { + fn from(vec: Vec) -> Body { + Body::Bytes(Bytes::from(vec)) + } +} + +impl From for Body { + fn from(s: String) -> Body { + s.into_bytes().into() + } +} + +impl<'a> From<&'a String> for Body { + fn from(s: &'a String) -> Body { + Body::Bytes(Bytes::copy_from_slice(AsRef::<[u8]>::as_ref(&s))) + } +} + +impl From for Body { + fn from(s: Bytes) -> Body { + Body::Bytes(s) + } +} + +impl From for Body { + fn from(s: BytesMut) -> Body { + Body::Bytes(s.freeze()) + } +} + +impl From for Body { + fn from(v: serde_json::Value) -> Body { + Body::Bytes(v.to_string().into()) + } +} + +impl From> for Body +where + S: Stream> + Unpin + 'static, +{ + fn from(s: SizedStream) -> Body { + Body::from_message(s) + } +} + +impl From> for Body +where + S: Stream> + Unpin + 'static, + E: Into + 'static, +{ + fn from(s: BodyStream) -> Body { + Body::from_message(s) + } +} diff --git a/actix-http/src/body/body_stream.rs b/actix-http/src/body/body_stream.rs new file mode 100644 index 000000000..60e33b161 --- /dev/null +++ b/actix-http/src/body/body_stream.rs @@ -0,0 +1,59 @@ +use std::{ + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::Bytes; +use futures_core::{ready, Stream}; + +use crate::error::Error; + +use super::{BodySize, MessageBody}; + +/// Streaming response wrapper. +/// +/// Response does not contain `Content-Length` header and appropriate transfer encoding is used. +pub struct BodyStream { + stream: S, +} + +impl BodyStream +where + S: Stream> + Unpin, + E: Into, +{ + pub fn new(stream: S) -> Self { + BodyStream { stream } + } +} + +impl MessageBody for BodyStream +where + S: Stream> + Unpin, + E: Into, +{ + fn size(&self) -> BodySize { + BodySize::Stream + } + + /// Attempts to pull out the next value of the underlying [`Stream`]. + /// + /// Empty values are skipped to prevent [`BodyStream`]'s transmission being + /// ended on a zero-length chunk, but rather proceed until the underlying + /// [`Stream`] ends. + fn poll_next( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + loop { + let stream = &mut self.as_mut().stream; + + let chunk = match ready!(Pin::new(stream).poll_next(cx)) { + Some(Ok(ref bytes)) if bytes.is_empty() => continue, + opt => opt.map(|res| res.map_err(Into::into)), + }; + + return Poll::Ready(chunk); + } + } +} diff --git a/actix-http/src/body/message_body.rs b/actix-http/src/body/message_body.rs new file mode 100644 index 000000000..012329146 --- /dev/null +++ b/actix-http/src/body/message_body.rs @@ -0,0 +1,142 @@ +//! [`MessageBody`] trait and foreign implementations. + +use std::{ + mem, + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::{Bytes, BytesMut}; + +use crate::error::Error; + +use super::BodySize; + +/// Type that implement this trait can be streamed to a peer. +pub trait MessageBody { + fn size(&self) -> BodySize; + + fn poll_next( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>>; + + downcast_get_type_id!(); +} + +downcast!(MessageBody); + +impl MessageBody for () { + fn size(&self) -> BodySize { + BodySize::Empty + } + + fn poll_next( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll>> { + Poll::Ready(None) + } +} + +impl MessageBody for Box { + fn size(&self) -> BodySize { + self.as_ref().size() + } + + fn poll_next( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + Pin::new(self.get_mut().as_mut()).poll_next(cx) + } +} + +impl MessageBody for Bytes { + fn size(&self) -> BodySize { + BodySize::Sized(self.len() as u64) + } + + fn poll_next( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll>> { + if self.is_empty() { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(mem::take(self.get_mut())))) + } + } +} + +impl MessageBody for BytesMut { + fn size(&self) -> BodySize { + BodySize::Sized(self.len() as u64) + } + + fn poll_next( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll>> { + if self.is_empty() { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(mem::take(self.get_mut()).freeze()))) + } + } +} + +impl MessageBody for &'static str { + fn size(&self) -> BodySize { + BodySize::Sized(self.len() as u64) + } + + fn poll_next( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll>> { + if self.is_empty() { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(Bytes::from_static( + mem::take(self.get_mut()).as_ref(), + )))) + } + } +} + +impl MessageBody for Vec { + fn size(&self) -> BodySize { + BodySize::Sized(self.len() as u64) + } + + fn poll_next( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll>> { + if self.is_empty() { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(Bytes::from(mem::take(self.get_mut()))))) + } + } +} + +impl MessageBody for String { + fn size(&self) -> BodySize { + BodySize::Sized(self.len() as u64) + } + + fn poll_next( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll>> { + if self.is_empty() { + Poll::Ready(None) + } else { + Poll::Ready(Some(Ok(Bytes::from( + mem::take(self.get_mut()).into_bytes(), + )))) + } + } +} diff --git a/actix-http/src/body/mod.rs b/actix-http/src/body/mod.rs new file mode 100644 index 000000000..a4d6ba2b6 --- /dev/null +++ b/actix-http/src/body/mod.rs @@ -0,0 +1,252 @@ +//! Traits and structures to aid consuming and writing HTTP payloads. + +#[allow(clippy::module_inception)] +mod body; +mod body_stream; +mod message_body; +mod response_body; +mod size; +mod sized_stream; + +pub use self::body::Body; +pub use self::body_stream::BodyStream; +pub use self::message_body::MessageBody; +pub use self::response_body::ResponseBody; +pub use self::size::BodySize; +pub use self::sized_stream::SizedStream; + +#[cfg(test)] +mod tests { + use std::pin::Pin; + + use actix_rt::pin; + use bytes::{Bytes, BytesMut}; + use futures_util::{future::poll_fn, stream}; + + use super::*; + + impl Body { + pub(crate) fn get_ref(&self) -> &[u8] { + match *self { + Body::Bytes(ref bin) => &bin, + _ => panic!(), + } + } + } + + impl ResponseBody { + pub(crate) fn get_ref(&self) -> &[u8] { + match *self { + ResponseBody::Body(ref b) => b.get_ref(), + ResponseBody::Other(ref b) => b.get_ref(), + } + } + } + + #[actix_rt::test] + async fn test_static_str() { + assert_eq!(Body::from("").size(), BodySize::Sized(0)); + assert_eq!(Body::from("test").size(), BodySize::Sized(4)); + assert_eq!(Body::from("test").get_ref(), b"test"); + + assert_eq!("test".size(), BodySize::Sized(4)); + assert_eq!( + poll_fn(|cx| Pin::new(&mut "test").poll_next(cx)) + .await + .unwrap() + .ok(), + Some(Bytes::from("test")) + ); + } + + #[actix_rt::test] + async fn test_static_bytes() { + assert_eq!(Body::from(b"test".as_ref()).size(), BodySize::Sized(4)); + assert_eq!(Body::from(b"test".as_ref()).get_ref(), b"test"); + assert_eq!( + Body::from_slice(b"test".as_ref()).size(), + BodySize::Sized(4) + ); + assert_eq!(Body::from_slice(b"test".as_ref()).get_ref(), b"test"); + let sb = Bytes::from(&b"test"[..]); + pin!(sb); + + assert_eq!(sb.size(), BodySize::Sized(4)); + assert_eq!( + poll_fn(|cx| sb.as_mut().poll_next(cx)).await.unwrap().ok(), + Some(Bytes::from("test")) + ); + } + + #[actix_rt::test] + async fn test_vec() { + assert_eq!(Body::from(Vec::from("test")).size(), BodySize::Sized(4)); + assert_eq!(Body::from(Vec::from("test")).get_ref(), b"test"); + let test_vec = Vec::from("test"); + pin!(test_vec); + + assert_eq!(test_vec.size(), BodySize::Sized(4)); + assert_eq!( + poll_fn(|cx| test_vec.as_mut().poll_next(cx)) + .await + .unwrap() + .ok(), + Some(Bytes::from("test")) + ); + } + + #[actix_rt::test] + async fn test_bytes() { + let b = Bytes::from("test"); + assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4)); + assert_eq!(Body::from(b.clone()).get_ref(), b"test"); + pin!(b); + + assert_eq!(b.size(), BodySize::Sized(4)); + assert_eq!( + poll_fn(|cx| b.as_mut().poll_next(cx)).await.unwrap().ok(), + Some(Bytes::from("test")) + ); + } + + #[actix_rt::test] + async fn test_bytes_mut() { + let b = BytesMut::from("test"); + assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4)); + assert_eq!(Body::from(b.clone()).get_ref(), b"test"); + pin!(b); + + assert_eq!(b.size(), BodySize::Sized(4)); + assert_eq!( + poll_fn(|cx| b.as_mut().poll_next(cx)).await.unwrap().ok(), + Some(Bytes::from("test")) + ); + } + + #[actix_rt::test] + async fn test_string() { + let b = "test".to_owned(); + assert_eq!(Body::from(b.clone()).size(), BodySize::Sized(4)); + assert_eq!(Body::from(b.clone()).get_ref(), b"test"); + assert_eq!(Body::from(&b).size(), BodySize::Sized(4)); + assert_eq!(Body::from(&b).get_ref(), b"test"); + pin!(b); + + assert_eq!(b.size(), BodySize::Sized(4)); + assert_eq!( + poll_fn(|cx| b.as_mut().poll_next(cx)).await.unwrap().ok(), + Some(Bytes::from("test")) + ); + } + + #[actix_rt::test] + async fn test_unit() { + assert_eq!(().size(), BodySize::Empty); + assert!(poll_fn(|cx| Pin::new(&mut ()).poll_next(cx)) + .await + .is_none()); + } + + #[actix_rt::test] + async fn test_box() { + let val = Box::new(()); + pin!(val); + assert_eq!(val.size(), BodySize::Empty); + assert!(poll_fn(|cx| val.as_mut().poll_next(cx)).await.is_none()); + } + + #[actix_rt::test] + async fn test_body_eq() { + assert!( + Body::Bytes(Bytes::from_static(b"1")) + == Body::Bytes(Bytes::from_static(b"1")) + ); + assert!(Body::Bytes(Bytes::from_static(b"1")) != Body::None); + } + + #[actix_rt::test] + async fn test_body_debug() { + assert!(format!("{:?}", Body::None).contains("Body::None")); + assert!(format!("{:?}", Body::Empty).contains("Body::Empty")); + assert!(format!("{:?}", Body::Bytes(Bytes::from_static(b"1"))).contains('1')); + } + + #[actix_rt::test] + async fn test_serde_json() { + use serde_json::json; + assert_eq!( + Body::from(serde_json::Value::String("test".into())).size(), + BodySize::Sized(6) + ); + assert_eq!( + Body::from(json!({"test-key":"test-value"})).size(), + BodySize::Sized(25) + ); + } + + #[actix_rt::test] + async fn body_stream_skips_empty_chunks() { + let body = BodyStream::new(stream::iter( + ["1", "", "2"] + .iter() + .map(|&v| Ok(Bytes::from(v)) as Result), + )); + pin!(body); + + assert_eq!( + poll_fn(|cx| body.as_mut().poll_next(cx)) + .await + .unwrap() + .ok(), + Some(Bytes::from("1")), + ); + assert_eq!( + poll_fn(|cx| body.as_mut().poll_next(cx)) + .await + .unwrap() + .ok(), + Some(Bytes::from("2")), + ); + } + + mod sized_stream { + use super::*; + + #[actix_rt::test] + async fn skips_empty_chunks() { + let body = SizedStream::new( + 2, + stream::iter(["1", "", "2"].iter().map(|&v| Ok(Bytes::from(v)))), + ); + pin!(body); + assert_eq!( + poll_fn(|cx| body.as_mut().poll_next(cx)) + .await + .unwrap() + .ok(), + Some(Bytes::from("1")), + ); + assert_eq!( + poll_fn(|cx| body.as_mut().poll_next(cx)) + .await + .unwrap() + .ok(), + Some(Bytes::from("2")), + ); + } + } + + #[actix_rt::test] + async fn test_body_casting() { + let mut body = String::from("hello cast"); + let resp_body: &mut dyn MessageBody = &mut body; + let body = resp_body.downcast_ref::().unwrap(); + assert_eq!(body, "hello cast"); + let body = &mut resp_body.downcast_mut::().unwrap(); + body.push('!'); + let body = resp_body.downcast_ref::().unwrap(); + assert_eq!(body, "hello cast!"); + let not_body = resp_body.downcast_ref::<()>(); + assert!(not_body.is_none()); + } +} diff --git a/actix-http/src/body/response_body.rs b/actix-http/src/body/response_body.rs new file mode 100644 index 000000000..97141e11e --- /dev/null +++ b/actix-http/src/body/response_body.rs @@ -0,0 +1,77 @@ +use std::{ + mem, + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::Bytes; +use futures_core::Stream; +use pin_project::pin_project; + +use crate::error::Error; + +use super::{Body, BodySize, MessageBody}; + +#[pin_project(project = ResponseBodyProj)] +pub enum ResponseBody { + Body(#[pin] B), + Other(Body), +} + +impl ResponseBody { + pub fn into_body(self) -> ResponseBody { + match self { + ResponseBody::Body(b) => ResponseBody::Other(b), + ResponseBody::Other(b) => ResponseBody::Other(b), + } + } +} + +impl ResponseBody { + pub fn take_body(&mut self) -> ResponseBody { + mem::replace(self, ResponseBody::Other(Body::None)) + } +} + +impl ResponseBody { + pub fn as_ref(&self) -> Option<&B> { + if let ResponseBody::Body(ref b) = self { + Some(b) + } else { + None + } + } +} + +impl MessageBody for ResponseBody { + fn size(&self) -> BodySize { + match self { + ResponseBody::Body(ref body) => body.size(), + ResponseBody::Other(ref body) => body.size(), + } + } + + fn poll_next( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + match self.project() { + ResponseBodyProj::Body(body) => body.poll_next(cx), + ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx), + } + } +} + +impl Stream for ResponseBody { + type Item = Result; + + fn poll_next( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + match self.project() { + ResponseBodyProj::Body(body) => body.poll_next(cx), + ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx), + } + } +} diff --git a/actix-http/src/body/size.rs b/actix-http/src/body/size.rs new file mode 100644 index 000000000..775d5b8f1 --- /dev/null +++ b/actix-http/src/body/size.rs @@ -0,0 +1,40 @@ +/// Body size hint. +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum BodySize { + /// Absence of body can be assumed from method or status code. + /// + /// Will skip writing Content-Length header. + None, + + /// Zero size body. + /// + /// Will write `Content-Length: 0` header. + Empty, + + /// Known size body. + /// + /// Will write `Content-Length: N` header. `Sized(0)` is treated the same as `Empty`. + Sized(u64), + + /// Unknown size body. + /// + /// Will not write Content-Length header. Can be used with chunked Transfer-Encoding. + Stream, +} + +impl BodySize { + /// Returns true if size hint indicates no or empty body. + /// + /// ``` + /// # use actix_http::body::BodySize; + /// assert!(BodySize::None.is_eof()); + /// assert!(BodySize::Empty.is_eof()); + /// assert!(BodySize::Sized(0).is_eof()); + /// + /// assert!(!BodySize::Sized(64).is_eof()); + /// assert!(!BodySize::Stream.is_eof()); + /// ``` + pub fn is_eof(&self) -> bool { + matches!(self, BodySize::None | BodySize::Empty | BodySize::Sized(0)) + } +} diff --git a/actix-http/src/body/sized_stream.rs b/actix-http/src/body/sized_stream.rs new file mode 100644 index 000000000..af995a0fb --- /dev/null +++ b/actix-http/src/body/sized_stream.rs @@ -0,0 +1,59 @@ +use std::{ + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::Bytes; +use futures_core::{ready, Stream}; + +use crate::error::Error; + +use super::{BodySize, MessageBody}; + +/// Known sized streaming response wrapper. +/// +/// This body implementation should be used if total size of stream is known. Data get sent as is +/// without using transfer encoding. +pub struct SizedStream { + size: u64, + stream: S, +} + +impl SizedStream +where + S: Stream> + Unpin, +{ + pub fn new(size: u64, stream: S) -> Self { + SizedStream { size, stream } + } +} + +impl MessageBody for SizedStream +where + S: Stream> + Unpin, +{ + fn size(&self) -> BodySize { + BodySize::Sized(self.size as u64) + } + + /// Attempts to pull out the next value of the underlying [`Stream`]. + /// + /// Empty values are skipped to prevent [`SizedStream`]'s transmission being + /// ended on a zero-length chunk, but rather proceed until the underlying + /// [`Stream`] ends. + fn poll_next( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + loop { + let stream = &mut self.as_mut().stream; + + let chunk = match ready!(Pin::new(stream).poll_next(cx)) { + Some(Ok(ref bytes)) if bytes.is_empty() => continue, + val => val, + }; + + return Poll::Ready(chunk); + } + } +} diff --git a/actix-http/src/client/connection.rs b/actix-http/src/client/connection.rs index 4c6a6dcb8..707d5551b 100644 --- a/actix-http/src/client/connection.rs +++ b/actix-http/src/client/connection.rs @@ -1,4 +1,3 @@ -use std::future::Future; use std::ops::{Deref, DerefMut}; use std::pin::Pin; use std::task::{Context, Poll}; @@ -8,7 +7,6 @@ use actix_codec::{AsyncRead, AsyncWrite, Framed, ReadBuf}; use actix_rt::task::JoinHandle; use bytes::Bytes; use futures_core::future::LocalBoxFuture; -use futures_util::future::{err, Either, FutureExt, Ready}; use h2::client::SendRequest; use pin_project::pin_project; @@ -26,9 +24,10 @@ pub(crate) enum ConnectionType { H2(H2Connection), } -// h2 connection has two parts: SendRequest and Connection. -// Connection is spawned as async task on runtime and H2Connection would hold a handle for -// this task. So it can wake up and quit the task when SendRequest is dropped. +/// `H2Connection` has two parts: `SendRequest` and `Connection`. +/// +/// `Connection` is spawned as an async task on runtime and `H2Connection` holds a handle for +/// this task. Therefore, it can wake up and quit the task when SendRequest is dropped. pub(crate) struct H2Connection { handle: JoinHandle<()>, sender: SendRequest, @@ -74,7 +73,6 @@ impl DerefMut for H2Connection { pub trait Connection { type Io: AsyncRead + AsyncWrite + Unpin; - type Future: Future>; fn protocol(&self) -> Protocol; @@ -83,14 +81,16 @@ pub trait Connection { self, head: H, body: B, - ) -> Self::Future; - - type TunnelFuture: Future< - Output = Result<(ResponseHead, Framed), SendRequestError>, - >; + ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>; /// Send request, returns Response and Framed - fn open_tunnel>(self, head: H) -> Self::TunnelFuture; + fn open_tunnel + 'static>( + self, + head: H, + ) -> LocalBoxFuture< + 'static, + Result<(ResponseHead, Framed), SendRequestError>, + >; } pub(crate) trait ConnectionLifetime: AsyncRead + AsyncWrite + 'static { @@ -103,7 +103,10 @@ pub(crate) trait ConnectionLifetime: AsyncRead + AsyncWrite + 'static { #[doc(hidden)] /// HTTP client connection -pub struct IoConnection { +pub struct IoConnection +where + T: AsyncWrite + Unpin + 'static, +{ io: Option>, created: time::Instant, pool: Option>, @@ -111,7 +114,7 @@ pub struct IoConnection { impl fmt::Debug for IoConnection where - T: fmt::Debug, + T: AsyncWrite + Unpin + fmt::Debug + 'static, { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self.io { @@ -138,6 +141,11 @@ impl IoConnection { pub(crate) fn into_inner(self) -> (ConnectionType, time::Instant) { (self.io.unwrap(), self.created) } + + #[cfg(test)] + pub(crate) fn into_parts(self) -> (ConnectionType, time::Instant, Acquired) { + (self.io.unwrap(), self.created, self.pool.unwrap()) + } } impl Connection for IoConnection @@ -145,8 +153,6 @@ where T: AsyncRead + AsyncWrite + Unpin + 'static, { type Io = T; - type Future = - LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>; fn protocol(&self) -> Protocol { match self.io { @@ -160,33 +166,35 @@ where mut self, head: H, body: B, - ) -> Self::Future { + ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>> { match self.io.take().unwrap() { - ConnectionType::H1(io) => { - h1proto::send_request(io, head.into(), body, self.created, self.pool) - .boxed_local() - } - ConnectionType::H2(io) => { - h2proto::send_request(io, head.into(), body, self.created, self.pool) - .boxed_local() - } + ConnectionType::H1(io) => Box::pin(h1proto::send_request( + io, + head.into(), + body, + self.created, + self.pool, + )), + ConnectionType::H2(io) => Box::pin(h2proto::send_request( + io, + head.into(), + body, + self.created, + self.pool, + )), } } - type TunnelFuture = Either< - LocalBoxFuture< - 'static, - Result<(ResponseHead, Framed), SendRequestError>, - >, - Ready), SendRequestError>>, - >; - /// Send request, returns Response and Framed - fn open_tunnel>(mut self, head: H) -> Self::TunnelFuture { + fn open_tunnel>( + mut self, + head: H, + ) -> LocalBoxFuture< + 'static, + Result<(ResponseHead, Framed), SendRequestError>, + > { match self.io.take().unwrap() { - ConnectionType::H1(io) => { - Either::Left(h1proto::open_tunnel(io, head.into()).boxed_local()) - } + ConnectionType::H1(io) => Box::pin(h1proto::open_tunnel(io, head.into())), ConnectionType::H2(io) => { if let Some(mut pool) = self.pool.take() { pool.release(IoConnection::new( @@ -195,14 +203,18 @@ where None, )); } - Either::Right(err(SendRequestError::TunnelNotSupported)) + Box::pin(async { Err(SendRequestError::TunnelNotSupported) }) } } } } #[allow(dead_code)] -pub(crate) enum EitherConnection { +pub(crate) enum EitherConnection +where + A: AsyncRead + AsyncWrite + Unpin + 'static, + B: AsyncRead + AsyncWrite + Unpin + 'static, +{ A(IoConnection), B(IoConnection), } @@ -213,8 +225,6 @@ where B: AsyncRead + AsyncWrite + Unpin + 'static, { type Io = EitherIo; - type Future = - LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>; fn protocol(&self) -> Protocol { match self { @@ -227,33 +237,30 @@ where self, head: H, body: RB, - ) -> Self::Future { + ) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>> { match self { EitherConnection::A(con) => con.send_request(head, body), EitherConnection::B(con) => con.send_request(head, body), } } - type TunnelFuture = LocalBoxFuture< + /// Send request, returns Response and Framed + fn open_tunnel + 'static>( + self, + head: H, + ) -> LocalBoxFuture< 'static, Result<(ResponseHead, Framed), SendRequestError>, - >; - - /// Send request, returns Response and Framed - fn open_tunnel>(self, head: H) -> Self::TunnelFuture { + > { match self { - EitherConnection::A(con) => con - .open_tunnel(head) - .map(|res| { - res.map(|(head, framed)| (head, framed.into_map_io(EitherIo::A))) - }) - .boxed_local(), - EitherConnection::B(con) => con - .open_tunnel(head) - .map(|res| { - res.map(|(head, framed)| (head, framed.into_map_io(EitherIo::B))) - }) - .boxed_local(), + EitherConnection::A(con) => Box::pin(async { + let (head, framed) = con.open_tunnel(head).await?; + Ok((head, framed.into_map_io(EitherIo::A))) + }), + EitherConnection::B(con) => Box::pin(async { + let (head, framed) = con.open_tunnel(head).await?; + Ok((head, framed.into_map_io(EitherIo::B))) + }), } } } diff --git a/actix-http/src/client/error.rs b/actix-http/src/client/error.rs index 7768462b8..d27363456 100644 --- a/actix-http/src/client/error.rs +++ b/actix-http/src/client/error.rs @@ -25,7 +25,7 @@ pub enum ConnectError { Resolver(Box), /// No dns records - #[display(fmt = "No dns records found for the input")] + #[display(fmt = "No DNS records found for the input")] NoRecords, /// Http2 error diff --git a/actix-http/src/client/h1proto.rs b/actix-http/src/client/h1proto.rs index 92c3c0e1b..082c4b8e2 100644 --- a/actix-http/src/client/h1proto.rs +++ b/actix-http/src/client/h1proto.rs @@ -8,7 +8,7 @@ use bytes::buf::BufMut; use bytes::{Bytes, BytesMut}; use futures_core::Stream; use futures_util::future::poll_fn; -use futures_util::{pin_mut, SinkExt, StreamExt}; +use futures_util::{SinkExt, StreamExt}; use crate::error::PayloadError; use crate::h1; @@ -127,7 +127,7 @@ where T: ConnectionLifetime + Unpin, B: MessageBody, { - pin_mut!(body); + actix_rt::pin!(body); let mut eof = false; while !eof { @@ -165,7 +165,10 @@ where #[doc(hidden)] /// HTTP client connection -pub struct H1Connection { +pub struct H1Connection +where + T: AsyncWrite + Unpin + 'static, +{ /// T should be `Unpin` io: Option, created: time::Instant, diff --git a/actix-http/src/client/h2proto.rs b/actix-http/src/client/h2proto.rs index a70bc1738..0deb5c014 100644 --- a/actix-http/src/client/h2proto.rs +++ b/actix-http/src/client/h2proto.rs @@ -5,7 +5,6 @@ use std::time; use actix_codec::{AsyncRead, AsyncWrite}; use bytes::Bytes; use futures_util::future::poll_fn; -use futures_util::pin_mut; use h2::{ client::{Builder, Connection, SendRequest}, SendStream, @@ -36,6 +35,7 @@ where B: MessageBody, { trace!("Sending client request: {:?} {:?}", head, body.size()); + let head_req = head.as_ref().method == Method::HEAD; let length = body.size(); let eof = matches!( @@ -130,7 +130,7 @@ async fn send_body( mut send: SendStream, ) -> Result<(), SendRequestError> { let mut buf = None; - pin_mut!(body); + actix_rt::pin!(body); loop { if buf.is_none() { match poll_fn(|cx| body.as_mut().poll_next(cx)).await { diff --git a/actix-http/src/client/pool.rs b/actix-http/src/client/pool.rs index 867ba5c0c..3800696fa 100644 --- a/actix-http/src/client/pool.rs +++ b/actix-http/src/client/pool.rs @@ -1,32 +1,30 @@ -use std::cell::RefCell; +//! Client connection pooling keyed on the authority part of the connection URI. + use std::collections::VecDeque; use std::future::Future; +use std::ops::Deref; use std::pin::Pin; use std::rc::Rc; +use std::sync::Arc; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; +use std::{cell::RefCell, io}; -use actix_codec::{AsyncRead, AsyncWrite, ReadBuf}; +use actix_codec::{AsyncRead, AsyncWrite}; use actix_rt::time::{sleep, Sleep}; use actix_service::Service; -use actix_utils::task::LocalWaker; use ahash::AHashMap; -use bytes::Bytes; -use futures_channel::oneshot; use futures_core::future::LocalBoxFuture; -use futures_util::future::{poll_fn, FutureExt}; -use h2::client::{Connection, SendRequest}; use http::uri::Authority; -use indexmap::IndexSet; use pin_project::pin_project; -use slab::Slab; +use tokio::io::ReadBuf; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use super::config::ConnectorConfig; -use super::connection::{ConnectionType, IoConnection}; +use super::connection::{ConnectionType, H2Connection, IoConnection}; use super::error::ConnectError; use super::h2proto::handshake; use super::Connect; -use crate::client::connection::H2Connection; #[derive(Clone, Copy, PartialEq)] /// Protocol version @@ -46,358 +44,286 @@ impl From for Key { } } -/// Connections pool -pub(crate) struct ConnectionPool(Rc, Rc>>); - -impl ConnectionPool +/// Connections pool for reuse Io type for certain [`http::uri::Authority`] as key. +pub(crate) struct ConnectionPool where - Io: AsyncRead + AsyncWrite + Unpin + 'static, - T: Service + 'static, + Io: AsyncWrite + Unpin + 'static, { - pub(crate) fn new(connector: T, config: ConnectorConfig) -> Self { - let connector_rc = Rc::new(connector); - let inner_rc = Rc::new(RefCell::new(Inner { - config, - acquired: 0, - waiters: Slab::new(), - waiters_queue: IndexSet::new(), - available: AHashMap::default(), - waker: LocalWaker::new(), - })); + connector: Rc, + inner: ConnectionPoolInner, +} - // start support future - actix_rt::spawn(ConnectorPoolSupport { - connector: Rc::clone(&connector_rc), - inner: Rc::clone(&inner_rc), - }); +/// wrapper type for check the ref count of Rc. +struct ConnectionPoolInner(Rc>) +where + Io: AsyncWrite + Unpin + 'static; - ConnectionPool(connector_rc, inner_rc) +impl ConnectionPoolInner +where + Io: AsyncWrite + Unpin + 'static, +{ + /// spawn a async for graceful shutdown h1 Io type with a timeout. + fn close(&self, conn: ConnectionType) { + if let Some(timeout) = self.config.disconnect_timeout { + if let ConnectionType::H1(io) = conn { + actix_rt::spawn(CloseConnection::new(io, timeout)); + } + } } } -impl Clone for ConnectionPool +impl Clone for ConnectionPoolInner where - Io: 'static, + Io: AsyncWrite + Unpin + 'static, { fn clone(&self) -> Self { - ConnectionPool(self.0.clone(), self.1.clone()) + Self(Rc::clone(&self.0)) } } -impl Drop for ConnectionPool { - fn drop(&mut self) { - // wake up the ConnectorPoolSupport when dropping so it can exit properly. - self.1.borrow().waker.wake(); - } -} - -impl Service for ConnectionPool +impl Deref for ConnectionPoolInner where + Io: AsyncWrite + Unpin + 'static, +{ + type Target = ConnectionPoolInnerPriv; + + fn deref(&self) -> &Self::Target { + &*self.0 + } +} + +impl Drop for ConnectionPoolInner +where + Io: AsyncWrite + Unpin + 'static, +{ + fn drop(&mut self) { + // When strong count is one it means the pool is dropped + // remove and drop all Io types. + if Rc::strong_count(&self.0) == 1 { + self.permits.close(); + std::mem::take(&mut *self.available.borrow_mut()) + .into_iter() + .for_each(|(_, conns)| { + conns.into_iter().for_each(|pooled| self.close(pooled.conn)) + }); + } + } +} + +struct ConnectionPoolInnerPriv +where + Io: AsyncWrite + Unpin + 'static, +{ + config: ConnectorConfig, + available: RefCell>>>, + permits: Arc, +} + +impl ConnectionPool +where + Io: AsyncWrite + Unpin + 'static, +{ + /// Construct a new connection pool. + /// + /// [`super::config::ConnectorConfig`]'s `limit` is used as the max permits allowed for + /// in-flight connections. + /// + /// The pool can only have equal to `limit` amount of requests spawning/using Io type + /// concurrently. + /// + /// Any requests beyond limit would be wait in fifo order and get notified in async manner + /// by [`tokio::sync::Semaphore`] + pub(crate) fn new(connector: S, config: ConnectorConfig) -> Self { + let permits = Arc::new(Semaphore::new(config.limit)); + let available = RefCell::new(AHashMap::default()); + let connector = Rc::new(connector); + + let inner = ConnectionPoolInner(Rc::new(ConnectionPoolInnerPriv { + config, + available, + permits, + })); + + Self { connector, inner } + } +} + +impl Clone for ConnectionPool +where + Io: AsyncWrite + Unpin + 'static, +{ + fn clone(&self) -> Self { + Self { + connector: self.connector.clone(), + inner: self.inner.clone(), + } + } +} + +impl Service for ConnectionPool +where + S: Service + 'static, Io: AsyncRead + AsyncWrite + Unpin + 'static, - T: Service + 'static, { type Response = IoConnection; type Error = ConnectError; type Future = LocalBoxFuture<'static, Result, ConnectError>>; - fn poll_ready(&self, cx: &mut Context<'_>) -> Poll> { - self.0.poll_ready(cx) - } + actix_service::forward_ready!(connector); fn call(&self, req: Connect) -> Self::Future { - let connector = self.0.clone(); - let inner = self.1.clone(); + let connector = self.connector.clone(); + let inner = self.inner.clone(); - let fut = async move { + Box::pin(async move { let key = if let Some(authority) = req.uri.authority() { authority.clone().into() } else { return Err(ConnectError::Unresolved); }; - // acquire connection - match poll_fn(|cx| Poll::Ready(inner.borrow_mut().acquire(&key, cx))).await { - Acquire::Acquired(io, created) => { - // use existing connection - Ok(IoConnection::new( - io, - created, - Some(Acquired(key, Some(inner))), - )) - } - Acquire::Available => { - // open tcp connection + // acquire an owned permit and carry it with connection + let permit = inner.permits.clone().acquire_owned().await.map_err(|_| { + ConnectError::Io(io::Error::new( + io::ErrorKind::Other, + "failed to acquire semaphore on client connection pool", + )) + })?; + + let conn = { + let mut conn = None; + + // check if there is idle connection for given key. + let mut map = inner.available.borrow_mut(); + + if let Some(conns) = map.get_mut(&key) { + let now = Instant::now(); + + while let Some(mut c) = conns.pop_front() { + let config = &inner.config; + let idle_dur = now - c.used; + let age = now - c.created; + let conn_ineligible = idle_dur > config.conn_keep_alive + || age > config.conn_lifetime; + + if conn_ineligible { + // drop connections that are too old + inner.close(c.conn); + } else { + // check if the connection is still usable + if let ConnectionType::H1(ref mut io) = c.conn { + let check = ConnectionCheckFuture { io }; + match check.await { + ConnectionState::Tainted => { + inner.close(c.conn); + continue; + } + ConnectionState::Skip => continue, + ConnectionState::Live => conn = Some(c), + } + } else { + conn = Some(c); + } + + break; + } + } + }; + + conn + }; + + // construct acquired. It's used to put Io type back to pool/ close the Io type. + // permit is carried with the whole lifecycle of Acquired. + let acquired = Some(Acquired { key, inner, permit }); + + // match the connection and spawn new one if did not get anything. + match conn { + Some(conn) => Ok(IoConnection::new(conn.conn, conn.created, acquired)), + None => { let (io, proto) = connector.call(req).await?; - let config = inner.borrow().config.clone(); - - let guard = OpenGuard::new(key, inner); - if proto == Protocol::Http1 { Ok(IoConnection::new( ConnectionType::H1(io), Instant::now(), - Some(guard.consume()), + acquired, )) } else { - let (sender, connection) = handshake(io, &config).await?; + let config = &acquired.as_ref().unwrap().inner.config; + let (sender, connection) = handshake(io, config).await?; Ok(IoConnection::new( ConnectionType::H2(H2Connection::new(sender, connection)), Instant::now(), - Some(guard.consume()), + acquired, )) } } - _ => { - // connection is not available, wait - let (rx, token) = inner.borrow_mut().wait_for(req); - - let guard = WaiterGuard::new(key, token, inner); - let res = match rx.await { - Err(_) => Err(ConnectError::Disconnected), - Ok(res) => res, - }; - guard.consume(); - res - } } + }) + } +} + +/// Type for check the connection and determine if it's usable. +struct ConnectionCheckFuture<'a, Io> { + io: &'a mut Io, +} + +enum ConnectionState { + /// IO is pending and a new request would wake it. + Live, + + /// IO unexpectedly has unread data and should be dropped. + Tainted, + + /// IO should be skipped but not dropped. + Skip, +} + +impl Future for ConnectionCheckFuture<'_, Io> +where + Io: AsyncRead + Unpin, +{ + type Output = ConnectionState; + + // this future is only used to get access to Context. + // It should never return Poll::Pending. + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + let mut buf = [0; 2]; + let mut read_buf = ReadBuf::new(&mut buf); + + let state = match Pin::new(&mut this.io).poll_read(cx, &mut read_buf) { + Poll::Ready(Ok(())) if !read_buf.filled().is_empty() => { + ConnectionState::Tainted + } + + Poll::Pending => ConnectionState::Live, + _ => ConnectionState::Skip, }; - fut.boxed_local() + Poll::Ready(state) } } -struct WaiterGuard -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - key: Key, - token: usize, - inner: Option>>>, -} - -impl WaiterGuard -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - fn new(key: Key, token: usize, inner: Rc>>) -> Self { - Self { - key, - token, - inner: Some(inner), - } - } - - fn consume(mut self) { - let _ = self.inner.take(); - } -} - -impl Drop for WaiterGuard -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - fn drop(&mut self) { - if let Some(i) = self.inner.take() { - let mut inner = i.as_ref().borrow_mut(); - inner.release_waiter(&self.key, self.token); - inner.check_availability(); - } - } -} - -struct OpenGuard -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - key: Key, - inner: Option>>>, -} - -impl OpenGuard -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - fn new(key: Key, inner: Rc>>) -> Self { - Self { - key, - inner: Some(inner), - } - } - - fn consume(mut self) -> Acquired { - Acquired(self.key.clone(), self.inner.take()) - } -} - -impl Drop for OpenGuard -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - fn drop(&mut self) { - if let Some(i) = self.inner.take() { - let mut inner = i.as_ref().borrow_mut(); - inner.release(); - inner.check_availability(); - } - } -} - -enum Acquire { - Acquired(ConnectionType, Instant), - Available, - NotAvailable, -} - -struct AvailableConnection { - io: ConnectionType, +struct PooledConnection { + conn: ConnectionType, used: Instant, created: Instant, } -pub(crate) struct Inner { - config: ConnectorConfig, - acquired: usize, - available: AHashMap>>, - waiters: Slab< - Option<( - Connect, - oneshot::Sender, ConnectError>>, - )>, - >, - waiters_queue: IndexSet<(Key, usize)>, - waker: LocalWaker, -} - -impl Inner { - fn reserve(&mut self) { - self.acquired += 1; - } - - fn release(&mut self) { - self.acquired -= 1; - } - - fn release_waiter(&mut self, key: &Key, token: usize) { - self.waiters.remove(token); - let _ = self.waiters_queue.shift_remove(&(key.clone(), token)); - } -} - -impl Inner -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - /// connection is not available, wait - fn wait_for( - &mut self, - connect: Connect, - ) -> ( - oneshot::Receiver, ConnectError>>, - usize, - ) { - let (tx, rx) = oneshot::channel(); - - let key: Key = connect.uri.authority().unwrap().clone().into(); - let entry = self.waiters.vacant_entry(); - let token = entry.key(); - entry.insert(Some((connect, tx))); - assert!(self.waiters_queue.insert((key, token))); - - (rx, token) - } - - fn acquire(&mut self, key: &Key, cx: &mut Context<'_>) -> Acquire { - // check limits - if self.config.limit > 0 && self.acquired >= self.config.limit { - return Acquire::NotAvailable; - } - - self.reserve(); - - // check if open connection is available - // cleanup stale connections at the same time - if let Some(ref mut connections) = self.available.get_mut(key) { - let now = Instant::now(); - while let Some(conn) = connections.pop_back() { - // check if it still usable - if (now - conn.used) > self.config.conn_keep_alive - || (now - conn.created) > self.config.conn_lifetime - { - if let Some(timeout) = self.config.disconnect_timeout { - if let ConnectionType::H1(io) = conn.io { - actix_rt::spawn(CloseConnection::new(io, timeout)); - } - } - } else { - let mut io = conn.io; - let mut buf = [0; 2]; - let mut read_buf = ReadBuf::new(&mut buf); - if let ConnectionType::H1(ref mut s) = io { - match Pin::new(s).poll_read(cx, &mut read_buf) { - Poll::Pending => {} - Poll::Ready(Ok(())) if !read_buf.filled().is_empty() => { - if let Some(timeout) = self.config.disconnect_timeout { - if let ConnectionType::H1(io) = io { - actix_rt::spawn(CloseConnection::new( - io, timeout, - )); - } - } - continue; - } - _ => continue, - } - } - return Acquire::Acquired(io, conn.created); - } - } - } - Acquire::Available - } - - fn release_conn(&mut self, key: &Key, io: ConnectionType, created: Instant) { - self.acquired -= 1; - self.available - .entry(key.clone()) - .or_insert_with(VecDeque::new) - .push_back(AvailableConnection { - io, - created, - used: Instant::now(), - }); - self.check_availability(); - } - - fn release_close(&mut self, io: ConnectionType) { - self.acquired -= 1; - if let Some(timeout) = self.config.disconnect_timeout { - if let ConnectionType::H1(io) = io { - actix_rt::spawn(CloseConnection::new(io, timeout)); - } - } - self.check_availability(); - } - - fn check_availability(&self) { - if !self.waiters_queue.is_empty() && self.acquired < self.config.limit { - self.waker.wake(); - } - } -} - -#[pin_project::pin_project] -struct CloseConnection { - io: T, +#[pin_project] +struct CloseConnection { + io: Io, #[pin] timeout: Sleep, } -impl CloseConnection +impl CloseConnection where - T: AsyncWrite + Unpin, + Io: AsyncWrite + Unpin, { - fn new(io: T, timeout: Duration) -> Self { + fn new(io: Io, timeout: Duration) -> Self { CloseConnection { io, timeout: sleep(timeout), @@ -405,9 +331,9 @@ where } } -impl Future for CloseConnection +impl Future for CloseConnection where - T: AsyncWrite + Unpin, + Io: AsyncWrite + Unpin, { type Output = (); @@ -416,230 +342,337 @@ where match this.timeout.poll(cx) { Poll::Ready(_) => Poll::Ready(()), - Poll::Pending => match Pin::new(this.io).poll_shutdown(cx) { - Poll::Ready(_) => Poll::Ready(()), - Poll::Pending => Poll::Pending, - }, + Poll::Pending => Pin::new(this.io).poll_shutdown(cx).map(|_| ()), } } } -#[pin_project] -struct ConnectorPoolSupport +pub(crate) struct Acquired where - Io: AsyncRead + AsyncWrite + Unpin + 'static, + Io: AsyncWrite + Unpin + 'static, { - connector: Rc, - inner: Rc>>, -} - -impl Future for ConnectorPoolSupport -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, - T: Service, - T::Future: 'static, -{ - type Output = (); - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let this = self.project(); - - if Rc::strong_count(this.inner) == 1 { - // If we are last copy of Inner it means the ConnectionPool is already gone - // and we are safe to exit. - return Poll::Ready(()); - } - - let mut inner = this.inner.borrow_mut(); - inner.waker.register(cx.waker()); - - // check waiters - loop { - let (key, token) = { - if let Some((key, token)) = inner.waiters_queue.get_index(0) { - (key.clone(), *token) - } else { - break; - } - }; - if inner.waiters.get(token).unwrap().is_none() { - continue; - } - - match inner.acquire(&key, cx) { - Acquire::NotAvailable => break, - Acquire::Acquired(io, created) => { - let tx = inner.waiters.get_mut(token).unwrap().take().unwrap().1; - if let Err(conn) = tx.send(Ok(IoConnection::new( - io, - created, - Some(Acquired(key.clone(), Some(this.inner.clone()))), - ))) { - let (io, created) = conn.unwrap().into_inner(); - inner.release_conn(&key, io, created); - } - } - Acquire::Available => { - let (connect, tx) = - inner.waiters.get_mut(token).unwrap().take().unwrap(); - OpenWaitingConnection::spawn( - key.clone(), - tx, - this.inner.clone(), - this.connector.call(connect), - inner.config.clone(), - ); - } - } - let _ = inner.waiters_queue.swap_remove_index(0); - } - - Poll::Pending - } -} - -#[pin_project::pin_project(PinnedDrop)] -struct OpenWaitingConnection -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - #[pin] - fut: F, key: Key, - h2: Option< - LocalBoxFuture< - 'static, - Result<(SendRequest, Connection), h2::Error>, - >, - >, - rx: Option, ConnectError>>>, - inner: Option>>>, - config: ConnectorConfig, + inner: ConnectionPoolInner, + permit: OwnedSemaphorePermit, } -impl OpenWaitingConnection +impl Acquired where - F: Future> + 'static, Io: AsyncRead + AsyncWrite + Unpin + 'static, { - fn spawn( - key: Key, - rx: oneshot::Sender, ConnectError>>, - inner: Rc>>, - fut: F, - config: ConnectorConfig, - ) { - actix_rt::spawn(OpenWaitingConnection { - key, - fut, - h2: None, - rx: Some(rx), - inner: Some(inner), - config, + /// Close the IO. + pub(crate) fn close(&mut self, conn: IoConnection) { + let (conn, _) = conn.into_inner(); + self.inner.close(conn); + } + + /// Release IO back into pool. + pub(crate) fn release(&mut self, conn: IoConnection) { + let (io, created) = conn.into_inner(); + let Acquired { key, inner, .. } = self; + + inner + .available + .borrow_mut() + .entry(key.clone()) + .or_insert_with(VecDeque::new) + .push_back(PooledConnection { + conn: io, + created, + used: Instant::now(), + }); + + let _ = &mut self.permit; + } +} + +#[cfg(test)] +mod test { + use std::{cell::Cell, io}; + + use http::Uri; + + use super::*; + use crate::client::connection::IoConnection; + + /// A stream type that always returns pending on async read. + /// + /// Mocks an idle TCP stream that is ready to be used for client connections. + struct TestStream(Rc>); + + impl Drop for TestStream { + fn drop(&mut self) { + self.0.set(self.0.get() - 1); + } + } + + impl AsyncRead for TestStream { + fn poll_read( + self: Pin<&mut Self>, + _: &mut Context<'_>, + _: &mut ReadBuf<'_>, + ) -> Poll> { + Poll::Pending + } + } + + impl AsyncWrite for TestStream { + fn poll_write( + self: Pin<&mut Self>, + _: &mut Context<'_>, + _: &[u8], + ) -> Poll> { + unimplemented!() + } + + fn poll_flush( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll> { + unimplemented!() + } + + fn poll_shutdown( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + } + + struct TestPoolConnector { + generated: Rc>, + } + + impl Service for TestPoolConnector { + type Response = (TestStream, Protocol); + type Error = ConnectError; + type Future = LocalBoxFuture<'static, Result>; + + actix_service::always_ready!(); + + fn call(&self, _: Connect) -> Self::Future { + self.generated.set(self.generated.get() + 1); + let generated = self.generated.clone(); + Box::pin(async { Ok((TestStream(generated), Protocol::Http1)) }) + } + } + + fn release(conn: IoConnection) + where + T: AsyncRead + AsyncWrite + Unpin + 'static, + { + let (conn, created, mut acquired) = conn.into_parts(); + acquired.release(IoConnection::new(conn, created, None)); + } + + #[actix_rt::test] + async fn test_pool_limit() { + let connector = TestPoolConnector { + generated: Rc::new(Cell::new(0)), + }; + + let config = ConnectorConfig { + limit: 1, + ..Default::default() + }; + + let pool = super::ConnectionPool::new(connector, config); + + let req = Connect { + uri: Uri::from_static("http://localhost"), + addr: None, + }; + + let conn = pool.call(req.clone()).await.unwrap(); + + let waiting = Rc::new(Cell::new(true)); + + let waiting_clone = waiting.clone(); + actix_rt::spawn(async move { + actix_rt::time::sleep(Duration::from_millis(100)).await; + waiting_clone.set(false); + drop(conn); }); - } -} - -#[pin_project::pinned_drop] -impl PinnedDrop for OpenWaitingConnection -where - Io: AsyncRead + AsyncWrite + Unpin + 'static, -{ - fn drop(self: Pin<&mut Self>) { - if let Some(inner) = self.project().inner.take() { - let mut inner = inner.as_ref().borrow_mut(); - inner.release(); - inner.check_availability(); - } - } -} - -impl Future for OpenWaitingConnection -where - F: Future>, - Io: AsyncRead + AsyncWrite + Unpin, -{ - type Output = (); - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let this = self.as_mut().project(); - - if let Some(ref mut h2) = this.h2 { - return match Pin::new(h2).poll(cx) { - Poll::Ready(Ok((sender, connection))) => { - let rx = this.rx.take().unwrap(); - let _ = rx.send(Ok(IoConnection::new( - ConnectionType::H2(H2Connection::new(sender, connection)), - Instant::now(), - Some(Acquired(this.key.clone(), this.inner.take())), - ))); - Poll::Ready(()) - } - Poll::Pending => Poll::Pending, - Poll::Ready(Err(err)) => { - let _ = this.inner.take(); - if let Some(rx) = this.rx.take() { - let _ = rx.send(Err(ConnectError::H2(err))); - } - Poll::Ready(()) - } - }; - } - - match this.fut.poll(cx) { - Poll::Ready(Err(err)) => { - let _ = this.inner.take(); - if let Some(rx) = this.rx.take() { - let _ = rx.send(Err(err)); - } - Poll::Ready(()) - } - Poll::Ready(Ok((io, proto))) => { - if proto == Protocol::Http1 { - let rx = this.rx.take().unwrap(); - let _ = rx.send(Ok(IoConnection::new( - ConnectionType::H1(io), - Instant::now(), - Some(Acquired(this.key.clone(), this.inner.take())), - ))); - Poll::Ready(()) - } else { - *this.h2 = Some(handshake(io, this.config).boxed_local()); - self.poll(cx) - } - } - Poll::Pending => Poll::Pending, - } - } -} - -pub(crate) struct Acquired(Key, Option>>>); - -impl Acquired -where - T: AsyncRead + AsyncWrite + Unpin + 'static, -{ - pub(crate) fn close(&mut self, conn: IoConnection) { - if let Some(inner) = self.1.take() { - let (io, _) = conn.into_inner(); - inner.as_ref().borrow_mut().release_close(io); - } - } - pub(crate) fn release(&mut self, conn: IoConnection) { - if let Some(inner) = self.1.take() { - let (io, created) = conn.into_inner(); - inner - .as_ref() - .borrow_mut() - .release_conn(&self.0, io, created); - } - } -} - -impl Drop for Acquired { - fn drop(&mut self) { - if let Some(inner) = self.1.take() { - inner.as_ref().borrow_mut().release(); - } + + assert!(waiting.get()); + + let now = Instant::now(); + let conn = pool.call(req).await.unwrap(); + + release(conn); + assert!(!waiting.get()); + assert!(now.elapsed() >= Duration::from_millis(100)); + } + + #[actix_rt::test] + async fn test_pool_keep_alive() { + let generated = Rc::new(Cell::new(0)); + let generated_clone = generated.clone(); + + let connector = TestPoolConnector { generated }; + + let config = ConnectorConfig { + conn_keep_alive: Duration::from_secs(1), + ..Default::default() + }; + + let pool = super::ConnectionPool::new(connector, config); + + let req = Connect { + uri: Uri::from_static("http://localhost"), + addr: None, + }; + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + actix_rt::time::sleep(Duration::from_millis(1500)).await; + actix_rt::task::yield_now().await; + + let conn = pool.call(req).await.unwrap(); + // Note: spawned recycle connection is not ran yet. + // This is tokio current thread runtime specific behavior. + assert_eq!(2, generated_clone.get()); + + // yield task so the old connection is properly dropped. + actix_rt::task::yield_now().await; + assert_eq!(1, generated_clone.get()); + + release(conn); + } + + #[actix_rt::test] + async fn test_pool_lifetime() { + let generated = Rc::new(Cell::new(0)); + let generated_clone = generated.clone(); + + let connector = TestPoolConnector { generated }; + + let config = ConnectorConfig { + conn_lifetime: Duration::from_secs(1), + ..Default::default() + }; + + let pool = super::ConnectionPool::new(connector, config); + + let req = Connect { + uri: Uri::from_static("http://localhost"), + addr: None, + }; + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + actix_rt::time::sleep(Duration::from_millis(1500)).await; + actix_rt::task::yield_now().await; + + let conn = pool.call(req).await.unwrap(); + // Note: spawned recycle connection is not ran yet. + // This is tokio current thread runtime specific behavior. + assert_eq!(2, generated_clone.get()); + + // yield task so the old connection is properly dropped. + actix_rt::task::yield_now().await; + assert_eq!(1, generated_clone.get()); + + release(conn); + } + + #[actix_rt::test] + async fn test_pool_authority_key() { + let generated = Rc::new(Cell::new(0)); + let generated_clone = generated.clone(); + + let connector = TestPoolConnector { generated }; + + let config = ConnectorConfig::default(); + + let pool = super::ConnectionPool::new(connector, config); + + let req = Connect { + uri: Uri::from_static("https://crates.io"), + addr: None, + }; + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + let conn = pool.call(req).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + let req = Connect { + uri: Uri::from_static("https://google.com"), + addr: None, + }; + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(2, generated_clone.get()); + release(conn); + let conn = pool.call(req).await.unwrap(); + assert_eq!(2, generated_clone.get()); + release(conn); + } + + #[actix_rt::test] + async fn test_pool_drop() { + let generated = Rc::new(Cell::new(0)); + let generated_clone = generated.clone(); + + let connector = TestPoolConnector { generated }; + + let config = ConnectorConfig::default(); + + let pool = Rc::new(super::ConnectionPool::new(connector, config)); + + let req = Connect { + uri: Uri::from_static("https://crates.io"), + addr: None, + }; + + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(1, generated_clone.get()); + release(conn); + + let req = Connect { + uri: Uri::from_static("https://google.com"), + addr: None, + }; + let conn = pool.call(req.clone()).await.unwrap(); + assert_eq!(2, generated_clone.get()); + release(conn); + + let clone1 = pool.clone(); + let clone2 = clone1.clone(); + + drop(clone2); + for _ in 0..2 { + actix_rt::task::yield_now().await; + } + assert_eq!(2, generated_clone.get()); + + drop(clone1); + for _ in 0..2 { + actix_rt::task::yield_now().await; + } + assert_eq!(2, generated_clone.get()); + + drop(pool); + for _ in 0..2 { + actix_rt::task::yield_now().await; + } + assert_eq!(0, generated_clone.get()); } } diff --git a/actix-http/src/encoding/encoder.rs b/actix-http/src/encoding/encoder.rs index 366ecb8c4..ee0587fbd 100644 --- a/actix-http/src/encoding/encoder.rs +++ b/actix-http/src/encoding/encoder.rs @@ -78,6 +78,7 @@ impl Encoder { }); } } + ResponseBody::Body(Encoder { body, eof: false, diff --git a/actix-http/src/error.rs b/actix-http/src/error.rs index 5eb3c157a..97f2b3eff 100644 --- a/actix-http/src/error.rs +++ b/actix-http/src/error.rs @@ -11,7 +11,6 @@ use actix_utils::dispatcher::DispatcherError as FramedDispatcherError; use actix_utils::timeout::TimeoutError; use bytes::BytesMut; use derive_more::{Display, From}; -pub use futures_channel::oneshot::Canceled; use http::uri::InvalidUri; use http::{header, Error as HttpError, StatusCode}; use serde::de::value::Error as DeError; @@ -19,10 +18,12 @@ use serde_json::error::Error as JsonError; use serde_urlencoded::ser::Error as FormError; use crate::body::Body; -pub use crate::cookie::ParseError as CookieParseError; use crate::helpers::Writer; use crate::response::{Response, ResponseBuilder}; +#[cfg(feature = "cookies")] +pub use crate::cookie::ParseError as CookieParseError; + /// A specialized [`std::result::Result`] /// for actix web operations /// @@ -184,9 +185,6 @@ impl ResponseError for DeError { } } -/// Returns [`StatusCode::INTERNAL_SERVER_ERROR`] for [`Canceled`]. -impl ResponseError for Canceled {} - /// Returns [`StatusCode::BAD_REQUEST`] for [`Utf8Error`]. impl ResponseError for Utf8Error { fn status_code(&self) -> StatusCode { @@ -397,6 +395,7 @@ impl ResponseError for PayloadError { } /// Return `BadRequest` for `cookie::ParseError` +#[cfg(feature = "cookies")] impl ResponseError for crate::cookie::ParseError { fn status_code(&self) -> StatusCode { StatusCode::BAD_REQUEST diff --git a/actix-http/src/h1/dispatcher.rs b/actix-http/src/h1/dispatcher.rs index 2e729b78d..f7d7f32c3 100644 --- a/actix-http/src/h1/dispatcher.rs +++ b/actix-http/src/h1/dispatcher.rs @@ -40,7 +40,6 @@ bitflags! { const SHUTDOWN = 0b0000_0100; const READ_DISCONNECT = 0b0000_1000; const WRITE_DISCONNECT = 0b0001_0000; - const UPGRADE = 0b0010_0000; } } @@ -215,10 +214,7 @@ where U::Error: fmt::Display, { fn can_read(&self, cx: &mut Context<'_>) -> bool { - if self - .flags - .intersects(Flags::READ_DISCONNECT | Flags::UPGRADE) - { + if self.flags.contains(Flags::READ_DISCONNECT) { false } else if let Some(ref info) = self.payload { info.need_read(cx) == PayloadStatus::Read @@ -501,7 +497,7 @@ where } /// Process one incoming request. - pub(self) fn poll_request( + fn poll_request( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Result { @@ -728,6 +724,53 @@ where let mut read_some = false; loop { + // Return early when read buf exceed decoder's max buffer size. + if this.read_buf.len() >= super::decoder::MAX_BUFFER_SIZE { + /* + At this point it's not known IO stream is still scheduled + to be waked up. so force wake up dispatcher just in case. + + Reason: + AsyncRead mostly would only have guarantee wake up + when the poll_read return Poll::Pending. + + Case: + When read_buf is beyond max buffer size the early return + could be successfully be parsed as a new Request. + This case would not generate ParseError::TooLarge + and at this point IO stream is not fully read to Pending + and would result in dispatcher stuck until timeout (KA) + + Note: + This is a perf choice to reduce branch on + ::decode. + + A Request head too large to parse is only checked on + httparse::Status::Partial condition. + */ + if this.payload.is_none() { + /* + When dispatcher has a payload the responsibility of + wake up it would be shift to h1::payload::Payload. + + Reason: + Self wake up when there is payload would waste poll + and/or result in over read. + + Case: + When payload is (partial) dropped by user there is + no need to do read anymore. + At this case read_buf could always remain beyond + MAX_BUFFER_SIZE and self wake up would be busy poll + dispatcher and waste resource. + + */ + cx.waker().wake_by_ref(); + } + + return Ok(false); + } + // grow buffer if necessary. let remaining = this.read_buf.capacity() - this.read_buf.len(); if remaining < LW_BUFFER_SIZE { @@ -735,35 +778,18 @@ where } match actix_codec::poll_read_buf(io.as_mut(), cx, this.read_buf) { - Poll::Pending => return Ok(false), Poll::Ready(Ok(n)) => { if n == 0 { return Ok(true); - } else { - // Return early when read buf exceed decoder's max buffer size. - if this.read_buf.len() >= super::decoder::MAX_BUFFER_SIZE { - // at this point it's not known io is still scheduled to - // be waked up. so force wake up dispatcher just in case. - // TODO: figure out the overhead. - if this.payload.is_none() { - // When dispatcher has a payload. The responsibility of - // wake up stream would be shift to PayloadSender. - // Therefore no self wake up is needed. - cx.waker().wake_by_ref(); - } - return Ok(false); - } - - read_some = true; } + read_some = true; } + Poll::Pending => return Ok(false), Poll::Ready(Err(err)) => { - return if err.kind() == io::ErrorKind::WouldBlock { - Ok(false) - } else if err.kind() == io::ErrorKind::ConnectionReset && read_some { - Ok(true) - } else { - Err(DispatchError::Io(err)) + return match err.kind() { + io::ErrorKind::WouldBlock => Ok(false), + io::ErrorKind::ConnectionReset if read_some => Ok(true), + _ => Err(DispatchError::Io(err)), } } } @@ -985,7 +1011,7 @@ mod tests { None, ); - futures_util::pin_mut!(h1); + actix_rt::pin!(h1); match h1.as_mut().poll(cx) { Poll::Pending => panic!(), @@ -1025,7 +1051,7 @@ mod tests { None, ); - futures_util::pin_mut!(h1); + actix_rt::pin!(h1); assert!(matches!(&h1.inner, DispatcherState::Normal(_))); @@ -1079,7 +1105,7 @@ mod tests { None, ); - futures_util::pin_mut!(h1); + actix_rt::pin!(h1); assert!(matches!(&h1.inner, DispatcherState::Normal(_))); @@ -1138,7 +1164,7 @@ mod tests { ", ); - futures_util::pin_mut!(h1); + actix_rt::pin!(h1); assert!(h1.as_mut().poll(cx).is_pending()); assert!(matches!(&h1.inner, DispatcherState::Normal(_))); @@ -1210,7 +1236,7 @@ mod tests { ", ); - futures_util::pin_mut!(h1); + actix_rt::pin!(h1); assert!(h1.as_mut().poll(cx).is_ready()); assert!(matches!(&h1.inner, DispatcherState::Normal(_))); @@ -1271,7 +1297,7 @@ mod tests { ", ); - futures_util::pin_mut!(h1); + actix_rt::pin!(h1); assert!(h1.as_mut().poll(cx).is_ready()); assert!(matches!(&h1.inner, DispatcherState::Upgrade(_))); diff --git a/actix-http/src/h1/encoder.rs b/actix-http/src/h1/encoder.rs index 69e69de42..97916e7db 100644 --- a/actix-http/src/h1/encoder.rs +++ b/actix-http/src/h1/encoder.rs @@ -549,7 +549,6 @@ mod tests { ); let data = String::from_utf8(Vec::from(bytes.split().freeze().as_ref())).unwrap(); - eprintln!("{}", &data); assert!(data.contains("Content-Length: 0\r\n")); assert!(data.contains("Connection: close\r\n")); diff --git a/actix-http/src/http_message.rs b/actix-http/src/http_message.rs index 2610b8784..b1f04e50d 100644 --- a/actix-http/src/http_message.rs +++ b/actix-http/src/http_message.rs @@ -5,12 +5,14 @@ use encoding_rs::{Encoding, UTF_8}; use http::header; use mime::Mime; -use crate::cookie::Cookie; -use crate::error::{ContentTypeError, CookieParseError, ParseError}; +use crate::error::{ContentTypeError, ParseError}; use crate::extensions::Extensions; use crate::header::{Header, HeaderMap}; use crate::payload::Payload; +#[cfg(feature = "cookies")] +use crate::{cookie::Cookie, error::CookieParseError}; +#[cfg(feature = "cookies")] struct Cookies(Vec>); /// Trait that implements general purpose operations on HTTP messages. @@ -104,7 +106,7 @@ pub trait HttpMessage: Sized { } /// Load request cookies. - #[inline] + #[cfg(feature = "cookies")] fn cookies(&self) -> Result>>, CookieParseError> { if self.extensions().get::().is_none() { let mut cookies = Vec::new(); @@ -119,12 +121,14 @@ pub trait HttpMessage: Sized { } self.extensions_mut().insert(Cookies(cookies)); } + Ok(Ref::map(self.extensions(), |ext| { &ext.get::().unwrap().0 })) } /// Return request cookie. + #[cfg(feature = "cookies")] fn cookie(&self, name: &str) -> Option> { if let Ok(cookies) = self.cookies() { for cookie in cookies.iter() { diff --git a/actix-http/src/lib.rs b/actix-http/src/lib.rs index a55aaadbf..574d4ef68 100644 --- a/actix-http/src/lib.rs +++ b/actix-http/src/lib.rs @@ -1,4 +1,19 @@ //! HTTP primitives for the Actix ecosystem. +//! +//! ## Crate Features +//! | Feature | Functionality | +//! | ---------------- | ----------------------------------------------------- | +//! | `openssl` | TLS support via [OpenSSL]. | +//! | `rustls` | TLS support via [rustls]. | +//! | `compress` | Payload compression support. (Deflate, Gzip & Brotli) | +//! | `cookies` | Support for cookies backed by the [cookie] crate. | +//! | `secure-cookies` | Adds for secure cookies. Enables `cookies` feature. | +//! | `trust-dns` | Use [trust-dns] as the client DNS resolver. | +//! +//! [OpenSSL]: https://crates.io/crates/openssl +//! [rustls]: https://crates.io/crates/rustls +//! [cookie]: https://crates.io/crates/cookie +//! [trust-dns]: https://crates.io/crates/trust-dns #![deny(rust_2018_idioms, nonstandard_style)] #![allow( @@ -34,13 +49,15 @@ mod response; mod service; mod time_parser; -pub use cookie; pub mod error; pub mod h1; pub mod h2; pub mod test; pub mod ws; +#[cfg(feature = "cookies")] +pub use cookie; + pub use self::builder::HttpServiceBuilder; pub use self::config::{KeepAlive, ServiceConfig}; pub use self::error::{Error, ResponseError, Result}; @@ -61,6 +78,7 @@ pub mod http { pub use http::{uri, Error, Uri}; pub use http::{Method, StatusCode, Version}; + #[cfg(feature = "cookies")] pub use crate::cookie::{Cookie, CookieBuilder}; pub use crate::header::HeaderMap; diff --git a/actix-http/src/response.rs b/actix-http/src/response.rs index 763243a63..471dacd28 100644 --- a/actix-http/src/response.rs +++ b/actix-http/src/response.rs @@ -16,13 +16,17 @@ use futures_core::Stream; use serde::Serialize; use crate::body::{Body, BodyStream, MessageBody, ResponseBody}; -use crate::cookie::{Cookie, CookieJar}; use crate::error::Error; use crate::extensions::Extensions; use crate::header::{IntoHeaderPair, IntoHeaderValue}; -use crate::http::header::{self, HeaderName, HeaderValue}; +use crate::http::header::{self, HeaderName}; use crate::http::{Error as HttpError, HeaderMap, StatusCode}; use crate::message::{BoxedResponseHead, ConnectionType, ResponseHead}; +#[cfg(feature = "cookies")] +use crate::{ + cookie::{Cookie, CookieJar}, + http::header::HeaderValue, +}; /// An HTTP Response pub struct Response { @@ -133,6 +137,7 @@ impl Response { } /// Get an iterator for the cookies set by this response + #[cfg(feature = "cookies")] #[inline] pub fn cookies(&self) -> CookieIter<'_> { CookieIter { @@ -141,6 +146,7 @@ impl Response { } /// Add a cookie to this response + #[cfg(feature = "cookies")] #[inline] pub fn add_cookie(&mut self, cookie: &Cookie<'_>) -> Result<(), HttpError> { let h = &mut self.head.headers; @@ -153,6 +159,7 @@ impl Response { /// Remove all cookies with the given name from this response. Returns /// the number of cookies removed. + #[cfg(feature = "cookies")] #[inline] pub fn del_cookie(&mut self, name: &str) -> usize { let h = &mut self.head.headers; @@ -298,10 +305,12 @@ impl Future for Response { } } +#[cfg(feature = "cookies")] pub struct CookieIter<'a> { iter: header::GetAll<'a>, } +#[cfg(feature = "cookies")] impl<'a> Iterator for CookieIter<'a> { type Item = Cookie<'a>; @@ -316,13 +325,13 @@ impl<'a> Iterator for CookieIter<'a> { } } -/// An HTTP response builder +/// An HTTP response builder. /// -/// This type can be used to construct an instance of `Response` through a -/// builder-like pattern. +/// This type can be used to construct an instance of `Response` through a builder-like pattern. pub struct ResponseBuilder { head: Option, err: Option, + #[cfg(feature = "cookies")] cookies: Option, } @@ -333,6 +342,7 @@ impl ResponseBuilder { ResponseBuilder { head: Some(BoxedResponseHead::new(status)), err: None, + #[cfg(feature = "cookies")] cookies: None, } } @@ -531,6 +541,7 @@ impl ResponseBuilder { /// .finish() /// } /// ``` + #[cfg(feature = "cookies")] pub fn cookie<'c>(&mut self, cookie: Cookie<'c>) -> &mut Self { if self.cookies.is_none() { let mut jar = CookieJar::new(); @@ -557,6 +568,7 @@ impl ResponseBuilder { /// builder.finish() /// } /// ``` + #[cfg(feature = "cookies")] pub fn del_cookie<'a>(&mut self, cookie: &Cookie<'a>) -> &mut Self { if self.cookies.is_none() { self.cookies = Some(CookieJar::new()) @@ -624,8 +636,11 @@ impl ResponseBuilder { return Response::from(Error::from(e)).into_body(); } + // allow unused mut when cookies feature is disabled + #[allow(unused_mut)] let mut response = self.head.take().expect("cannot reuse response builder"); + #[cfg(feature = "cookies")] if let Some(ref jar) = self.cookies { for cookie in jar.delta() { match HeaderValue::from_str(&cookie.to_string()) { @@ -693,6 +708,7 @@ impl ResponseBuilder { ResponseBuilder { head: self.head.take(), err: self.err.take(), + #[cfg(feature = "cookies")] cookies: self.cookies.take(), } } @@ -712,21 +728,28 @@ fn parts<'a>( /// Convert `Response` to a `ResponseBuilder`. Body get dropped. impl From> for ResponseBuilder { fn from(res: Response) -> ResponseBuilder { - // If this response has cookies, load them into a jar - let mut jar: Option = None; - for c in res.cookies() { - if let Some(ref mut j) = jar { - j.add_original(c.into_owned()); - } else { - let mut j = CookieJar::new(); - j.add_original(c.into_owned()); - jar = Some(j); + #[cfg(feature = "cookies")] + let jar = { + // If this response has cookies, load them into a jar + let mut jar: Option = None; + + for c in res.cookies() { + if let Some(ref mut j) = jar { + j.add_original(c.into_owned()); + } else { + let mut j = CookieJar::new(); + j.add_original(c.into_owned()); + jar = Some(j); + } } - } + + jar + }; ResponseBuilder { head: Some(res.head), err: None, + #[cfg(feature = "cookies")] cookies: jar, } } @@ -735,22 +758,6 @@ impl From> for ResponseBuilder { /// Convert `ResponseHead` to a `ResponseBuilder` impl<'a> From<&'a ResponseHead> for ResponseBuilder { fn from(head: &'a ResponseHead) -> ResponseBuilder { - // If this response has cookies, load them into a jar - let mut jar: Option = None; - - let cookies = CookieIter { - iter: head.headers.get_all(header::SET_COOKIE), - }; - for c in cookies { - if let Some(ref mut j) = jar { - j.add_original(c.into_owned()); - } else { - let mut j = CookieJar::new(); - j.add_original(c.into_owned()); - jar = Some(j); - } - } - let mut msg = BoxedResponseHead::new(head.status); msg.version = head.version; msg.reason = head.reason; @@ -761,9 +768,32 @@ impl<'a> From<&'a ResponseHead> for ResponseBuilder { msg.no_chunking(!head.chunked()); + #[cfg(feature = "cookies")] + let jar = { + // If this response has cookies, load them into a jar + let mut jar: Option = None; + + let cookies = CookieIter { + iter: head.headers.get_all(header::SET_COOKIE), + }; + + for c in cookies { + if let Some(ref mut j) = jar { + j.add_original(c.into_owned()); + } else { + let mut j = CookieJar::new(); + j.add_original(c.into_owned()); + jar = Some(j); + } + } + + jar + }; + ResponseBuilder { head: Some(msg), err: None, + #[cfg(feature = "cookies")] cookies: jar, } } diff --git a/actix-http/src/test.rs b/actix-http/src/test.rs index 8f0a7d21a..870a656df 100644 --- a/actix-http/src/test.rs +++ b/actix-http/src/test.rs @@ -11,13 +11,14 @@ use std::{ use actix_codec::{AsyncRead, AsyncWrite, ReadBuf}; use bytes::{Bytes, BytesMut}; -use http::{ - header::{self, HeaderValue}, - Method, Uri, Version, -}; +use http::{Method, Uri, Version}; +#[cfg(feature = "cookies")] use crate::{ cookie::{Cookie, CookieJar}, + header::{self, HeaderValue}, +}; +use crate::{ header::{HeaderMap, IntoHeaderPair}, payload::Payload, Request, @@ -53,6 +54,7 @@ struct Inner { method: Method, uri: Uri, headers: HeaderMap, + #[cfg(feature = "cookies")] cookies: CookieJar, payload: Option, } @@ -64,6 +66,7 @@ impl Default for TestRequest { uri: Uri::from_str("/").unwrap(), version: Version::HTTP_11, headers: HeaderMap::new(), + #[cfg(feature = "cookies")] cookies: CookieJar::new(), payload: None, })) @@ -132,6 +135,7 @@ impl TestRequest { } /// Set cookie for this request. + #[cfg(feature = "cookies")] pub fn cookie<'a>(&mut self, cookie: Cookie<'a>) -> &mut Self { parts(&mut self.0).cookies.add(cookie.into_owned()); self @@ -165,17 +169,20 @@ impl TestRequest { head.version = inner.version; head.headers = inner.headers; - let cookie: String = inner - .cookies - .delta() - // ensure only name=value is written to cookie header - .map(|c| Cookie::new(c.name(), c.value()).encoded().to_string()) - .collect::>() - .join("; "); + #[cfg(feature = "cookies")] + { + let cookie: String = inner + .cookies + .delta() + // ensure only name=value is written to cookie header + .map(|c| Cookie::new(c.name(), c.value()).encoded().to_string()) + .collect::>() + .join("; "); - if !cookie.is_empty() { - head.headers - .insert(header::COOKIE, HeaderValue::from_str(&cookie).unwrap()); + if !cookie.is_empty() { + head.headers + .insert(header::COOKIE, HeaderValue::from_str(&cookie).unwrap()); + } } req diff --git a/actix-http/tests/test_openssl.rs b/actix-http/tests/test_openssl.rs index 188516f83..f44968baa 100644 --- a/actix-http/tests/test_openssl.rs +++ b/actix-http/tests/test_openssl.rs @@ -14,7 +14,11 @@ use actix_service::{fn_service, ServiceFactoryExt}; use bytes::{Bytes, BytesMut}; use futures_util::future::{err, ok, ready}; use futures_util::stream::{once, Stream, StreamExt}; -use openssl::ssl::{AlpnError, SslAcceptor, SslFiletype, SslMethod}; +use openssl::{ + pkey::PKey, + ssl::{SslAcceptor, SslMethod}, + x509::X509, +}; async fn load_body(stream: S) -> Result where @@ -34,29 +38,26 @@ where Ok(body) } -fn ssl_acceptor() -> SslAcceptor { - // load ssl keys +fn tls_config() -> SslAcceptor { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); + let cert = X509::from_pem(cert_file.as_bytes()).unwrap(); + let key = PKey::private_key_from_pem(key_file.as_bytes()).unwrap(); + let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("../tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("../tests/cert.pem") - .unwrap(); + builder.set_certificate(&cert).unwrap(); + builder.set_private_key(&key).unwrap(); + builder.set_alpn_select_callback(|_, protos| { const H2: &[u8] = b"\x02h2"; - const H11: &[u8] = b"\x08http/1.1"; if protos.windows(3).any(|window| window == H2) { Ok(b"h2") - } else if protos.windows(9).any(|window| window == H11) { - Ok(b"http/1.1") } else { - Err(AlpnError::NOACK) + Err(openssl::ssl::AlpnError::NOACK) } }); - builder - .set_alpn_protos(b"\x08http/1.1\x02h2") - .expect("Can not contrust SslAcceptor"); + builder.set_alpn_protos(b"\x02h2").unwrap(); builder.build() } @@ -66,7 +67,7 @@ async fn test_h2() -> io::Result<()> { let srv = test_server(move || { HttpService::build() .h2(|_| ok::<_, Error>(Response::Ok().finish())) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -85,7 +86,7 @@ async fn test_h2_1() -> io::Result<()> { assert_eq!(req.version(), Version::HTTP_2); ok::<_, Error>(Response::Ok().finish()) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -104,7 +105,7 @@ async fn test_h2_body() -> io::Result<()> { let body = load_body(req.take_payload()).await?; Ok::<_, Error>(Response::Ok().body(body)) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -133,7 +134,7 @@ async fn test_h2_content_length() { ]; ok::<_, ()>(Response::new(statuses[indx])) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -195,7 +196,7 @@ async fn test_h2_headers() { } ok::<_, ()>(builder.body(data.clone())) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }).await; @@ -234,7 +235,7 @@ async fn test_h2_body2() { let mut srv = test_server(move || { HttpService::build() .h2(|_| ok::<_, ()>(Response::Ok().body(STR))) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -252,7 +253,7 @@ async fn test_h2_head_empty() { let mut srv = test_server(move || { HttpService::build() .finish(|_| ok::<_, ()>(Response::Ok().body(STR))) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -276,7 +277,7 @@ async fn test_h2_head_binary() { let mut srv = test_server(move || { HttpService::build() .h2(|_| ok::<_, ()>(Response::Ok().body(STR))) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -299,7 +300,7 @@ async fn test_h2_head_binary2() { let srv = test_server(move || { HttpService::build() .h2(|_| ok::<_, ()>(Response::Ok().body(STR))) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -323,7 +324,7 @@ async fn test_h2_body_length() { Response::Ok().body(body::SizedStream::new(STR.len() as u64, body)), ) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -348,7 +349,7 @@ async fn test_h2_body_chunked_explicit() { .streaming(body), ) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -376,7 +377,7 @@ async fn test_h2_response_http_error_handling() { .body(STR), ) })) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -394,7 +395,7 @@ async fn test_h2_service_error() { let mut srv = test_server(move || { HttpService::build() .h2(|_| err::(ErrorBadRequest("error"))) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; @@ -418,7 +419,7 @@ async fn test_h2_on_connect() { assert!(req.extensions().contains::()); ok::<_, ()>(Response::Ok().finish()) }) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; diff --git a/actix-http/tests/test_rustls.rs b/actix-http/tests/test_rustls.rs index 59ffcfeb0..a36400910 100644 --- a/actix-http/tests/test_rustls.rs +++ b/actix-http/tests/test_rustls.rs @@ -31,14 +31,19 @@ where Ok(body) } -fn ssl_acceptor() -> RustlsServerConfig { - // load ssl keys +fn tls_config() -> RustlsServerConfig { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); + let mut config = RustlsServerConfig::new(NoClientAuth::new()); - let cert_file = &mut BufReader::new(File::open("../tests/cert.pem").unwrap()); - let key_file = &mut BufReader::new(File::open("../tests/key.pem").unwrap()); + let cert_file = &mut BufReader::new(cert_file.as_bytes()); + let key_file = &mut BufReader::new(key_file.as_bytes()); + let cert_chain = certs(cert_file).unwrap(); let mut keys = pkcs8_private_keys(key_file).unwrap(); config.set_single_cert(cert_chain, keys.remove(0)).unwrap(); + config } @@ -47,7 +52,7 @@ async fn test_h1() -> io::Result<()> { let srv = test_server(move || { HttpService::build() .h1(|_| future::ok::<_, Error>(Response::Ok().finish())) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -61,7 +66,7 @@ async fn test_h2() -> io::Result<()> { let srv = test_server(move || { HttpService::build() .h2(|_| future::ok::<_, Error>(Response::Ok().finish())) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -79,7 +84,7 @@ async fn test_h1_1() -> io::Result<()> { assert_eq!(req.version(), Version::HTTP_11); future::ok::<_, Error>(Response::Ok().finish()) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -97,7 +102,7 @@ async fn test_h2_1() -> io::Result<()> { assert_eq!(req.version(), Version::HTTP_2); future::ok::<_, Error>(Response::Ok().finish()) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -115,7 +120,7 @@ async fn test_h2_body1() -> io::Result<()> { let body = load_body(req.take_payload()).await?; Ok::<_, Error>(Response::Ok().body(body)) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -143,7 +148,7 @@ async fn test_h2_content_length() { ]; future::ok::<_, ()>(Response::new(statuses[indx])) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -203,7 +208,7 @@ async fn test_h2_headers() { } future::ok::<_, ()>(config.body(data.clone())) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }).await; let response = srv.sget("/").send().await.unwrap(); @@ -241,7 +246,7 @@ async fn test_h2_body2() { let mut srv = test_server(move || { HttpService::build() .h2(|_| future::ok::<_, ()>(Response::Ok().body(STR))) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -258,7 +263,7 @@ async fn test_h2_head_empty() { let mut srv = test_server(move || { HttpService::build() .finish(|_| ok::<_, ()>(Response::Ok().body(STR))) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -284,7 +289,7 @@ async fn test_h2_head_binary() { let mut srv = test_server(move || { HttpService::build() .h2(|_| ok::<_, ()>(Response::Ok().body(STR))) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -309,7 +314,7 @@ async fn test_h2_head_binary2() { let srv = test_server(move || { HttpService::build() .h2(|_| ok::<_, ()>(Response::Ok().body(STR))) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -335,7 +340,7 @@ async fn test_h2_body_length() { Response::Ok().body(body::SizedStream::new(STR.len() as u64, body)), ) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -359,7 +364,7 @@ async fn test_h2_body_chunked_explicit() { .streaming(body), ) }) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -388,7 +393,7 @@ async fn test_h2_response_http_error_handling() { ) })) })) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -405,7 +410,7 @@ async fn test_h2_service_error() { let mut srv = test_server(move || { HttpService::build() .h2(|_| err::(error::ErrorBadRequest("error"))) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; @@ -422,7 +427,7 @@ async fn test_h1_service_error() { let mut srv = test_server(move || { HttpService::build() .h1(|_| err::(error::ErrorBadRequest("error"))) - .rustls(ssl_acceptor()) + .rustls(tls_config()) }) .await; diff --git a/awc/CHANGES.md b/awc/CHANGES.md index 20e8af6df..c67f65560 100644 --- a/awc/CHANGES.md +++ b/awc/CHANGES.md @@ -1,6 +1,14 @@ # Changes ## Unreleased - 2021-xx-xx +### Added +* `ClientResponse::timeout` for set the timeout of collecting response body. [#1931] + +### Changed +* Feature `cookies` is now optional and enabled by default. [#1981] + +[#1931]: https://github.com/actix/actix-web/pull/1931 +[#1981]: https://github.com/actix/actix-web/pull/1981 ## 3.0.0-beta.2 - 2021-02-10 diff --git a/awc/Cargo.toml b/awc/Cargo.toml index 615a76e09..9beecc6d4 100644 --- a/awc/Cargo.toml +++ b/awc/Cargo.toml @@ -22,10 +22,11 @@ name = "awc" path = "src/lib.rs" [package.metadata.docs.rs] -features = ["openssl", "rustls", "compress"] +# features that docs.rs will build with +features = ["openssl", "rustls", "compress", "cookies"] [features] -default = ["compress"] +default = ["compress", "cookies"] # openssl openssl = ["tls-openssl", "actix-http/openssl"] @@ -36,6 +37,9 @@ rustls = ["tls-rustls", "actix-http/rustls"] # content-encoding support compress = ["actix-http/compress"] +# cookie parsing and cookie jar +cookies = ["actix-http/cookies"] + # trust-dns as dns resolver trust-dns = ["actix-http/trust-dns"] diff --git a/awc/src/builder.rs b/awc/src/builder.rs index 94ffb8a71..3d1613c66 100644 --- a/awc/src/builder.rs +++ b/awc/src/builder.rs @@ -52,7 +52,6 @@ impl ClientBuilder { where T: Service + 'static, T::Response: Connection, - ::Future: 'static, T::Future: 'static, { self.connector = Some(Box::new(ConnectorWrapper(connector))); diff --git a/awc/src/connect.rs b/awc/src/connect.rs index a9b8f9f83..9a2ded195 100644 --- a/awc/src/connect.rs +++ b/awc/src/connect.rs @@ -41,8 +41,6 @@ where T: Service, T::Response: Connection, ::Io: 'static, - ::Future: 'static, - ::TunnelFuture: 'static, T::Future: 'static, { fn send_request( diff --git a/awc/src/error.rs b/awc/src/error.rs index c60339f76..f86224e62 100644 --- a/awc/src/error.rs +++ b/awc/src/error.rs @@ -1,4 +1,5 @@ -//! Http client errors +//! HTTP client errors + pub use actix_http::client::{ConnectError, FreezeRequestError, InvalidUrl, SendRequestError}; pub use actix_http::error::PayloadError; pub use actix_http::http::Error as HttpError; diff --git a/awc/src/lib.rs b/awc/src/lib.rs index bd52f7ab1..e50c19c8c 100644 --- a/awc/src/lib.rs +++ b/awc/src/lib.rs @@ -97,7 +97,9 @@ use std::convert::TryFrom; use std::rc::Rc; use std::time::Duration; -pub use actix_http::{client::Connector, cookie, http}; +#[cfg(feature = "cookies")] +pub use actix_http::cookie; +pub use actix_http::{client::Connector, http}; use actix_http::http::{Error as HttpError, HeaderMap, Method, Uri}; use actix_http::RequestHead; diff --git a/awc/src/request.rs b/awc/src/request.rs index db8196c5b..812c0e805 100644 --- a/awc/src/request.rs +++ b/awc/src/request.rs @@ -8,6 +8,7 @@ use futures_core::Stream; use serde::Serialize; use actix_http::body::Body; +#[cfg(feature = "cookies")] use actix_http::cookie::{Cookie, CookieJar}; use actix_http::http::header::{self, IntoHeaderPair}; use actix_http::http::{ @@ -54,10 +55,12 @@ pub struct ClientRequest { pub(crate) head: RequestHead, err: Option, addr: Option, - cookies: Option, response_decompress: bool, timeout: Option, config: Rc, + + #[cfg(feature = "cookies")] + cookies: Option, } impl ClientRequest { @@ -72,6 +75,7 @@ impl ClientRequest { head: RequestHead::default(), err: None, addr: None, + #[cfg(feature = "cookies")] cookies: None, timeout: None, response_decompress: true, @@ -290,6 +294,7 @@ impl ClientRequest { /// println!("Response: {:?}", resp); /// } /// ``` + #[cfg(feature = "cookies")] pub fn cookie(mut self, cookie: Cookie<'_>) -> Self { if self.cookies.is_none() { let mut jar = CookieJar::new(); @@ -472,7 +477,8 @@ impl ClientRequest { ) } - fn prep_for_sending(mut self) -> Result { + // allow unused mut when cookies feature is disabled + fn prep_for_sending(#[allow(unused_mut)] mut self) -> Result { if let Some(e) = self.err { return Err(e.into()); } @@ -493,6 +499,7 @@ impl ClientRequest { } // set cookies + #[cfg(feature = "cookies")] if let Some(ref mut jar) = self.cookies { let cookie: String = jar .delta() diff --git a/awc/src/response.rs b/awc/src/response.rs index bbb47ceba..514b8a90b 100644 --- a/awc/src/response.rs +++ b/awc/src/response.rs @@ -1,23 +1,27 @@ -use std::cell::{Ref, RefMut}; -use std::fmt; -use std::future::Future; -use std::io; -use std::marker::PhantomData; -use std::pin::Pin; -use std::task::{Context, Poll}; -use std::time::{Duration, Instant}; +use std::{ + cell::{Ref, RefMut}, + fmt, + future::Future, + io, + marker::PhantomData, + pin::Pin, + task::{Context, Poll}, + time::{Duration, Instant}, +}; +use actix_http::{ + error::PayloadError, + http::{header, HeaderMap, StatusCode, Version}, + Extensions, HttpMessage, Payload, PayloadStream, ResponseHead, +}; +use actix_rt::time::{sleep, Sleep}; use bytes::{Bytes, BytesMut}; use futures_core::{ready, Stream}; - -use actix_http::cookie::Cookie; -use actix_http::error::{CookieParseError, PayloadError}; -use actix_http::http::header::{CONTENT_LENGTH, SET_COOKIE}; -use actix_http::http::{HeaderMap, StatusCode, Version}; -use actix_http::{Extensions, HttpMessage, Payload, PayloadStream, ResponseHead}; -use actix_rt::time::{sleep, Sleep}; use serde::de::DeserializeOwned; +#[cfg(feature = "cookies")] +use actix_http::{cookie::Cookie, error::CookieParseError}; + use crate::error::JsonPayloadError; /// Client Response @@ -78,13 +82,13 @@ impl HttpMessage for ClientResponse { } /// Load request cookies. - #[inline] + #[cfg(feature = "cookies")] fn cookies(&self) -> Result>>, CookieParseError> { struct Cookies(Vec>); if self.extensions().get::().is_none() { let mut cookies = Vec::new(); - for hdr in self.headers().get_all(&SET_COOKIE) { + for hdr in self.headers().get_all(&header::SET_COOKIE) { let s = std::str::from_utf8(hdr.as_bytes()).map_err(CookieParseError::from)?; cookies.push(Cookie::parse_encoded(s)?.into_owned()); } @@ -152,15 +156,13 @@ impl ClientResponse { pub fn timeout(self, dur: Duration) -> Self { let timeout = match self.timeout { ResponseTimeout::Disabled(Some(mut timeout)) - | ResponseTimeout::Enabled(mut timeout) => { - match Instant::now().checked_add(dur) { - Some(deadline) => { - timeout.as_mut().reset(deadline.into()); - ResponseTimeout::Enabled(timeout) - } - None => ResponseTimeout::Enabled(Box::pin(sleep(dur))), + | ResponseTimeout::Enabled(mut timeout) => match Instant::now().checked_add(dur) { + Some(deadline) => { + timeout.as_mut().reset(deadline.into()); + ResponseTimeout::Enabled(timeout) } - } + None => ResponseTimeout::Enabled(Box::pin(sleep(dur))), + }, _ => ResponseTimeout::Enabled(Box::pin(sleep(dur))), }; @@ -207,10 +209,7 @@ where { type Item = Result; - fn poll_next( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - ) -> Poll> { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let this = self.get_mut(); this.timeout.poll_timeout(cx)?; @@ -244,7 +243,7 @@ where /// Create `MessageBody` for request. pub fn new(res: &mut ClientResponse) -> MessageBody { let mut len = None; - if let Some(l) = res.headers().get(&CONTENT_LENGTH) { + if let Some(l) = res.headers().get(&header::CONTENT_LENGTH) { if let Ok(s) = l.to_str() { if let Ok(l) = s.parse::() { len = Some(l) @@ -345,7 +344,8 @@ where } let mut len = None; - if let Some(l) = res.headers().get(&CONTENT_LENGTH) { + + if let Some(l) = res.headers().get(&header::CONTENT_LENGTH) { if let Ok(s) = l.to_str() { if let Ok(l) = s.parse::() { len = Some(l) diff --git a/awc/src/sender.rs b/awc/src/sender.rs index 039f245f6..6bac401c5 100644 --- a/awc/src/sender.rs +++ b/awc/src/sender.rs @@ -22,9 +22,7 @@ use futures_core::{ready, Stream}; use serde::Serialize; #[cfg(feature = "compress")] -use actix_http::{ - encoding::Decoder, http::header::ContentEncoding, Payload, PayloadStream, -}; +use actix_http::{encoding::Decoder, http::header::ContentEncoding, Payload, PayloadStream}; use crate::error::{FreezeRequestError, InvalidUrl, SendRequestError}; use crate::response::ClientResponse; diff --git a/awc/src/test.rs b/awc/src/test.rs index 84646b9f7..97bbb9c3d 100644 --- a/awc/src/test.rs +++ b/awc/src/test.rs @@ -1,9 +1,13 @@ //! Test helpers for actix http client to use during testing. use std::convert::TryFrom; -use actix_http::cookie::{Cookie, CookieJar}; -use actix_http::http::header::{self, Header, HeaderValue, IntoHeaderValue}; +use actix_http::http::header::{Header, IntoHeaderValue}; use actix_http::http::{Error as HttpError, HeaderName, StatusCode, Version}; +#[cfg(feature = "cookies")] +use actix_http::{ + cookie::{Cookie, CookieJar}, + http::header::{self, HeaderValue}, +}; use actix_http::{h1, Payload, ResponseHead}; use bytes::Bytes; @@ -12,6 +16,7 @@ use crate::ClientResponse; /// Test `ClientResponse` builder pub struct TestResponse { head: ResponseHead, + #[cfg(feature = "cookies")] cookies: CookieJar, payload: Option, } @@ -20,6 +25,7 @@ impl Default for TestResponse { fn default() -> TestResponse { TestResponse { head: ResponseHead::new(StatusCode::OK), + #[cfg(feature = "cookies")] cookies: CookieJar::new(), payload: None, } @@ -69,6 +75,7 @@ impl TestResponse { } /// Set cookie for this response + #[cfg(feature = "cookies")] pub fn cookie(mut self, cookie: Cookie<'_>) -> Self { self.cookies.add(cookie.into_owned()); self @@ -84,8 +91,11 @@ impl TestResponse { /// Complete response creation and generate `ClientResponse` instance pub fn finish(self) -> ClientResponse { + // allow unused mut when cookies feature is disabled + #[allow(unused_mut)] let mut head = self.head; + #[cfg(feature = "cookies")] for cookie in self.cookies.delta() { head.headers.insert( header::SET_COOKIE, diff --git a/awc/src/ws.rs b/awc/src/ws.rs index 8db002836..d5528595d 100644 --- a/awc/src/ws.rs +++ b/awc/src/ws.rs @@ -32,6 +32,7 @@ use std::rc::Rc; use std::{fmt, str}; use actix_codec::Framed; +#[cfg(feature = "cookies")] use actix_http::cookie::{Cookie, CookieJar}; use actix_http::{ws, Payload, RequestHead}; use actix_rt::time::timeout; @@ -54,8 +55,10 @@ pub struct WebsocketsRequest { addr: Option, max_size: usize, server_mode: bool, - cookies: Option, config: Rc, + + #[cfg(feature = "cookies")] + cookies: Option, } impl WebsocketsRequest { @@ -89,6 +92,7 @@ impl WebsocketsRequest { protocols: None, max_size: 65_536, server_mode: false, + #[cfg(feature = "cookies")] cookies: None, } } @@ -117,6 +121,7 @@ impl WebsocketsRequest { } /// Set a cookie + #[cfg(feature = "cookies")] pub fn cookie(mut self, cookie: Cookie<'_>) -> Self { if self.cookies.is_none() { let mut jar = CookieJar::new(); @@ -270,6 +275,7 @@ impl WebsocketsRequest { } // set cookies + #[cfg(feature = "cookies")] if let Some(ref mut jar) = self.cookies { let cookie: String = jar .delta() diff --git a/awc/tests/test_connector.rs b/awc/tests/test_connector.rs index 4e4fa5833..fd725506d 100644 --- a/awc/tests/test_connector.rs +++ b/awc/tests/test_connector.rs @@ -7,17 +7,23 @@ use actix_http_test::test_server; use actix_service::{map_config, ServiceFactoryExt}; use actix_web::http::Version; use actix_web::{dev::AppConfig, web, App, HttpResponse}; -use openssl::ssl::{SslAcceptor, SslConnector, SslFiletype, SslMethod, SslVerifyMode}; +use openssl::{ + pkey::PKey, + ssl::{SslAcceptor, SslConnector, SslMethod, SslVerifyMode}, + x509::X509, +}; + +fn tls_config() -> SslAcceptor { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); + let cert = X509::from_pem(cert_file.as_bytes()).unwrap(); + let key = PKey::private_key_from_pem(key_file.as_bytes()).unwrap(); -fn ssl_acceptor() -> SslAcceptor { - // load ssl keys let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("../tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("../tests/cert.pem") - .unwrap(); + builder.set_certificate(&cert).unwrap(); + builder.set_private_key(&key).unwrap(); + builder.set_alpn_select_callback(|_, protos| { const H2: &[u8] = b"\x02h2"; if protos.windows(3).any(|window| window == H2) { @@ -27,6 +33,7 @@ fn ssl_acceptor() -> SslAcceptor { } }); builder.set_alpn_protos(b"\x02h2").unwrap(); + builder.build() } @@ -38,7 +45,7 @@ async fn test_connection_window_size() { App::new().service(web::resource("/").route(web::to(HttpResponse::Ok))), |_| AppConfig::default(), )) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()) }) .await; diff --git a/awc/tests/test_ssl_client.rs b/awc/tests/test_ssl_client.rs index b93c729e5..08aa125cd 100644 --- a/awc/tests/test_ssl_client.rs +++ b/awc/tests/test_ssl_client.rs @@ -11,17 +11,23 @@ use actix_service::{map_config, pipeline_factory, ServiceFactoryExt}; use actix_web::http::Version; use actix_web::{dev::AppConfig, web, App, HttpResponse}; use futures_util::future::ok; -use openssl::ssl::{SslAcceptor, SslConnector, SslFiletype, SslMethod, SslVerifyMode}; +use openssl::{ + pkey::PKey, + ssl::{SslAcceptor, SslConnector, SslMethod, SslVerifyMode}, + x509::X509, +}; + +fn tls_config() -> SslAcceptor { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); + let cert = X509::from_pem(cert_file.as_bytes()).unwrap(); + let key = PKey::private_key_from_pem(key_file.as_bytes()).unwrap(); -fn ssl_acceptor() -> SslAcceptor { - // load ssl keys let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("../tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("../tests/cert.pem") - .unwrap(); + builder.set_certificate(&cert).unwrap(); + builder.set_private_key(&key).unwrap(); + builder.set_alpn_select_callback(|_, protos| { const H2: &[u8] = b"\x02h2"; if protos.windows(3).any(|window| window == H2) { @@ -31,6 +37,7 @@ fn ssl_acceptor() -> SslAcceptor { } }); builder.set_alpn_protos(b"\x02h2").unwrap(); + builder.build() } @@ -51,7 +58,7 @@ async fn test_connection_reuse_h2() { App::new().service(web::resource("/").route(web::to(HttpResponse::Ok))), |_| AppConfig::default(), )) - .openssl(ssl_acceptor()) + .openssl(tls_config()) .map_err(|_| ()), ) }) diff --git a/src/error.rs b/src/error.rs index 1d7c781d8..0865257d3 100644 --- a/src/error.rs +++ b/src/error.rs @@ -13,9 +13,11 @@ pub enum UrlGenerationError { /// Resource not found #[display(fmt = "Resource not found")] ResourceNotFound, + /// Not all path pattern covered #[display(fmt = "Not all path pattern covered")] NotEnoughElements, + /// URL parse error #[display(fmt = "{}", _0)] ParseError(UrlParseError), diff --git a/src/lib.rs b/src/lib.rs index 39cfaf197..16b2ab186 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -61,6 +61,7 @@ //! ## Crate Features //! //! * `compress` - content encoding compression support (enabled by default) +//! * `cookies` - cookies support (enabled by default) //! * `openssl` - HTTPS support via `openssl` crate, supports `HTTP/2` //! * `rustls` - HTTPS support via `rustls` crate, supports `HTTP/2` //! * `secure-cookies` - secure cookies support @@ -98,8 +99,10 @@ pub mod test; pub(crate) mod types; pub mod web; +#[cfg(feature = "cookies")] +pub use actix_http::cookie; pub use actix_http::Response as HttpResponse; -pub use actix_http::{body, cookie, http, Error, HttpMessage, ResponseError, Result}; +pub use actix_http::{body, http, Error, HttpMessage, ResponseError, Result}; pub use actix_rt as rt; pub use actix_web_codegen::*; diff --git a/src/middleware/compat.rs b/src/middleware/compat.rs index 97742e92c..6f60264b1 100644 --- a/src/middleware/compat.rs +++ b/src/middleware/compat.rs @@ -123,18 +123,24 @@ impl MapServiceResponseBody for ServiceRespons #[cfg(test)] mod tests { + // easier to code when cookies feature is disabled + #![allow(unused_imports)] + use super::*; use actix_service::IntoService; use crate::dev::ServiceRequest; use crate::http::StatusCode; - use crate::middleware::{Compress, Condition, Logger}; + use crate::middleware::{self, Condition, Logger}; use crate::test::{call_service, init_service, TestRequest}; use crate::{web, App, HttpResponse}; #[actix_rt::test] + #[cfg(feature = "cookies")] async fn test_scope_middleware() { + use crate::middleware::Compress; + let logger = Logger::default(); let compress = Compress::default(); @@ -154,7 +160,10 @@ mod tests { } #[actix_rt::test] + #[cfg(feature = "cookies")] async fn test_resource_scope_middleware() { + use crate::middleware::Compress; + let logger = Logger::default(); let compress = Compress::default(); diff --git a/src/request.rs b/src/request.rs index d62fca300..514b7466e 100644 --- a/src/request.rs +++ b/src/request.rs @@ -429,12 +429,14 @@ mod tests { } #[test] + #[cfg(feature = "cookies")] fn test_no_request_cookies() { let req = TestRequest::default().to_http_request(); assert!(req.cookies().unwrap().is_empty()); } #[test] + #[cfg(feature = "cookies")] fn test_request_cookies() { let req = TestRequest::default() .append_header((header::COOKIE, "cookie1=value1")) diff --git a/src/test.rs b/src/test.rs index d576838de..2ec4252b1 100644 --- a/src/test.rs +++ b/src/test.rs @@ -6,10 +6,12 @@ use std::sync::mpsc; use std::{fmt, net, thread, time}; use actix_codec::{AsyncRead, AsyncWrite, Framed}; +#[cfg(feature = "cookies")] +use actix_http::cookie::Cookie; use actix_http::http::header::{ContentType, IntoHeaderPair}; use actix_http::http::{Method, StatusCode, Uri, Version}; use actix_http::test::TestRequest as HttpTestRequest; -use actix_http::{cookie::Cookie, ws, Extensions, HttpService, Request}; +use actix_http::{ws, Extensions, HttpService, Request}; use actix_router::{Path, ResourceDef, Url}; use actix_rt::{time::sleep, System}; use actix_service::{map_config, IntoService, IntoServiceFactory, Service, ServiceFactory}; @@ -438,6 +440,7 @@ impl TestRequest { } /// Set cookie for this request. + #[cfg(feature = "cookies")] pub fn cookie(mut self, cookie: Cookie<'_>) -> Self { self.req.cookie(cookie); self diff --git a/tests/cert.pem b/tests/cert.pem deleted file mode 100644 index 0eeb6721d..000000000 --- a/tests/cert.pem +++ /dev/null @@ -1,19 +0,0 @@ ------BEGIN CERTIFICATE----- -MIIDEDCCAfgCCQCQdmIZc/Ib/jANBgkqhkiG9w0BAQsFADBKMQswCQYDVQQGEwJ1 -czELMAkGA1UECAwCY2ExCzAJBgNVBAcMAnNmMSEwHwYJKoZIhvcNAQkBFhJmYWZo -cmQ5MUBnbWFpbC5jb20wHhcNMTkxMTE5MTEwNjU1WhcNMjkxMTE2MTEwNjU1WjBK -MQswCQYDVQQGEwJ1czELMAkGA1UECAwCY2ExCzAJBgNVBAcMAnNmMSEwHwYJKoZI -hvcNAQkBFhJmYWZocmQ5MUBnbWFpbC5jb20wggEiMA0GCSqGSIb3DQEBAQUAA4IB -DwAwggEKAoIBAQDcnaz12CKzUL7248V7Axhms/O9UQXfAdw0yolEfC3P5jADa/1C -+kLWKjAc2coqDSbGsrsR6KiH2g06Kunx+tSGqUO+Sct7HEehmxndiSwx/hfMWezy -XRe/olcHFTeCk/Tllz4xGEplhPua6GLhJygLOhAMiV8cwCYrgyPqsDduExLDFCqc -K2xntIPreumXpiE3QY4+MWyteiJko4IWDFf/UwwsdCY5MlFfw1F/Uv9vz7FfOfvu -GccHd/ex8cOwotUqd6emZb+0bVE24Sv8U+yLnHIVx/tOkxgMAnJEpAnf2G3Wp3zU -b2GJosbmfGaf+xTfnGGhTLLL7kCtva+NvZr5AgMBAAEwDQYJKoZIhvcNAQELBQAD -ggEBANftoL8zDGrjCwWvct8kOOqset2ukK8vjIGwfm88CKsy0IfSochNz2qeIu9R -ZuO7c0pfjmRkir9ZQdq9vXgG3ccL9UstFsferPH9W3YJ83kgXg3fa0EmCiN/0hwz -6Ij1ZBiN1j3+d6+PJPgyYFNu2nGwox5mJ9+aRAGe0/9c63PEOY8P2TI4HsiPmYSl -fFR8k/03vr6e+rTKW85BgctjvYKe/TnFxeCQ7dZ+na7vlEtch4tNmy6O/vEk2kCt -5jW0DUxhmRsv2wGmfFRI0+LotHjoXQQZi6nN5aGL3odaGF3gYwIVlZNd3AdkwDQz -BzG0ZwXuDDV9bSs3MfWEWcy4xuU= ------END CERTIFICATE----- diff --git a/tests/key.pem b/tests/key.pem deleted file mode 100644 index a6d308168..000000000 --- a/tests/key.pem +++ /dev/null @@ -1,28 +0,0 @@ ------BEGIN PRIVATE KEY----- -MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDcnaz12CKzUL72 -48V7Axhms/O9UQXfAdw0yolEfC3P5jADa/1C+kLWKjAc2coqDSbGsrsR6KiH2g06 -Kunx+tSGqUO+Sct7HEehmxndiSwx/hfMWezyXRe/olcHFTeCk/Tllz4xGEplhPua -6GLhJygLOhAMiV8cwCYrgyPqsDduExLDFCqcK2xntIPreumXpiE3QY4+MWyteiJk -o4IWDFf/UwwsdCY5MlFfw1F/Uv9vz7FfOfvuGccHd/ex8cOwotUqd6emZb+0bVE2 -4Sv8U+yLnHIVx/tOkxgMAnJEpAnf2G3Wp3zUb2GJosbmfGaf+xTfnGGhTLLL7kCt -va+NvZr5AgMBAAECggEBAKoU0UwzVgVCQgca8Jt2dnBvWYDhnxIfYAI/BvaKedMm -1ms87OKfB7oOiksjyI0E2JklH72dzZf2jm4CuZt5UjGC+xwPzlTaJ4s6hQVbBHyC -NRyxU1BCXtW5tThbrhD4OjxqjmLRJEIB9OunLtwAEQoeuFLB8Va7+HFhR+Zd9k3f -7aVA93pC5A50NRbZlke4miJ3Q8n7ZF0+UmxkBfm3fbqLk7aMWkoEKwLLTadjRlu1 -bBp0YDStX66I/p1kujqBOdh6VpPvxFOa1sV9pq0jeiGc9YfSkzRSKzIn8GoyviFB -fHeszQdNlcnrSDSNnMABAw+ZpxUO7SCaftjwejEmKZUCgYEA+TY43VpmV95eY7eo -WKwGepiHE0fwQLuKGELmZdZI80tFi73oZMuiB5WzwmkaKGcJmm7KGE9KEvHQCo9j -xvmktBR0VEZH8pmVfun+4h6+0H7m/NKMBBeOyv/IK8jBgHjkkB6e6nmeR7CqTxCw -tf9tbajl1QN8gNzXZSjBDT/lanMCgYEA4qANOKOSiEARtgwyXQeeSJcM2uPv6zF3 -ffM7vjSedtuEOHUSVeyBP/W8KDt7zyPppO/WNbURHS+HV0maS9yyj6zpVS2HGmbs -3fetswsQ+zYVdokW89x4oc2z4XOGHd1LcSlyhRwPt0u2g1E9L0irwTQLWU0npFmG -PRf7sN9+LeMCgYAGkDUDL2ROoB6gRa/7Vdx90hKMoXJkYgwLA4gJ2pDlR3A3c/Lw -5KQJyxmG3zm/IqeQF6be6QesZA30mT4peV2rGHbP2WH/s6fKReNelSy1VQJEWk8x -tGUgV4gwDwN5nLV4TjYlOrq+bJqvpmLhCC8bmj0jVQosYqSRl3cuICasnQKBgGlV -VO/Xb1su1EyWPK5qxRIeSxZOTYw2sMB01nbgxCqge0M2fvA6/hQ5ZlwY0cIEgits -YlcSMsMq/TAAANxz1vbaupUhlSMbZcsBvNV0Nk9c4vr2Wxm7hsJF9u66IEMvQUp2 -pkjiMxfR9CHzF4orr9EcHI5EQ0Grbq5kwFKEfoRbAoGAcWoFPILeJOlp2yW/Ds3E -g2fQdI9BAamtEZEaslJmZMmsDTg5ACPcDkOSFEQIaJ7wLPXeZy74FVk/NrY5F8Gz -bjX9OD/xzwp852yW5L9r62vYJakAlXef5jI6CFdYKDDCcarU0S7W5k6kq9n+wrBR -i1NklYmUAMr2q59uJA5zsic= ------END PRIVATE KEY----- diff --git a/tests/test_httpserver.rs b/tests/test_httpserver.rs index 34d57008d..3aa1d36b0 100644 --- a/tests/test_httpserver.rs +++ b/tests/test_httpserver.rs @@ -73,15 +73,22 @@ async fn test_start() { #[cfg(feature = "openssl")] fn ssl_acceptor() -> std::io::Result { - use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; - // load ssl keys + use openssl::{ + pkey::PKey, + ssl::{SslAcceptor, SslMethod}, + x509::X509, + }; + + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); + let cert = X509::from_pem(cert_file.as_bytes()).unwrap(); + let key = PKey::private_key_from_pem(key_file.as_bytes()).unwrap(); + let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("tests/cert.pem") - .unwrap(); + builder.set_certificate(&cert).unwrap(); + builder.set_private_key(&key).unwrap(); + Ok(builder) } diff --git a/tests/test_server.rs b/tests/test_server.rs index c2caa9eb2..2466730f9 100644 --- a/tests/test_server.rs +++ b/tests/test_server.rs @@ -21,6 +21,11 @@ use flate2::{ Compression, }; use futures_util::ready; +use openssl::{ + pkey::PKey, + ssl::{SslAcceptor, SslMethod}, + x509::X509, +}; use rand::{distributions::Alphanumeric, Rng}; use actix_web::dev::BodyEncoding; @@ -49,6 +54,30 @@ const STR: &str = "Hello World Hello World Hello World Hello World Hello World \ Hello World Hello World Hello World Hello World Hello World \ Hello World Hello World Hello World Hello World Hello World"; +fn openssl_config() -> SslAcceptor { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); + let cert = X509::from_pem(cert_file.as_bytes()).unwrap(); + let key = PKey::private_key_from_pem(key_file.as_bytes()).unwrap(); + + let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); + builder.set_certificate(&cert).unwrap(); + builder.set_private_key(&key).unwrap(); + + builder.set_alpn_select_callback(|_, protos| { + const H2: &[u8] = b"\x02h2"; + if protos.windows(3).any(|window| window == H2) { + Ok(b"h2") + } else { + Err(openssl::ssl::AlpnError::NOACK) + } + }); + builder.set_alpn_protos(b"\x02h2").unwrap(); + + builder.build() +} + struct TestBody { data: Bytes, chunk_size: usize, @@ -700,18 +729,8 @@ async fn test_brotli_encoding_large() { #[cfg(feature = "openssl")] #[actix_rt::test] async fn test_brotli_encoding_large_openssl() { - // load ssl keys - use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; - let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap(); - builder - .set_private_key_file("tests/key.pem", SslFiletype::PEM) - .unwrap(); - builder - .set_certificate_chain_file("tests/cert.pem") - .unwrap(); - let data = STR.repeat(10); - let srv = test::start_with(test::config().openssl(builder.build()), move || { + let srv = test::start_with(test::config().openssl(openssl_config()), move || { App::new().service(web::resource("/").route(web::to(|bytes: Bytes| { HttpResponse::Ok() .encoding(actix_web::http::ContentEncoding::Identity) @@ -739,53 +758,72 @@ async fn test_brotli_encoding_large_openssl() { } #[cfg(all(feature = "rustls", feature = "openssl"))] -#[actix_rt::test] -async fn test_reading_deflate_encoding_large_random_rustls() { - use rustls::internal::pemfile::{certs, pkcs8_private_keys}; - use rustls::{NoClientAuth, ServerConfig}; - use std::fs::File; +mod plus_rustls { use std::io::BufReader; - let data = rand::thread_rng() - .sample_iter(&Alphanumeric) - .take(160_000) - .map(char::from) - .collect::(); + use rustls::{ + internal::pemfile::{certs, pkcs8_private_keys}, + NoClientAuth, ServerConfig as RustlsServerConfig, + }; - // load ssl keys - let mut config = ServerConfig::new(NoClientAuth::new()); - let cert_file = &mut BufReader::new(File::open("tests/cert.pem").unwrap()); - let key_file = &mut BufReader::new(File::open("tests/key.pem").unwrap()); - let cert_chain = certs(cert_file).unwrap(); - let mut keys = pkcs8_private_keys(key_file).unwrap(); - config.set_single_cert(cert_chain, keys.remove(0)).unwrap(); + use super::*; - let srv = test::start_with(test::config().rustls(config), || { - App::new().service(web::resource("/").route(web::to(|bytes: Bytes| { - HttpResponse::Ok() - .encoding(actix_web::http::ContentEncoding::Identity) - .body(bytes) - }))) - }); + fn rustls_config() -> RustlsServerConfig { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap(); + let cert_file = cert.serialize_pem().unwrap(); + let key_file = cert.serialize_private_key_pem(); - // encode data - let mut e = ZlibEncoder::new(Vec::new(), Compression::default()); - e.write_all(data.as_ref()).unwrap(); - let enc = e.finish().unwrap(); + let mut config = RustlsServerConfig::new(NoClientAuth::new()); + let cert_file = &mut BufReader::new(cert_file.as_bytes()); + let key_file = &mut BufReader::new(key_file.as_bytes()); - // client request - let req = srv - .post("/") - .insert_header((actix_web::http::header::CONTENT_ENCODING, "deflate")) - .send_stream(TestBody::new(Bytes::from(enc), 1024)); + let cert_chain = certs(cert_file).unwrap(); + let mut keys = pkcs8_private_keys(key_file).unwrap(); + config.set_single_cert(cert_chain, keys.remove(0)).unwrap(); - let mut response = req.await.unwrap(); - assert!(response.status().is_success()); + config + } - // read response - let bytes = response.body().await.unwrap(); - assert_eq!(bytes.len(), data.len()); - assert_eq!(bytes, Bytes::from(data)); + #[actix_rt::test] + async fn test_reading_deflate_encoding_large_random_rustls() { + use rustls::internal::pemfile::{certs, pkcs8_private_keys}; + use rustls::{NoClientAuth, ServerConfig}; + use std::fs::File; + use std::io::BufReader; + + let data = rand::thread_rng() + .sample_iter(&Alphanumeric) + .take(160_000) + .map(char::from) + .collect::(); + + let srv = test::start_with(test::config().rustls(rustls_config()), || { + App::new().service(web::resource("/").route(web::to(|bytes: Bytes| { + HttpResponse::Ok() + .encoding(actix_web::http::ContentEncoding::Identity) + .body(bytes) + }))) + }); + + // encode data + let mut e = ZlibEncoder::new(Vec::new(), Compression::default()); + e.write_all(data.as_ref()).unwrap(); + let enc = e.finish().unwrap(); + + // client request + let req = srv + .post("/") + .insert_header((actix_web::http::header::CONTENT_ENCODING, "deflate")) + .send_stream(TestBody::new(Bytes::from(enc), 1024)); + + let mut response = req.await.unwrap(); + assert!(response.status().is_success()); + + // read response + let bytes = response.body().await.unwrap(); + assert_eq!(bytes.len(), data.len()); + assert_eq!(bytes, Bytes::from(data)); + } } #[actix_rt::test] @@ -806,15 +844,15 @@ async fn test_server_cookies() { })) }); + let req = srv.get("/"); + let res = req.send().await.unwrap(); + assert!(res.status().is_success()); + let first_cookie = http::CookieBuilder::new("first", "first_value") .http_only(true) .finish(); let second_cookie = http::Cookie::new("second", "second_value"); - let req = srv.get("/"); - let res = req.send().await.unwrap(); - assert!(res.status().is_success()); - let cookies = res.cookies().expect("To have cookies"); assert_eq!(cookies.len(), 2); if cookies[0] == first_cookie {