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

Merge branch 'master' into feat/awc_response_timeout

This commit is contained in:
fakeshadow 2021-02-17 02:10:39 -08:00
parent 9684c9f4e7 2cc6b47fcf
commit daa2c57773
52 changed files with 1946 additions and 1720 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.github/workflows/ci.yml vendored
View File

@ -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'

4
CHANGES.md
View File

@ -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

27
Cargo.toml
View File

@ -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

1
actix-files/Cargo.toml
View File

@ -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"

85
actix-files/src/range.rs
View File

@ -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<Vec<HttpRange>, ParseRangeErr> {
if header.is_empty() {
return Ok(Vec::new());
}
if !header.starts_with(PREFIX) {
return Err(ParseRangeErr(()));
}
let size_sig = size as i64;
let mut no_overlap = false;
let all_ranges: Vec<Option<HttpRange>> = 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,
}))
}
match http_range::HttpRange::parse(header, size) {
Ok(ranges) => Ok(ranges
.iter()
.map(|range| HttpRange {
start: range.start,
length: range.length,
})
.collect::<Result<_, _>>()?;
let ranges: Vec<HttpRange> = all_ranges.into_iter().filter_map(|x| x).collect();
if no_overlap && ranges.is_empty() {
return Err(ParseRangeErr(()));
.collect()),
Err(_) => Err(ParseRangeErr(())),
}
Ok(ranges)
}
}

9
actix-http/CHANGES.md
View File

@ -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

24
actix-http/Cargo.toml
View File

@ -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" }

714
actix-http/src/body.rs
View File

@ -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<Option<Result<Bytes, Error>>>;
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<Option<Result<Bytes, Error>>> {
Poll::Ready(None)
}
}
impl<T: MessageBody + Unpin> MessageBody for Box<T> {
fn size(&self) -> BodySize {
self.as_ref().size()
}
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
Pin::new(self.get_mut().as_mut()).poll_next(cx)
}
}
#[pin_project(project = ResponseBodyProj)]
pub enum ResponseBody<B> {
Body(#[pin] B),
Other(Body),
}
impl ResponseBody<Body> {
pub fn into_body<B>(self) -> ResponseBody<B> {
match self {
ResponseBody::Body(b) => ResponseBody::Other(b),
ResponseBody::Other(b) => ResponseBody::Other(b),
}
}
}
impl<B> ResponseBody<B> {
pub fn take_body(&mut self) -> ResponseBody<B> {
mem::replace(self, ResponseBody::Other(Body::None))
}
}
impl<B: MessageBody> ResponseBody<B> {
pub fn as_ref(&self) -> Option<&B> {
if let ResponseBody::Body(ref b) = self {
Some(b)
} else {
None
}
}
}
impl<B: MessageBody> MessageBody for ResponseBody<B> {
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<Option<Result<Bytes, Error>>> {
match self.project() {
ResponseBodyProj::Body(body) => body.poll_next(cx),
ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx),
}
}
}
impl<B: MessageBody> Stream for ResponseBody<B> {
type Item = Result<Bytes, Error>;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
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<dyn MessageBody + Unpin>),
}
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<B: MessageBody + Unpin + 'static>(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<Option<Result<Bytes, Error>>> {
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<Vec<u8>> for Body {
fn from(vec: Vec<u8>) -> Body {
Body::Bytes(Bytes::from(vec))
}
}
impl From<String> 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<Bytes> for Body {
fn from(s: Bytes) -> Body {
Body::Bytes(s)
}
}
impl From<BytesMut> for Body {
fn from(s: BytesMut) -> Body {
Body::Bytes(s.freeze())
}
}
impl From<serde_json::Value> for Body {
fn from(v: serde_json::Value) -> Body {
Body::Bytes(v.to_string().into())
}
}
impl<S> From<SizedStream<S>> for Body
where
S: Stream<Item = Result<Bytes, Error>> + Unpin + 'static,
{
fn from(s: SizedStream<S>) -> Body {
Body::from_message(s)
}
}
impl<S, E> From<BodyStream<S>> for Body
where
S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
E: Into<Error> + 'static,
{
fn from(s: BodyStream<S>) -> 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<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
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<u8> {
fn size(&self) -> BodySize {
BodySize::Sized(self.len() as u64)
}
fn poll_next(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
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<S: Unpin> {
stream: S,
}
impl<S, E> BodyStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
E: Into<Error>,
{
pub fn new(stream: S) -> Self {
BodyStream { stream }
}
}
impl<S, E> MessageBody for BodyStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
E: Into<Error>,
{
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<Option<Result<Bytes, Error>>> {
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<S: Unpin> {
size: u64,
stream: S,
}
impl<S> SizedStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + Unpin,
{
pub fn new(size: u64, stream: S) -> Self {
SizedStream { size, stream }
}
}
impl<S> MessageBody for SizedStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + 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<Option<Result<Bytes, Error>>> {
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<Body> {
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<Bytes, ()>),
));
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::<String>().unwrap();
assert_eq!(body, "hello cast");
let body = &mut resp_body.downcast_mut::<String>().unwrap();
body.push('!');
let body = resp_body.downcast_ref::<String>().unwrap();
assert_eq!(body, "hello cast!");
let not_body = resp_body.downcast_ref::<()>();
assert!(not_body.is_none());
}
}

158
actix-http/src/body/body.rs Normal file
View File

@ -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<dyn MessageBody + Unpin>),
}
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<B: MessageBody + Unpin + 'static>(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<Option<Result<Bytes, Error>>> {
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<Vec<u8>> for Body {
fn from(vec: Vec<u8>) -> Body {
Body::Bytes(Bytes::from(vec))
}
}
impl From<String> 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<Bytes> for Body {
fn from(s: Bytes) -> Body {
Body::Bytes(s)
}
}
impl From<BytesMut> for Body {
fn from(s: BytesMut) -> Body {
Body::Bytes(s.freeze())
}
}
impl From<serde_json::Value> for Body {
fn from(v: serde_json::Value) -> Body {
Body::Bytes(v.to_string().into())
}
}
impl<S> From<SizedStream<S>> for Body
where
S: Stream<Item = Result<Bytes, Error>> + Unpin + 'static,
{
fn from(s: SizedStream<S>) -> Body {
Body::from_message(s)
}
}
impl<S, E> From<BodyStream<S>> for Body
where
S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
E: Into<Error> + 'static,
{
fn from(s: BodyStream<S>) -> Body {
Body::from_message(s)
}
}

59
actix-http/src/body/body_stream.rs Normal file
View File

@ -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<S: Unpin> {
stream: S,
}
impl<S, E> BodyStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
E: Into<Error>,
{
pub fn new(stream: S) -> Self {
BodyStream { stream }
}
}
impl<S, E> MessageBody for BodyStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
E: Into<Error>,
{
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<Option<Result<Bytes, Error>>> {
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);
}
}
}

142
actix-http/src/body/message_body.rs Normal file
View File

@ -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<Option<Result<Bytes, Error>>>;
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<Option<Result<Bytes, Error>>> {
Poll::Ready(None)
}
}
impl<T: MessageBody + Unpin> MessageBody for Box<T> {
fn size(&self) -> BodySize {
self.as_ref().size()
}
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
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<u8> {
fn size(&self) -> BodySize {
BodySize::Sized(self.len() as u64)
}
fn poll_next(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
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<Option<Result<Bytes, Error>>> {
if self.is_empty() {
Poll::Ready(None)
} else {
Poll::Ready(Some(Ok(Bytes::from(
mem::take(self.get_mut()).into_bytes(),
))))
}
}
}

252
actix-http/src/body/mod.rs Normal file
View File

@ -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<Body> {
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<Bytes, ()>),
));
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::<String>().unwrap();
assert_eq!(body, "hello cast");
let body = &mut resp_body.downcast_mut::<String>().unwrap();
body.push('!');
let body = resp_body.downcast_ref::<String>().unwrap();
assert_eq!(body, "hello cast!");
let not_body = resp_body.downcast_ref::<()>();
assert!(not_body.is_none());
}
}

77
actix-http/src/body/response_body.rs Normal file
View File

@ -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<B> {
Body(#[pin] B),
Other(Body),
}
impl ResponseBody<Body> {
pub fn into_body<B>(self) -> ResponseBody<B> {
match self {
ResponseBody::Body(b) => ResponseBody::Other(b),
ResponseBody::Other(b) => ResponseBody::Other(b),
}
}
}
impl<B> ResponseBody<B> {
pub fn take_body(&mut self) -> ResponseBody<B> {
mem::replace(self, ResponseBody::Other(Body::None))
}
}
impl<B: MessageBody> ResponseBody<B> {
pub fn as_ref(&self) -> Option<&B> {
if let ResponseBody::Body(ref b) = self {
Some(b)
} else {
None
}
}
}
impl<B: MessageBody> MessageBody for ResponseBody<B> {
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<Option<Result<Bytes, Error>>> {
match self.project() {
ResponseBodyProj::Body(body) => body.poll_next(cx),
ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx),
}
}
}
impl<B: MessageBody> Stream for ResponseBody<B> {
type Item = Result<Bytes, Error>;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
match self.project() {
ResponseBodyProj::Body(body) => body.poll_next(cx),
ResponseBodyProj::Other(body) => Pin::new(body).poll_next(cx),
}
}
}

40
actix-http/src/body/size.rs Normal file
View File

@ -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))
}
}

59
actix-http/src/body/sized_stream.rs Normal file
View File

@ -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<S: Unpin> {
size: u64,
stream: S,
}
impl<S> SizedStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + Unpin,
{
pub fn new(size: u64, stream: S) -> Self {
SizedStream { size, stream }
}
}
impl<S> MessageBody for SizedStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + 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<Option<Result<Bytes, Error>>> {
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);
}
}
}

125
actix-http/src/client/connection.rs
View File

@ -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<Io> {
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<Bytes>,
@ -74,7 +73,6 @@ impl DerefMut for H2Connection {
pub trait Connection {
type Io: AsyncRead + AsyncWrite + Unpin;
type Future: Future<Output = Result<(ResponseHead, Payload), SendRequestError>>;
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<Self::Io, ClientCodec>), SendRequestError>,
>;
) -> LocalBoxFuture<'static, Result<(ResponseHead, Payload), SendRequestError>>;
/// Send request, returns Response and Framed
fn open_tunnel<H: Into<RequestHeadType>>(self, head: H) -> Self::TunnelFuture;
fn open_tunnel<H: Into<RequestHeadType> + 'static>(
self,
head: H,
) -> LocalBoxFuture<
'static,
Result<(ResponseHead, Framed<Self::Io, ClientCodec>), 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<T> {
pub struct IoConnection<T>
where
T: AsyncWrite + Unpin + 'static,
{
io: Option<ConnectionType<T>>,
created: time::Instant,
pool: Option<Acquired<T>>,
@ -111,7 +114,7 @@ pub struct IoConnection<T> {
impl<T> fmt::Debug for IoConnection<T>
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<T: AsyncRead + AsyncWrite + Unpin> IoConnection<T> {
pub(crate) fn into_inner(self) -> (ConnectionType<T>, time::Instant) {
(self.io.unwrap(), self.created)
}
#[cfg(test)]
pub(crate) fn into_parts(self) -> (ConnectionType<T>, time::Instant, Acquired<T>) {
(self.io.unwrap(), self.created, self.pool.unwrap())
}
}
impl<T> Connection for IoConnection<T>
@ -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<Self::Io, ClientCodec>), SendRequestError>,
>,
Ready<Result<(ResponseHead, Framed<Self::Io, ClientCodec>), SendRequestError>>,
>;
/// Send request, returns Response and Framed
fn open_tunnel<H: Into<RequestHeadType>>(mut self, head: H) -> Self::TunnelFuture {
fn open_tunnel<H: Into<RequestHeadType>>(
mut self,
head: H,
) -> LocalBoxFuture<
'static,
Result<(ResponseHead, Framed<Self::Io, ClientCodec>), 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<A, B> {
pub(crate) enum EitherConnection<A, B>
where
A: AsyncRead + AsyncWrite + Unpin + 'static,
B: AsyncRead + AsyncWrite + Unpin + 'static,
{
A(IoConnection<A>),
B(IoConnection<B>),
}
@ -213,8 +225,6 @@ where
B: AsyncRead + AsyncWrite + Unpin + 'static,
{
type Io = EitherIo<A, B>;
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<H: Into<RequestHeadType> + 'static>(
self,
head: H,
) -> LocalBoxFuture<
'static,
Result<(ResponseHead, Framed<Self::Io, ClientCodec>), SendRequestError>,
>;
/// Send request, returns Response and Framed
fn open_tunnel<H: Into<RequestHeadType>>(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)))
}),
}
}
}

2
actix-http/src/client/error.rs
View File

@ -25,7 +25,7 @@ pub enum ConnectError {
Resolver(Box<dyn std::error::Error>),
/// No dns records
#[display(fmt = "No dns records found for the input")]
#[display(fmt = "No DNS records found for the input")]
NoRecords,
/// Http2 error

9
actix-http/src/client/h1proto.rs
View File

@ -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<T> {
pub struct H1Connection<T>
where
T: AsyncWrite + Unpin + 'static,
{
/// T should be `Unpin`
io: Option<T>,
created: time::Instant,

4
actix-http/src/client/h2proto.rs
View File

@ -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<B: MessageBody>(
mut send: SendStream<Bytes>,
) -> 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 {

1029
actix-http/src/client/pool.rs
View File

File diff suppressed because it is too large Load Diff

1
actix-http/src/encoding/encoder.rs
View File

@ -78,6 +78,7 @@ impl<B: MessageBody> Encoder<B> {
});
}
}
ResponseBody::Body(Encoder {
body,
eof: false,

9
actix-http/src/error.rs
View File

@ -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

94
actix-http/src/h1/dispatcher.rs
View File

@ -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<bool, DispatchError> {
@ -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
<Request as MessageType>::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;
}
}
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(_)));

1
actix-http/src/h1/encoder.rs
View File

@ -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"));

10
actix-http/src/http_message.rs
View File

@ -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<Cookie<'static>>);
/// 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<Ref<'_, Vec<Cookie<'static>>>, CookieParseError> {
if self.extensions().get::<Cookies>().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::<Cookies>().unwrap().0
}))
}
/// Return request cookie.
#[cfg(feature = "cookies")]
fn cookie(&self, name: &str) -> Option<Cookie<'static>> {
if let Ok(cookies) = self.cookies() {
for cookie in cookies.iter() {

20
actix-http/src/lib.rs
View File

@ -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;

72
actix-http/src/response.rs
View File

@ -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<B = Body> {
@ -133,6 +137,7 @@ impl<B> Response<B> {
}
/// 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<B> Response<B> {
}
/// 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<B> Response<B> {
/// 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<BoxedResponseHead>,
err: Option<HttpError>,
#[cfg(feature = "cookies")]
cookies: Option<CookieJar>,
}
@ -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,8 +728,11 @@ fn parts<'a>(
/// Convert `Response` to a `ResponseBuilder`. Body get dropped.
impl<B> From<Response<B>> for ResponseBuilder {
fn from(res: Response<B>) -> ResponseBuilder {
#[cfg(feature = "cookies")]
let jar = {
// If this response has cookies, load them into a jar
let mut jar: Option<CookieJar> = None;
for c in res.cookies() {
if let Some(ref mut j) = jar {
j.add_original(c.into_owned());
@ -724,9 +743,13 @@ impl<B> From<Response<B>> for ResponseBuilder {
}
}
jar
};
ResponseBuilder {
head: Some(res.head),
err: None,
#[cfg(feature = "cookies")]
cookies: jar,
}
}
@ -735,22 +758,6 @@ impl<B> From<Response<B>> 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<CookieJar> = 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<CookieJar> = 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,
}
}

15
actix-http/src/test.rs
View File

@ -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<Payload>,
}
@ -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,6 +169,8 @@ impl TestRequest {
head.version = inner.version;
head.headers = inner.headers;
#[cfg(feature = "cookies")]
{
let cookie: String = inner
.cookies
.delta()
@ -177,6 +183,7 @@ impl TestRequest {
head.headers
.insert(header::COOKIE, HeaderValue::from_str(&cookie).unwrap());
}
}
req
}

61
actix-http/tests/test_openssl.rs
View File

@ -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<S>(stream: S) -> Result<BytesMut, PayloadError>
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::<Response, Error>(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::<isize>());
ok::<_, ()>(Response::Ok().finish())
})
.openssl(ssl_acceptor())
.openssl(tls_config())
.map_err(|_| ())
})
.await;

45
actix-http/tests/test_rustls.rs
View File

@ -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::<Response, Error>(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::<Response, Error>(error::ErrorBadRequest("error")))
.rustls(ssl_acceptor())
.rustls(tls_config())
})
.await;

8
awc/CHANGES.md
View File

@ -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

8
awc/Cargo.toml
View File

@ -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"]

1
awc/src/builder.rs
View File

@ -52,7 +52,6 @@ impl ClientBuilder {
where
T: Service<HttpConnect, Error = ConnectError> + 'static,
T::Response: Connection,
<T::Response as Connection>::Future: 'static,
T::Future: 'static,
{
self.connector = Some(Box::new(ConnectorWrapper(connector)));

2
awc/src/connect.rs
View File

@ -41,8 +41,6 @@ where
T: Service<ClientConnect, Error = ConnectError>,
T::Response: Connection,
<T::Response as Connection>::Io: 'static,
<T::Response as Connection>::Future: 'static,
<T::Response as Connection>::TunnelFuture: 'static,
T::Future: 'static,
{
fn send_request(

3
awc/src/error.rs
View File

@ -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;

4
awc/src/lib.rs
View File

@ -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;

11
awc/src/request.rs
View File

@ -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<HttpError>,
addr: Option<net::SocketAddr>,
cookies: Option<CookieJar>,
response_decompress: bool,
timeout: Option<Duration>,
config: Rc<ClientConfig>,
#[cfg(feature = "cookies")]
cookies: Option<CookieJar>,
}
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<Self, PrepForSendingError> {
// allow unused mut when cookies feature is disabled
fn prep_for_sending(#[allow(unused_mut)] mut self) -> Result<Self, PrepForSendingError> {
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()

54
awc/src/response.rs
View File

@ -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<S> HttpMessage for ClientResponse<S> {
}
/// Load request cookies.
#[inline]
#[cfg(feature = "cookies")]
fn cookies(&self) -> Result<Ref<'_, Vec<Cookie<'static>>>, CookieParseError> {
struct Cookies(Vec<Cookie<'static>>);
if self.extensions().get::<Cookies>().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<S> ClientResponse<S> {
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) {
| 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<Bytes, PayloadError>;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
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<S>) -> MessageBody<S> {
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::<usize>() {
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::<usize>() {
len = Some(l)

4
awc/src/sender.rs
View File

@ -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;

14
awc/src/test.rs
View File

@ -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<Payload>,
}
@ -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,

8
awc/src/ws.rs
View File

@ -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<SocketAddr>,
max_size: usize,
server_mode: bool,
cookies: Option<CookieJar>,
config: Rc<ClientConfig>,
#[cfg(feature = "cookies")]
cookies: Option<CookieJar>,
}
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()

27
awc/tests/test_connector.rs
View File

@ -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;

27
awc/tests/test_ssl_client.rs
View File

@ -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(|_| ()),
)
})

2
src/error.rs
View File

@ -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),

5
src/lib.rs
View File

@ -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::*;

11
src/middleware/compat.rs
View File

@ -123,18 +123,24 @@ impl<B: MessageBody + Unpin + 'static> 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();

2
src/request.rs
View File

@ -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"))

5
src/test.rs
View File

@ -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

19
tests/cert.pem
View File

@ -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-----

28
tests/key.pem
View File

@ -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-----

23
tests/test_httpserver.rs
View File

@ -73,15 +73,22 @@ async fn test_start() {
#[cfg(feature = "openssl")]
fn ssl_acceptor() -> std::io::Result<SslAcceptorBuilder> {
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)
}

86
tests/test_server.rs
View File

@ -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,6 +758,32 @@ async fn test_brotli_encoding_large_openssl() {
}
#[cfg(all(feature = "rustls", feature = "openssl"))]
mod plus_rustls {
use std::io::BufReader;
use rustls::{
internal::pemfile::{certs, pkcs8_private_keys},
NoClientAuth, ServerConfig as RustlsServerConfig,
};
use super::*;
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();
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());
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
}
#[actix_rt::test]
async fn test_reading_deflate_encoding_large_random_rustls() {
use rustls::internal::pemfile::{certs, pkcs8_private_keys};
@ -752,15 +797,7 @@ async fn test_reading_deflate_encoding_large_random_rustls() {
.map(char::from)
.collect::<String>();
// 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();
let srv = test::start_with(test::config().rustls(config), || {
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)
@ -787,6 +824,7 @@ async fn test_reading_deflate_encoding_large_random_rustls() {
assert_eq!(bytes.len(), data.len());
assert_eq!(bytes, Bytes::from(data));
}
}
#[actix_rt::test]
async fn test_server_cookies() {
@ -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 {