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

Merge branch 'master' into mlodato517-improve-errorhandlers-docs

This commit is contained in:
Rob Ede 2021-12-25 02:35:09 +00:00 committed by GitHub
parent c00590497c 01cbfc5724
commit 14096cf96d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
42 changed files with 1283 additions and 1035 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
.github/workflows/ci-master.yml vendored Normal file
View File

@ -0,0 +1,66 @@
name: CI (master only)
on:
push:
branches: [master]
jobs:
ci_feature_powerset_check:
name: Verify Feature Combinations
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Install stable
uses: actions-rs/toolchain@v1
with:
toolchain: stable-x86_64-unknown-linux-gnu
profile: minimal
override: true
- name: Generate Cargo.lock
uses: actions-rs/cargo@v1
with: { command: generate-lockfile }
- name: Cache Dependencies
uses: Swatinem/rust-cache@v1.2.0
- name: Install cargo-hack
uses: actions-rs/cargo@v1
with:
command: install
args: cargo-hack
- name: check feature combinations
uses: actions-rs/cargo@v1
with: { command: ci-check-all-feature-powerset }
- name: check feature combinations
uses: actions-rs/cargo@v1
with: { command: ci-check-all-feature-powerset-linux }
coverage:
name: coverage
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Install stable
uses: actions-rs/toolchain@v1
with:
toolchain: stable-x86_64-unknown-linux-gnu
profile: minimal
override: true
- name: Generate Cargo.lock
uses: actions-rs/cargo@v1
with: { command: generate-lockfile }
- name: Cache Dependencies
uses: Swatinem/rust-cache@v1.2.0
- name: Generate coverage file
run: |
cargo install cargo-tarpaulin --vers "^0.13"
cargo tarpaulin --workspace --features=rustls,openssl --out Xml --verbose
- name: Upload to Codecov
uses: codecov/codecov-action@v1
with: { file: cobertura.xml }

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

@ -96,68 +96,6 @@ jobs:
cargo install cargo-cache --version 0.6.3 --no-default-features --features ci-autoclean cargo install cargo-cache --version 0.6.3 --no-default-features --features ci-autoclean
cargo-cache cargo-cache
ci_feature_powerset_check:
name: Verify Feature Combinations
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Install stable
uses: actions-rs/toolchain@v1
with:
toolchain: stable-x86_64-unknown-linux-gnu
profile: minimal
override: true
- name: Generate Cargo.lock
uses: actions-rs/cargo@v1
with: { command: generate-lockfile }
- name: Cache Dependencies
uses: Swatinem/rust-cache@v1.2.0
- name: Install cargo-hack
uses: actions-rs/cargo@v1
with:
command: install
args: cargo-hack
- name: check feature combinations
uses: actions-rs/cargo@v1
with: { command: ci-check-all-feature-powerset }
- name: check feature combinations
uses: actions-rs/cargo@v1
with: { command: ci-check-all-feature-powerset-linux }
coverage:
name: coverage
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Install stable
uses: actions-rs/toolchain@v1
with:
toolchain: stable-x86_64-unknown-linux-gnu
profile: minimal
override: true
- name: Generate Cargo.lock
uses: actions-rs/cargo@v1
with: { command: generate-lockfile }
- name: Cache Dependencies
uses: Swatinem/rust-cache@v1.2.0
- name: Generate coverage file
if: github.ref == 'refs/heads/master'
run: |
cargo install cargo-tarpaulin --vers "^0.13"
cargo tarpaulin --workspace --features=rustls,openssl --out Xml --verbose
- name: Upload to Codecov
if: github.ref == 'refs/heads/master'
uses: codecov/codecov-action@v1
with: { file: cobertura.xml }
rustdoc: rustdoc:
name: doc tests name: doc tests
runs-on: ubuntu-latest runs-on: ubuntu-latest

9
actix-http/CHANGES.md
View File

@ -3,8 +3,17 @@
## Unreleased - 2021-xx-xx ## Unreleased - 2021-xx-xx
### Changes ### Changes
- `HeaderMap::get_all` now returns a `std::slice::Iter`. [#2527] - `HeaderMap::get_all` now returns a `std::slice::Iter`. [#2527]
- `Payload` inner fields are now named. [#2545]
- `impl Stream` for `Payload` no longer requires the `Stream` variant be `Unpin`. [#2545]
- `impl Future` for `h1::SendResponse` no longer requires the body type be `Unpin`. [#2545]
- `impl Stream` for `encoding::Decoder` no longer requires the stream type be `Unpin`. [#2545]
- Rename `PayloadStream` to `BoxedPayloadStream`. [#2545]
### Removed
- `h1::Payload::readany`. [#2545]
[#2527]: https://github.com/actix/actix-web/pull/2527 [#2527]: https://github.com/actix/actix-web/pull/2527
[#2545]: https://github.com/actix/actix-web/pull/2545
## 3.0.0-beta.16 - 2021-12-17 ## 3.0.0-beta.16 - 2021-12-17

31
actix-http/src/encoding/decoder.rs
View File

@ -28,11 +28,14 @@ use crate::{
const MAX_CHUNK_SIZE_DECODE_IN_PLACE: usize = 2049; const MAX_CHUNK_SIZE_DECODE_IN_PLACE: usize = 2049;
pub struct Decoder<S> { pin_project_lite::pin_project! {
pub struct Decoder<S> {
decoder: Option<ContentDecoder>, decoder: Option<ContentDecoder>,
#[pin]
stream: S, stream: S,
eof: bool, eof: bool,
fut: Option<JoinHandle<Result<(Option<Bytes>, ContentDecoder), io::Error>>>, fut: Option<JoinHandle<Result<(Option<Bytes>, ContentDecoder), io::Error>>>,
}
} }
impl<S> Decoder<S> impl<S> Decoder<S>
@ -89,42 +92,44 @@ where
impl<S> Stream for Decoder<S> impl<S> Stream for Decoder<S>
where where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin, S: Stream<Item = Result<Bytes, PayloadError>>,
{ {
type Item = Result<Bytes, PayloadError>; type Item = Result<Bytes, PayloadError>;
fn poll_next(mut 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 mut this = self.project();
loop { loop {
if let Some(ref mut fut) = self.fut { if let Some(ref mut fut) = this.fut {
let (chunk, decoder) = let (chunk, decoder) =
ready!(Pin::new(fut).poll(cx)).map_err(|_| BlockingError)??; ready!(Pin::new(fut).poll(cx)).map_err(|_| BlockingError)??;
self.decoder = Some(decoder); *this.decoder = Some(decoder);
self.fut.take(); this.fut.take();
if let Some(chunk) = chunk { if let Some(chunk) = chunk {
return Poll::Ready(Some(Ok(chunk))); return Poll::Ready(Some(Ok(chunk)));
} }
} }
if self.eof { if *this.eof {
return Poll::Ready(None); return Poll::Ready(None);
} }
match ready!(Pin::new(&mut self.stream).poll_next(cx)) { match ready!(this.stream.as_mut().poll_next(cx)) {
Some(Err(err)) => return Poll::Ready(Some(Err(err))), Some(Err(err)) => return Poll::Ready(Some(Err(err))),
Some(Ok(chunk)) => { Some(Ok(chunk)) => {
if let Some(mut decoder) = self.decoder.take() { if let Some(mut decoder) = this.decoder.take() {
if chunk.len() < MAX_CHUNK_SIZE_DECODE_IN_PLACE { if chunk.len() < MAX_CHUNK_SIZE_DECODE_IN_PLACE {
let chunk = decoder.feed_data(chunk)?; let chunk = decoder.feed_data(chunk)?;
self.decoder = Some(decoder); *this.decoder = Some(decoder);
if let Some(chunk) = chunk { if let Some(chunk) = chunk {
return Poll::Ready(Some(Ok(chunk))); return Poll::Ready(Some(Ok(chunk)));
} }
} else { } else {
self.fut = Some(spawn_blocking(move || { *this.fut = Some(spawn_blocking(move || {
let chunk = decoder.feed_data(chunk)?; let chunk = decoder.feed_data(chunk)?;
Ok((chunk, decoder)) Ok((chunk, decoder))
})); }));
@ -137,9 +142,9 @@ where
} }
None => { None => {
self.eof = true; *this.eof = true;
return if let Some(mut decoder) = self.decoder.take() { return if let Some(mut decoder) = this.decoder.take() {
match decoder.feed_eof() { match decoder.feed_eof() {
Ok(Some(res)) => Poll::Ready(Some(Ok(res))), Ok(Some(res)) => Poll::Ready(Some(Ok(res))),
Ok(None) => Poll::Ready(None), Ok(None) => Poll::Ready(None),

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

@ -646,10 +646,11 @@ where
Payload is attached to Request and passed to Service::call Payload is attached to Request and passed to Service::call
where the state can be collected and consumed. where the state can be collected and consumed.
*/ */
let (ps, pl) = Payload::create(false); let (sender, payload) = Payload::create(false);
let (req1, _) = req.replace_payload(crate::Payload::H1(pl)); let (req1, _) =
req.replace_payload(crate::Payload::H1 { payload });
req = req1; req = req1;
*this.payload = Some(ps); *this.payload = Some(sender);
} }
// Request has no payload. // Request has no payload.

76
actix-http/src/h1/payload.rs
View File

@ -1,9 +1,12 @@
//! Payload stream //! Payload stream
use std::cell::RefCell;
use std::collections::VecDeque; use std::{
use std::pin::Pin; cell::RefCell,
use std::rc::{Rc, Weak}; collections::VecDeque,
use std::task::{Context, Poll, Waker}; pin::Pin,
rc::{Rc, Weak},
task::{Context, Poll, Waker},
};
use bytes::Bytes; use bytes::Bytes;
use futures_core::Stream; use futures_core::Stream;
@ -22,39 +25,32 @@ pub enum PayloadStatus {
/// Buffered stream of bytes chunks /// Buffered stream of bytes chunks
/// ///
/// Payload stores chunks in a vector. First chunk can be received with /// Payload stores chunks in a vector. First chunk can be received with `poll_next`. Payload does
/// `.readany()` method. Payload stream is not thread safe. Payload does not /// not notify current task when new data is available.
/// notify current task when new data is available.
/// ///
/// Payload stream can be used as `Response` body stream. /// Payload can be used as `Response` body stream.
#[derive(Debug)] #[derive(Debug)]
pub struct Payload { pub struct Payload {
inner: Rc<RefCell<Inner>>, inner: Rc<RefCell<Inner>>,
} }
impl Payload { impl Payload {
/// Create payload stream. /// Creates a payload stream.
/// ///
/// This method construct two objects responsible for bytes stream /// This method construct two objects responsible for bytes stream generation:
/// generation. /// - `PayloadSender` - *Sender* side of the stream
/// /// - `Payload` - *Receiver* side of the stream
/// * `PayloadSender` - *Sender* side of the stream
///
/// * `Payload` - *Receiver* side of the stream
pub fn create(eof: bool) -> (PayloadSender, Payload) { pub fn create(eof: bool) -> (PayloadSender, Payload) {
let shared = Rc::new(RefCell::new(Inner::new(eof))); let shared = Rc::new(RefCell::new(Inner::new(eof)));
( (
PayloadSender { PayloadSender::new(Rc::downgrade(&shared)),
inner: Rc::downgrade(&shared),
},
Payload { inner: shared }, Payload { inner: shared },
) )
} }
/// Create empty payload /// Creates an empty payload.
#[doc(hidden)] pub(crate) fn empty() -> Payload {
pub fn empty() -> Payload {
Payload { Payload {
inner: Rc::new(RefCell::new(Inner::new(true))), inner: Rc::new(RefCell::new(Inner::new(true))),
} }
@ -77,14 +73,6 @@ impl Payload {
pub fn unread_data(&mut self, data: Bytes) { pub fn unread_data(&mut self, data: Bytes) {
self.inner.borrow_mut().unread_data(data); self.inner.borrow_mut().unread_data(data);
} }
#[inline]
pub fn readany(
&mut self,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, PayloadError>>> {
self.inner.borrow_mut().readany(cx)
}
} }
impl Stream for Payload { impl Stream for Payload {
@ -94,7 +82,7 @@ impl Stream for Payload {
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, PayloadError>>> { ) -> Poll<Option<Result<Bytes, PayloadError>>> {
self.inner.borrow_mut().readany(cx) Pin::new(&mut *self.inner.borrow_mut()).poll_next(cx)
} }
} }
@ -104,6 +92,10 @@ pub struct PayloadSender {
} }
impl PayloadSender { impl PayloadSender {
fn new(inner: Weak<RefCell<Inner>>) -> Self {
Self { inner }
}
#[inline] #[inline]
pub fn set_error(&mut self, err: PayloadError) { pub fn set_error(&mut self, err: PayloadError) {
if let Some(shared) = self.inner.upgrade() { if let Some(shared) = self.inner.upgrade() {
@ -227,7 +219,10 @@ impl Inner {
self.len self.len
} }
fn readany(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, PayloadError>>> { fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, PayloadError>>> {
if let Some(data) = self.items.pop_front() { if let Some(data) = self.items.pop_front() {
self.len -= data.len(); self.len -= data.len();
self.need_read = self.len < MAX_BUFFER_SIZE; self.need_read = self.len < MAX_BUFFER_SIZE;
@ -257,8 +252,18 @@ impl Inner {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use std::panic::{RefUnwindSafe, UnwindSafe};
use actix_utils::future::poll_fn; use actix_utils::future::poll_fn;
use static_assertions::{assert_impl_all, assert_not_impl_any};
use super::*;
assert_impl_all!(Payload: Unpin);
assert_not_impl_any!(Payload: Send, Sync, UnwindSafe, RefUnwindSafe);
assert_impl_all!(Inner: Unpin, Send, Sync);
assert_not_impl_any!(Inner: UnwindSafe, RefUnwindSafe);
#[actix_rt::test] #[actix_rt::test]
async fn test_unread_data() { async fn test_unread_data() {
@ -270,7 +275,10 @@ mod tests {
assert_eq!( assert_eq!(
Bytes::from("data"), Bytes::from("data"),
poll_fn(|cx| payload.readany(cx)).await.unwrap().unwrap() poll_fn(|cx| Pin::new(&mut payload).poll_next(cx))
.await
.unwrap()
.unwrap()
); );
} }
} }

4
actix-http/src/h1/utils.rs
View File

@ -45,7 +45,7 @@ where
impl<T, B> Future for SendResponse<T, B> impl<T, B> Future for SendResponse<T, B>
where where
T: AsyncRead + AsyncWrite + Unpin, T: AsyncRead + AsyncWrite + Unpin,
B: MessageBody + Unpin, B: MessageBody,
B::Error: Into<Error>, B::Error: Into<Error>,
{ {
type Output = Result<Framed<T, Codec>, Error>; type Output = Result<Framed<T, Codec>, Error>;
@ -81,7 +81,7 @@ where
// body is done when item is None // body is done when item is None
body_done = item.is_none(); body_done = item.is_none();
if body_done { if body_done {
let _ = this.body.take(); this.body.set(None);
} }
let framed = this.framed.as_mut().as_pin_mut().unwrap(); let framed = this.framed.as_mut().as_pin_mut().unwrap();
framed framed

4
actix-http/src/h2/dispatcher.rs
View File

@ -108,8 +108,8 @@ where
match Pin::new(&mut this.connection).poll_accept(cx)? { match Pin::new(&mut this.connection).poll_accept(cx)? {
Poll::Ready(Some((req, tx))) => { Poll::Ready(Some((req, tx))) => {
let (parts, body) = req.into_parts(); let (parts, body) = req.into_parts();
let pl = crate::h2::Payload::new(body); let payload = crate::h2::Payload::new(body);
let pl = Payload::H2(pl); let pl = Payload::H2 { payload };
let mut req = Request::with_payload(pl); let mut req = Request::with_payload(pl);
let head = req.head_mut(); let head = req.head_mut();

11
actix-http/src/h2/mod.rs
View File

@ -98,3 +98,14 @@ where
} }
} }
} }
#[cfg(test)]
mod tests {
use std::panic::{RefUnwindSafe, UnwindSafe};
use static_assertions::assert_impl_all;
use super::*;
assert_impl_all!(Payload: Unpin, Send, Sync, UnwindSafe, RefUnwindSafe);
}

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

@ -58,7 +58,8 @@ pub use self::header::ContentEncoding;
pub use self::http_message::HttpMessage; pub use self::http_message::HttpMessage;
pub use self::message::ConnectionType; pub use self::message::ConnectionType;
pub use self::message::Message; pub use self::message::Message;
pub use self::payload::{Payload, PayloadStream}; #[allow(deprecated)]
pub use self::payload::{BoxedPayloadStream, Payload, PayloadStream};
pub use self::requests::{Request, RequestHead, RequestHeadType}; pub use self::requests::{Request, RequestHead, RequestHeadType};
pub use self::responses::{Response, ResponseBuilder, ResponseHead}; pub use self::responses::{Response, ResponseBuilder, ResponseHead};
pub use self::service::HttpService; pub use self::service::HttpService;

71
actix-http/src/payload.rs
View File

@ -1,70 +1,89 @@
use std::{ use std::{
mem,
pin::Pin, pin::Pin,
task::{Context, Poll}, task::{Context, Poll},
}; };
use bytes::Bytes; use bytes::Bytes;
use futures_core::Stream; use futures_core::Stream;
use h2::RecvStream;
use crate::error::PayloadError; use crate::error::PayloadError;
// TODO: rename to boxed payload /// A boxed payload stream.
/// A boxed payload. pub type BoxedPayloadStream = Pin<Box<dyn Stream<Item = Result<Bytes, PayloadError>>>>;
pub type PayloadStream = Pin<Box<dyn Stream<Item = Result<Bytes, PayloadError>>>>;
/// A streaming payload. #[deprecated(since = "4.0.0", note = "Renamed to `BoxedPayloadStream`.")]
pub enum Payload<S = PayloadStream> { pub type PayloadStream = BoxedPayloadStream;
pin_project_lite::pin_project! {
/// A streaming payload.
#[project = PayloadProj]
pub enum Payload<S = BoxedPayloadStream> {
None, None,
H1(crate::h1::Payload), H1 { payload: crate::h1::Payload },
H2(crate::h2::Payload), H2 { payload: crate::h2::Payload },
Stream(S), Stream { #[pin] payload: S },
}
} }
impl<S> From<crate::h1::Payload> for Payload<S> { impl<S> From<crate::h1::Payload> for Payload<S> {
fn from(v: crate::h1::Payload) -> Self { fn from(payload: crate::h1::Payload) -> Self {
Payload::H1(v) Payload::H1 { payload }
} }
} }
impl<S> From<crate::h2::Payload> for Payload<S> { impl<S> From<crate::h2::Payload> for Payload<S> {
fn from(v: crate::h2::Payload) -> Self { fn from(payload: crate::h2::Payload) -> Self {
Payload::H2(v) Payload::H2 { payload }
} }
} }
impl<S> From<RecvStream> for Payload<S> { impl<S> From<h2::RecvStream> for Payload<S> {
fn from(v: RecvStream) -> Self { fn from(stream: h2::RecvStream) -> Self {
Payload::H2(crate::h2::Payload::new(v)) Payload::H2 {
payload: crate::h2::Payload::new(stream),
}
} }
} }
impl From<PayloadStream> for Payload { impl From<BoxedPayloadStream> for Payload {
fn from(pl: PayloadStream) -> Self { fn from(payload: BoxedPayloadStream) -> Self {
Payload::Stream(pl) Payload::Stream { payload }
} }
} }
impl<S> Payload<S> { impl<S> Payload<S> {
/// Takes current payload and replaces it with `None` value /// Takes current payload and replaces it with `None` value
pub fn take(&mut self) -> Payload<S> { pub fn take(&mut self) -> Payload<S> {
std::mem::replace(self, Payload::None) mem::replace(self, Payload::None)
} }
} }
impl<S> Stream for Payload<S> impl<S> Stream for Payload<S>
where where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin, S: Stream<Item = Result<Bytes, PayloadError>>,
{ {
type Item = Result<Bytes, PayloadError>; type Item = Result<Bytes, PayloadError>;
#[inline] #[inline]
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>> {
match self.get_mut() { match self.project() {
Payload::None => Poll::Ready(None), PayloadProj::None => Poll::Ready(None),
Payload::H1(ref mut pl) => pl.readany(cx), PayloadProj::H1 { payload } => Pin::new(payload).poll_next(cx),
Payload::H2(ref mut pl) => Pin::new(pl).poll_next(cx), PayloadProj::H2 { payload } => Pin::new(payload).poll_next(cx),
Payload::Stream(ref mut pl) => Pin::new(pl).poll_next(cx), PayloadProj::Stream { payload } => payload.poll_next(cx),
} }
} }
} }
#[cfg(test)]
mod tests {
use std::panic::{RefUnwindSafe, UnwindSafe};
use static_assertions::{assert_impl_all, assert_not_impl_any};
use super::*;
assert_impl_all!(Payload: Unpin);
assert_not_impl_any!(Payload: Send, Sync, UnwindSafe, RefUnwindSafe);
}

11
actix-http/src/requests/request.rs
View File

@ -10,11 +10,12 @@ use std::{
use http::{header, Method, Uri, Version}; use http::{header, Method, Uri, Version};
use crate::{ use crate::{
header::HeaderMap, Extensions, HttpMessage, Message, Payload, PayloadStream, RequestHead, header::HeaderMap, BoxedPayloadStream, Extensions, HttpMessage, Message, Payload,
RequestHead,
}; };
/// An HTTP request. /// An HTTP request.
pub struct Request<P = PayloadStream> { pub struct Request<P = BoxedPayloadStream> {
pub(crate) payload: Payload<P>, pub(crate) payload: Payload<P>,
pub(crate) head: Message<RequestHead>, pub(crate) head: Message<RequestHead>,
pub(crate) conn_data: Option<Rc<Extensions>>, pub(crate) conn_data: Option<Rc<Extensions>>,
@ -46,7 +47,7 @@ impl<P> HttpMessage for Request<P> {
} }
} }
impl From<Message<RequestHead>> for Request<PayloadStream> { impl From<Message<RequestHead>> for Request<BoxedPayloadStream> {
fn from(head: Message<RequestHead>) -> Self { fn from(head: Message<RequestHead>) -> Self {
Request { Request {
head, head,
@ -57,10 +58,10 @@ impl From<Message<RequestHead>> for Request<PayloadStream> {
} }
} }
impl Request<PayloadStream> { impl Request<BoxedPayloadStream> {
/// Create new Request instance /// Create new Request instance
#[allow(clippy::new_without_default)] #[allow(clippy::new_without_default)]
pub fn new() -> Request<PayloadStream> { pub fn new() -> Request<BoxedPayloadStream> {
Request { Request {
head: Message::new(), head: Message::new(),
payload: Payload::None, payload: Payload::None,

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

@ -120,7 +120,7 @@ impl TestRequest {
} }
/// Set request payload. /// Set request payload.
pub fn set_payload<B: Into<Bytes>>(&mut self, data: B) -> &mut Self { pub fn set_payload(&mut self, data: impl Into<Bytes>) -> &mut Self {
let mut payload = crate::h1::Payload::empty(); let mut payload = crate::h1::Payload::empty();
payload.unread_data(data.into()); payload.unread_data(data.into());
parts(&mut self.0).payload = Some(payload.into()); parts(&mut self.0).payload = Some(payload.into());

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

@ -7,6 +7,7 @@ use std::{
io::{self, BufReader, Write}, io::{self, BufReader, Write},
net::{SocketAddr, TcpStream as StdTcpStream}, net::{SocketAddr, TcpStream as StdTcpStream},
sync::Arc, sync::Arc,
task::Poll,
}; };
use actix_http::{ use actix_http::{
@ -16,25 +17,37 @@ use actix_http::{
Error, HttpService, Method, Request, Response, StatusCode, Version, Error, HttpService, Method, Request, Response, StatusCode, Version,
}; };
use actix_http_test::test_server; use actix_http_test::test_server;
use actix_rt::pin;
use actix_service::{fn_factory_with_config, fn_service}; use actix_service::{fn_factory_with_config, fn_service};
use actix_tls::connect::rustls::webpki_roots_cert_store; use actix_tls::connect::rustls::webpki_roots_cert_store;
use actix_utils::future::{err, ok}; use actix_utils::future::{err, ok, poll_fn};
use bytes::{Bytes, BytesMut}; use bytes::{Bytes, BytesMut};
use derive_more::{Display, Error}; use derive_more::{Display, Error};
use futures_core::Stream; use futures_core::{ready, Stream};
use futures_util::stream::{once, StreamExt as _}; use futures_util::stream::once;
use rustls::{Certificate, PrivateKey, ServerConfig as RustlsServerConfig, ServerName}; use rustls::{Certificate, PrivateKey, ServerConfig as RustlsServerConfig, ServerName};
use rustls_pemfile::{certs, pkcs8_private_keys}; use rustls_pemfile::{certs, pkcs8_private_keys};
async fn load_body<S>(mut stream: S) -> Result<BytesMut, PayloadError> async fn load_body<S>(stream: S) -> Result<BytesMut, PayloadError>
where where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin, S: Stream<Item = Result<Bytes, PayloadError>>,
{ {
let mut body = BytesMut::new(); let mut buf = BytesMut::new();
while let Some(item) = stream.next().await {
body.extend_from_slice(&item?) pin!(stream);
poll_fn(|cx| loop {
let body = stream.as_mut();
match ready!(body.poll_next(cx)) {
Some(Ok(bytes)) => buf.extend_from_slice(&*bytes),
None => return Poll::Ready(Ok(())),
Some(Err(err)) => return Poll::Ready(Err(err)),
} }
Ok(body) })
.await?;
Ok(buf)
} }
fn tls_config() -> RustlsServerConfig { fn tls_config() -> RustlsServerConfig {

2
actix-multipart/src/server.rs
View File

@ -1233,7 +1233,7 @@ mod tests {
// and should not consume the payload // and should not consume the payload
match payload { match payload {
actix_web::dev::Payload::H1(_) => {} //expected actix_web::dev::Payload::H1 { .. } => {} //expected
_ => unreachable!(), _ => unreachable!(),
} }
} }

6
awc/CHANGES.md
View File

@ -3,8 +3,14 @@
## Unreleased - 2021-xx-xx ## Unreleased - 2021-xx-xx
- Rename `Connector::{ssl => openssl}`. [#2503] - Rename `Connector::{ssl => openssl}`. [#2503]
- Improve `Client` instantiation efficiency when using `openssl` by only building connectors once. [#2503] - Improve `Client` instantiation efficiency when using `openssl` by only building connectors once. [#2503]
- `ClientRequest::send_body` now takes an `impl MessageBody`. [#2546]
- Rename `MessageBody => ResponseBody` to avoid conflicts with `MessageBody` trait. [#2546]
- `impl Future` for `ResponseBody` no longer requires the body type be `Unpin`. [#2546]
- `impl Future` for `JsonBody` no longer requires the body type be `Unpin`. [#2546]
- `impl Stream` for `ClientResponse` no longer requires the body type be `Unpin`. [#2546]
[#2503]: https://github.com/actix/actix-web/pull/2503 [#2503]: https://github.com/actix/actix-web/pull/2503
[#2546]: https://github.com/actix/actix-web/pull/2546
## 3.0.0-beta.14 - 2021-12-17 ## 3.0.0-beta.14 - 2021-12-17

19
awc/src/any_body.rs
View File

@ -77,10 +77,27 @@ impl<B> AnyBody<B>
where where
B: MessageBody + 'static, B: MessageBody + 'static,
{ {
/// Converts a [`MessageBody`] type into the best possible representation.
///
/// Checks size for `None` and tries to convert to `Bytes`. Otherwise, uses the `Body` variant.
pub fn from_message_body(body: B) -> Self
where
B: MessageBody,
{
if matches!(body.size(), BodySize::None) {
return Self::None;
}
match body.try_into_bytes() {
Ok(body) => Self::Bytes { body },
Err(body) => Self::new(body),
}
}
pub fn into_boxed(self) -> AnyBody { pub fn into_boxed(self) -> AnyBody {
match self { match self {
Self::None => AnyBody::None, Self::None => AnyBody::None,
Self::Bytes { body: bytes } => AnyBody::Bytes { body: bytes }, Self::Bytes { body } => AnyBody::Bytes { body },
Self::Body { body } => AnyBody::new_boxed(body), Self::Body { body } => AnyBody::new_boxed(body),
} }
} }

6
awc/src/builder.rs
View File

@ -9,11 +9,13 @@ use actix_rt::net::{ActixStream, TcpStream};
use actix_service::{boxed, Service}; use actix_service::{boxed, Service};
use crate::{ use crate::{
client::{ConnectInfo, Connector, ConnectorService, TcpConnectError, TcpConnection}, client::{
ClientConfig, ConnectInfo, Connector, ConnectorService, TcpConnectError, TcpConnection,
},
connect::DefaultConnector, connect::DefaultConnector,
error::SendRequestError, error::SendRequestError,
middleware::{NestTransform, Redirect, Transform}, middleware::{NestTransform, Redirect, Transform},
Client, ClientConfig, ConnectRequest, ConnectResponse, Client, ConnectRequest, ConnectResponse,
}; };
/// An HTTP Client builder /// An HTTP Client builder

4
awc/src/client/connection.rs
View File

@ -267,7 +267,9 @@ where
Connection::Tls(ConnectionType::H2(conn)) => { Connection::Tls(ConnectionType::H2(conn)) => {
h2proto::send_request(conn, head.into(), body).await h2proto::send_request(conn, head.into(), body).await
} }
_ => unreachable!("Plain Tcp connection can be used only in Http1 protocol"), _ => {
unreachable!("Plain TCP connection can be used only with HTTP/1.1 protocol")
}
} }
}) })
} }

16
awc/src/client/h1proto.rs
View File

@ -13,16 +13,17 @@ use actix_http::{
Payload, RequestHeadType, ResponseHead, StatusCode, Payload, RequestHeadType, ResponseHead, StatusCode,
}; };
use actix_utils::future::poll_fn; use actix_utils::future::poll_fn;
use bytes::buf::BufMut; use bytes::{buf::BufMut, Bytes, BytesMut};
use bytes::{Bytes, BytesMut};
use futures_core::{ready, Stream}; use futures_core::{ready, Stream};
use futures_util::SinkExt as _; use futures_util::SinkExt as _;
use pin_project_lite::pin_project; use pin_project_lite::pin_project;
use crate::BoxError; use crate::BoxError;
use super::connection::{ConnectionIo, H1Connection}; use super::{
use super::error::{ConnectError, SendRequestError}; connection::{ConnectionIo, H1Connection},
error::{ConnectError, SendRequestError},
};
pub(crate) async fn send_request<Io, B>( pub(crate) async fn send_request<Io, B>(
io: H1Connection<Io>, io: H1Connection<Io>,
@ -123,7 +124,12 @@ where
Ok((head, Payload::None)) Ok((head, Payload::None))
} }
_ => Ok((head, Payload::Stream(Box::pin(PlStream::new(framed))))), _ => Ok((
head,
Payload::Stream {
payload: Box::pin(PlStream::new(framed)),
},
)),
} }
} }

188
awc/src/client/mod.rs
View File

@ -1,6 +1,15 @@
//! HTTP client. //! HTTP client.
use http::Uri; use std::{convert::TryFrom, rc::Rc, time::Duration};
use actix_http::{error::HttpError, header::HeaderMap, Method, RequestHead, Uri};
use actix_rt::net::TcpStream;
use actix_service::Service;
pub use actix_tls::connect::{
ConnectError as TcpConnectError, ConnectInfo, Connection as TcpConnection,
};
use crate::{ws, BoxConnectorService, ClientBuilder, ClientRequest};
mod config; mod config;
mod connection; mod connection;
@ -10,10 +19,6 @@ mod h1proto;
mod h2proto; mod h2proto;
mod pool; mod pool;
pub use actix_tls::connect::{
ConnectError as TcpConnectError, ConnectInfo, Connection as TcpConnection,
};
pub use self::connection::{Connection, ConnectionIo}; pub use self::connection::{Connection, ConnectionIo};
pub use self::connector::{Connector, ConnectorService}; pub use self::connector::{Connector, ConnectorService};
pub use self::error::{ConnectError, FreezeRequestError, InvalidUrl, SendRequestError}; pub use self::error::{ConnectError, FreezeRequestError, InvalidUrl, SendRequestError};
@ -23,3 +28,176 @@ pub struct Connect {
pub uri: Uri, pub uri: Uri,
pub addr: Option<std::net::SocketAddr>, pub addr: Option<std::net::SocketAddr>,
} }
/// An asynchronous HTTP and WebSocket client.
///
/// You should take care to create, at most, one `Client` per thread. Otherwise, expect higher CPU
/// and memory usage.
///
/// # Examples
/// ```
/// use awc::Client;
///
/// #[actix_rt::main]
/// async fn main() {
/// let mut client = Client::default();
///
/// let res = client.get("http://www.rust-lang.org")
/// .insert_header(("User-Agent", "my-app/1.2"))
/// .send()
/// .await;
///
/// println!("Response: {:?}", res);
/// }
/// ```
#[derive(Clone)]
pub struct Client(pub(crate) ClientConfig);
#[derive(Clone)]
pub(crate) struct ClientConfig {
pub(crate) connector: BoxConnectorService,
pub(crate) default_headers: Rc<HeaderMap>,
pub(crate) timeout: Option<Duration>,
}
impl Default for Client {
fn default() -> Self {
ClientBuilder::new().finish()
}
}
impl Client {
/// Create new client instance with default settings.
pub fn new() -> Client {
Client::default()
}
/// Create `Client` builder.
/// This function is equivalent of `ClientBuilder::new()`.
pub fn builder() -> ClientBuilder<
impl Service<
ConnectInfo<Uri>,
Response = TcpConnection<Uri, TcpStream>,
Error = TcpConnectError,
> + Clone,
> {
ClientBuilder::new()
}
/// Construct HTTP request.
pub fn request<U>(&self, method: Method, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
let mut req = ClientRequest::new(method, url, self.0.clone());
for header in self.0.default_headers.iter() {
// header map is empty
// TODO: probably append instead
req = req.insert_header_if_none(header);
}
req
}
/// Create `ClientRequest` from `RequestHead`
///
/// It is useful for proxy requests. This implementation
/// copies all headers and the method.
pub fn request_from<U>(&self, url: U, head: &RequestHead) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
let mut req = self.request(head.method.clone(), url);
for header in head.headers.iter() {
req = req.insert_header_if_none(header);
}
req
}
/// Construct HTTP *GET* request.
pub fn get<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::GET, url)
}
/// Construct HTTP *HEAD* request.
pub fn head<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::HEAD, url)
}
/// Construct HTTP *PUT* request.
pub fn put<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::PUT, url)
}
/// Construct HTTP *POST* request.
pub fn post<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::POST, url)
}
/// Construct HTTP *PATCH* request.
pub fn patch<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::PATCH, url)
}
/// Construct HTTP *DELETE* request.
pub fn delete<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::DELETE, url)
}
/// Construct HTTP *OPTIONS* request.
pub fn options<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::OPTIONS, url)
}
/// Initialize a WebSocket connection.
/// Returns a WebSocket connection builder.
pub fn ws<U>(&self, url: U) -> ws::WebsocketsRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
let mut req = ws::WebsocketsRequest::new(url, self.0.clone());
for (key, value) in self.0.default_headers.iter() {
req.head.headers.insert(key.clone(), value.clone());
}
req
}
/// Get default HeaderMap of Client.
///
/// Returns Some(&mut HeaderMap) when Client object is unique
/// (No other clone of client exists at the same time).
pub fn headers(&mut self) -> Option<&mut HeaderMap> {
Rc::get_mut(&mut self.0.default_headers)
}
}

2
awc/src/connect.rs
View File

@ -16,7 +16,7 @@ use crate::{
client::{ client::{
Connect as ClientConnect, ConnectError, Connection, ConnectionIo, SendRequestError, Connect as ClientConnect, ConnectError, Connection, ConnectionIo, SendRequestError,
}, },
response::ClientResponse, ClientResponse,
}; };
pub type BoxConnectorService = Rc< pub type BoxConnectorService = Rc<

1
awc/src/error.rs
View File

@ -1,5 +1,6 @@
//! HTTP client errors //! HTTP client errors
// TODO: figure out how best to expose http::Error vs actix_http::Error
pub use actix_http::{ pub use actix_http::{
error::{HttpError, PayloadError}, error::{HttpError, PayloadError},
header::HeaderValue, header::HeaderValue,

9
awc/src/frozen.rs
View File

@ -5,15 +5,16 @@ use futures_core::Stream;
use serde::Serialize; use serde::Serialize;
use actix_http::{ use actix_http::{
body::MessageBody,
error::HttpError, error::HttpError,
header::{HeaderMap, HeaderName, TryIntoHeaderValue}, header::{HeaderMap, HeaderName, TryIntoHeaderValue},
Method, RequestHead, Uri, Method, RequestHead, Uri,
}; };
use crate::{ use crate::{
any_body::AnyBody, client::ClientConfig,
sender::{RequestSender, SendClientRequest}, sender::{RequestSender, SendClientRequest},
BoxError, ClientConfig, BoxError,
}; };
/// `FrozenClientRequest` struct represents cloneable client request. /// `FrozenClientRequest` struct represents cloneable client request.
@ -46,7 +47,7 @@ impl FrozenClientRequest {
/// Send a body. /// Send a body.
pub fn send_body<B>(&self, body: B) -> SendClientRequest pub fn send_body<B>(&self, body: B) -> SendClientRequest
where where
B: Into<AnyBody>, B: MessageBody + 'static,
{ {
RequestSender::Rc(self.head.clone(), None).send_body( RequestSender::Rc(self.head.clone(), None).send_body(
self.addr, self.addr,
@ -159,7 +160,7 @@ impl FrozenSendBuilder {
/// Complete request construction and send a body. /// Complete request construction and send a body.
pub fn send_body<B>(self, body: B) -> SendClientRequest pub fn send_body<B>(self, body: B) -> SendClientRequest
where where
B: Into<AnyBody>, B: MessageBody + 'static,
{ {
if let Some(e) = self.err { if let Some(e) = self.err {
return e.into(); return e.into();

205
awc/src/lib.rs
View File

@ -105,6 +105,11 @@
#![doc(html_logo_url = "https://actix.rs/img/logo.png")] #![doc(html_logo_url = "https://actix.rs/img/logo.png")]
#![doc(html_favicon_url = "https://actix.rs/favicon.ico")] #![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
pub use actix_http::body;
#[cfg(feature = "cookies")]
pub use cookie;
mod any_body; mod any_body;
mod builder; mod builder;
mod client; mod client;
@ -113,203 +118,27 @@ pub mod error;
mod frozen; mod frozen;
pub mod middleware; pub mod middleware;
mod request; mod request;
mod response; mod responses;
mod sender; mod sender;
pub mod test; pub mod test;
pub mod ws; pub mod ws;
// TODO: hmmmmmm pub mod http {
pub use actix_http as http; //! Various HTTP related types.
#[cfg(feature = "cookies")]
pub use cookie; // TODO: figure out how best to expose http::Error vs actix_http::Error
pub use actix_http::{
header, uri, ConnectionType, Error, Method, StatusCode, Uri, Version,
};
}
pub use self::builder::ClientBuilder; pub use self::builder::ClientBuilder;
pub use self::client::Connector; pub use self::client::{Client, Connector};
pub use self::connect::{BoxConnectorService, BoxedSocket, ConnectRequest, ConnectResponse}; pub use self::connect::{BoxConnectorService, BoxedSocket, ConnectRequest, ConnectResponse};
pub use self::frozen::{FrozenClientRequest, FrozenSendBuilder}; pub use self::frozen::{FrozenClientRequest, FrozenSendBuilder};
pub use self::request::ClientRequest; pub use self::request::ClientRequest;
pub use self::response::{ClientResponse, JsonBody, MessageBody}; #[allow(deprecated)]
pub use self::responses::{ClientResponse, JsonBody, MessageBody, ResponseBody};
pub use self::sender::SendClientRequest; pub use self::sender::SendClientRequest;
use std::{convert::TryFrom, rc::Rc, time::Duration};
use actix_http::{error::HttpError, header::HeaderMap, Method, RequestHead, Uri};
use actix_rt::net::TcpStream;
use actix_service::Service;
use self::client::{ConnectInfo, TcpConnectError, TcpConnection};
pub(crate) type BoxError = Box<dyn std::error::Error>; pub(crate) type BoxError = Box<dyn std::error::Error>;
/// An asynchronous HTTP and WebSocket client.
///
/// You should take care to create, at most, one `Client` per thread. Otherwise, expect higher CPU
/// and memory usage.
///
/// # Examples
/// ```
/// use awc::Client;
///
/// #[actix_rt::main]
/// async fn main() {
/// let mut client = Client::default();
///
/// let res = client.get("http://www.rust-lang.org")
/// .insert_header(("User-Agent", "my-app/1.2"))
/// .send()
/// .await;
///
/// println!("Response: {:?}", res);
/// }
/// ```
#[derive(Clone)]
pub struct Client(ClientConfig);
#[derive(Clone)]
pub(crate) struct ClientConfig {
pub(crate) connector: BoxConnectorService,
pub(crate) default_headers: Rc<HeaderMap>,
pub(crate) timeout: Option<Duration>,
}
impl Default for Client {
fn default() -> Self {
ClientBuilder::new().finish()
}
}
impl Client {
/// Create new client instance with default settings.
pub fn new() -> Client {
Client::default()
}
/// Create `Client` builder.
/// This function is equivalent of `ClientBuilder::new()`.
pub fn builder() -> ClientBuilder<
impl Service<
ConnectInfo<Uri>,
Response = TcpConnection<Uri, TcpStream>,
Error = TcpConnectError,
> + Clone,
> {
ClientBuilder::new()
}
/// Construct HTTP request.
pub fn request<U>(&self, method: Method, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
let mut req = ClientRequest::new(method, url, self.0.clone());
for header in self.0.default_headers.iter() {
// header map is empty
// TODO: probably append instead
req = req.insert_header_if_none(header);
}
req
}
/// Create `ClientRequest` from `RequestHead`
///
/// It is useful for proxy requests. This implementation
/// copies all headers and the method.
pub fn request_from<U>(&self, url: U, head: &RequestHead) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
let mut req = self.request(head.method.clone(), url);
for header in head.headers.iter() {
req = req.insert_header_if_none(header);
}
req
}
/// Construct HTTP *GET* request.
pub fn get<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::GET, url)
}
/// Construct HTTP *HEAD* request.
pub fn head<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::HEAD, url)
}
/// Construct HTTP *PUT* request.
pub fn put<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::PUT, url)
}
/// Construct HTTP *POST* request.
pub fn post<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::POST, url)
}
/// Construct HTTP *PATCH* request.
pub fn patch<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::PATCH, url)
}
/// Construct HTTP *DELETE* request.
pub fn delete<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::DELETE, url)
}
/// Construct HTTP *OPTIONS* request.
pub fn options<U>(&self, url: U) -> ClientRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
self.request(Method::OPTIONS, url)
}
/// Initialize a WebSocket connection.
/// Returns a WebSocket connection builder.
pub fn ws<U>(&self, url: U) -> ws::WebsocketsRequest
where
Uri: TryFrom<U>,
<Uri as TryFrom<U>>::Error: Into<HttpError>,
{
let mut req = ws::WebsocketsRequest::new(url, self.0.clone());
for (key, value) in self.0.default_headers.iter() {
req.head.headers.insert(key.clone(), value.clone());
}
req
}
/// Get default HeaderMap of Client.
///
/// Returns Some(&mut HeaderMap) when Client object is unique
/// (No other clone of client exists at the same time).
pub fn headers(&mut self) -> Option<&mut HeaderMap> {
Rc::get_mut(&mut self.0.default_headers)
}
}

4
awc/src/middleware/redirect.rs
View File

@ -190,9 +190,7 @@ where
let body_new = if is_redirect { let body_new = if is_redirect {
// try to reuse body // try to reuse body
match body { match body {
Some(ref bytes) => AnyBody::Bytes { Some(ref bytes) => AnyBody::from(bytes.clone()),
body: bytes.clone(),
},
// TODO: should this be AnyBody::Empty or AnyBody::None. // TODO: should this be AnyBody::Empty or AnyBody::None.
_ => AnyBody::empty(), _ => AnyBody::empty(),
} }

48
awc/src/request.rs
View File

@ -5,17 +5,18 @@ use futures_core::Stream;
use serde::Serialize; use serde::Serialize;
use actix_http::{ use actix_http::{
body::MessageBody,
error::HttpError, error::HttpError,
header::{self, HeaderMap, HeaderValue, TryIntoHeaderPair}, header::{self, HeaderMap, HeaderValue, TryIntoHeaderPair},
ConnectionType, Method, RequestHead, Uri, Version, ConnectionType, Method, RequestHead, Uri, Version,
}; };
use crate::{ use crate::{
any_body::AnyBody, client::ClientConfig,
error::{FreezeRequestError, InvalidUrl}, error::{FreezeRequestError, InvalidUrl},
frozen::FrozenClientRequest, frozen::FrozenClientRequest,
sender::{PrepForSendingError, RequestSender, SendClientRequest}, sender::{PrepForSendingError, RequestSender, SendClientRequest},
BoxError, ClientConfig, BoxError,
}; };
#[cfg(feature = "cookies")] #[cfg(feature = "cookies")]
@ -26,9 +27,9 @@ use crate::cookie::{Cookie, CookieJar};
/// This type can be used to construct an instance of `ClientRequest` through a /// This type can be used to construct an instance of `ClientRequest` through a
/// builder-like pattern. /// builder-like pattern.
/// ///
/// ``` /// ```no_run
/// #[actix_rt::main] /// # #[actix_rt::main]
/// async fn main() { /// # async fn main() {
/// let response = awc::Client::new() /// let response = awc::Client::new()
/// .get("http://www.rust-lang.org") // <- Create request builder /// .get("http://www.rust-lang.org") // <- Create request builder
/// .insert_header(("User-Agent", "Actix-web")) /// .insert_header(("User-Agent", "Actix-web"))
@ -39,7 +40,7 @@ use crate::cookie::{Cookie, CookieJar};
/// println!("Response: {:?}", response); /// println!("Response: {:?}", response);
/// Ok(()) /// Ok(())
/// }); /// });
/// } /// # }
/// ``` /// ```
pub struct ClientRequest { pub struct ClientRequest {
pub(crate) head: RequestHead, pub(crate) head: RequestHead,
@ -174,17 +175,13 @@ impl ClientRequest {
/// Append a header, keeping any that were set with an equivalent field name. /// Append a header, keeping any that were set with an equivalent field name.
/// ///
/// ``` /// ```no_run
/// # #[actix_rt::main] /// use awc::{http::header, Client};
/// # async fn main() {
/// # use awc::Client;
/// use awc::http::header::CONTENT_TYPE;
/// ///
/// Client::new() /// Client::new()
/// .get("http://www.rust-lang.org") /// .get("http://www.rust-lang.org")
/// .insert_header(("X-TEST", "value")) /// .insert_header(("X-TEST", "value"))
/// .insert_header((CONTENT_TYPE, mime::APPLICATION_JSON)); /// .insert_header((header::CONTENT_TYPE, mime::APPLICATION_JSON));
/// # }
/// ``` /// ```
pub fn append_header(mut self, header: impl TryIntoHeaderPair) -> Self { pub fn append_header(mut self, header: impl TryIntoHeaderPair) -> Self {
match header.try_into_pair() { match header.try_into_pair() {
@ -252,23 +249,18 @@ impl ClientRequest {
/// Set a cookie /// Set a cookie
/// ///
/// ``` /// ```no_run
/// #[actix_rt::main] /// use awc::{cookie::Cookie, Client};
/// async fn main() { ///
/// let resp = awc::Client::new().get("https://www.rust-lang.org") /// # #[actix_rt::main]
/// .cookie( /// # async fn main() {
/// awc::cookie::Cookie::build("name", "value") /// let res = Client::new().get("https://httpbin.org/cookies")
/// .domain("www.rust-lang.org") /// .cookie(Cookie::new("name", "value"))
/// .path("/")
/// .secure(true)
/// .http_only(true)
/// .finish(),
/// )
/// .send() /// .send()
/// .await; /// .await;
/// ///
/// println!("Response: {:?}", resp); /// println!("Response: {:?}", res);
/// } /// # }
/// ``` /// ```
#[cfg(feature = "cookies")] #[cfg(feature = "cookies")]
pub fn cookie(mut self, cookie: Cookie<'_>) -> Self { pub fn cookie(mut self, cookie: Cookie<'_>) -> Self {
@ -340,7 +332,7 @@ impl ClientRequest {
/// Complete request construction and send body. /// Complete request construction and send body.
pub fn send_body<B>(self, body: B) -> SendClientRequest pub fn send_body<B>(self, body: B) -> SendClientRequest
where where
B: Into<AnyBody>, B: MessageBody + 'static,
{ {
let slf = match self.prep_for_sending() { let slf = match self.prep_for_sending() {
Ok(slf) => slf, Ok(slf) => slf,

556
awc/src/response.rs
View File

@ -1,556 +0,0 @@
use std::{
cell::{Ref, RefMut},
fmt,
future::Future,
io,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
time::{Duration, Instant},
};
use actix_http::{
error::PayloadError, header, header::HeaderMap, Extensions, HttpMessage, Payload,
PayloadStream, ResponseHead, StatusCode, Version,
};
use actix_rt::time::{sleep, Sleep};
use bytes::{Bytes, BytesMut};
use futures_core::{ready, Stream};
use serde::de::DeserializeOwned;
#[cfg(feature = "cookies")]
use crate::cookie::{Cookie, ParseError as CookieParseError};
use crate::error::JsonPayloadError;
/// Client Response
pub struct ClientResponse<S = PayloadStream> {
pub(crate) head: ResponseHead,
pub(crate) payload: Payload<S>,
pub(crate) timeout: ResponseTimeout,
}
/// helper enum with reusable sleep passed from `SendClientResponse`.
/// See `ClientResponse::_timeout` for reason.
pub(crate) enum ResponseTimeout {
Disabled(Option<Pin<Box<Sleep>>>),
Enabled(Pin<Box<Sleep>>),
}
impl Default for ResponseTimeout {
fn default() -> Self {
Self::Disabled(None)
}
}
impl ResponseTimeout {
fn poll_timeout(&mut self, cx: &mut Context<'_>) -> Result<(), PayloadError> {
match *self {
Self::Enabled(ref mut timeout) => {
if timeout.as_mut().poll(cx).is_ready() {
Err(PayloadError::Io(io::Error::new(
io::ErrorKind::TimedOut,
"Response Payload IO timed out",
)))
} else {
Ok(())
}
}
Self::Disabled(_) => Ok(()),
}
}
}
impl<S> HttpMessage for ClientResponse<S> {
type Stream = S;
fn headers(&self) -> &HeaderMap {
&self.head.headers
}
fn take_payload(&mut self) -> Payload<S> {
std::mem::replace(&mut self.payload, Payload::None)
}
fn extensions(&self) -> Ref<'_, Extensions> {
self.head.extensions()
}
fn extensions_mut(&self) -> RefMut<'_, Extensions> {
self.head.extensions_mut()
}
}
impl<S> ClientResponse<S> {
/// Create new Request instance
pub(crate) fn new(head: ResponseHead, payload: Payload<S>) -> Self {
ClientResponse {
head,
payload,
timeout: ResponseTimeout::default(),
}
}
#[inline]
pub(crate) fn head(&self) -> &ResponseHead {
&self.head
}
/// Read the Request Version.
#[inline]
pub fn version(&self) -> Version {
self.head().version
}
/// Get the status from the server.
#[inline]
pub fn status(&self) -> StatusCode {
self.head().status
}
#[inline]
/// Returns request's headers.
pub fn headers(&self) -> &HeaderMap {
&self.head().headers
}
/// Set a body and return previous body value
pub fn map_body<F, U>(mut self, f: F) -> ClientResponse<U>
where
F: FnOnce(&mut ResponseHead, Payload<S>) -> Payload<U>,
{
let payload = f(&mut self.head, self.payload);
ClientResponse {
payload,
head: self.head,
timeout: self.timeout,
}
}
/// Set a timeout duration for [`ClientResponse`](self::ClientResponse).
///
/// This duration covers the duration of processing the response body stream
/// and would end it as timeout error when deadline met.
///
/// Disabled by default.
pub fn timeout(self, dur: Duration) -> Self {
let timeout = match self.timeout {
ResponseTimeout::Disabled(Some(mut timeout))
| ResponseTimeout::Enabled(mut timeout) => match Instant::now().checked_add(dur) {
Some(deadline) => {
timeout.as_mut().reset(deadline.into());
ResponseTimeout::Enabled(timeout)
}
None => ResponseTimeout::Enabled(Box::pin(sleep(dur))),
},
_ => ResponseTimeout::Enabled(Box::pin(sleep(dur))),
};
Self {
payload: self.payload,
head: self.head,
timeout,
}
}
/// This method does not enable timeout. It's used to pass the boxed `Sleep` from
/// `SendClientRequest` and reuse it's heap allocation together with it's slot in
/// timer wheel.
pub(crate) fn _timeout(mut self, timeout: Option<Pin<Box<Sleep>>>) -> Self {
self.timeout = ResponseTimeout::Disabled(timeout);
self
}
/// Load request cookies.
#[cfg(feature = "cookies")]
pub 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(&header::SET_COOKIE) {
let s = std::str::from_utf8(hdr.as_bytes()).map_err(CookieParseError::from)?;
cookies.push(Cookie::parse_encoded(s)?.into_owned());
}
self.extensions_mut().insert(Cookies(cookies));
}
Ok(Ref::map(self.extensions(), |ext| {
&ext.get::<Cookies>().unwrap().0
}))
}
/// Return request cookie.
#[cfg(feature = "cookies")]
pub fn cookie(&self, name: &str) -> Option<Cookie<'static>> {
if let Ok(cookies) = self.cookies() {
for cookie in cookies.iter() {
if cookie.name() == name {
return Some(cookie.to_owned());
}
}
}
None
}
}
impl<S> ClientResponse<S>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
{
/// Loads HTTP response's body.
pub fn body(&mut self) -> MessageBody<S> {
MessageBody::new(self)
}
/// Loads and parse `application/json` encoded body.
/// Return `JsonBody<T>` future. It resolves to a `T` value.
///
/// Returns error:
///
/// * content type is not `application/json`
/// * content length is greater than 256k
pub fn json<T: DeserializeOwned>(&mut self) -> JsonBody<S, T> {
JsonBody::new(self)
}
}
impl<S> Stream for ClientResponse<S>
where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
{
type Item = Result<Bytes, PayloadError>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
this.timeout.poll_timeout(cx)?;
Pin::new(&mut this.payload).poll_next(cx)
}
}
impl<S> fmt::Debug for ClientResponse<S> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(f, "\nClientResponse {:?} {}", self.version(), self.status(),)?;
writeln!(f, " headers:")?;
for (key, val) in self.headers().iter() {
writeln!(f, " {:?}: {:?}", key, val)?;
}
Ok(())
}
}
const DEFAULT_BODY_LIMIT: usize = 2 * 1024 * 1024;
/// Future that resolves to a complete HTTP message body.
pub struct MessageBody<S> {
length: Option<usize>,
timeout: ResponseTimeout,
body: Result<ReadBody<S>, Option<PayloadError>>,
}
impl<S> MessageBody<S>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
{
/// Create `MessageBody` for request.
pub fn new(res: &mut ClientResponse<S>) -> MessageBody<S> {
let length = match res.headers().get(&header::CONTENT_LENGTH) {
Some(value) => {
let len = value.to_str().ok().and_then(|s| s.parse::<usize>().ok());
match len {
None => return Self::err(PayloadError::UnknownLength),
len => len,
}
}
None => None,
};
MessageBody {
length,
timeout: std::mem::take(&mut res.timeout),
body: Ok(ReadBody::new(res.take_payload(), DEFAULT_BODY_LIMIT)),
}
}
/// Change max size of payload. By default max size is 2048kB
pub fn limit(mut self, limit: usize) -> Self {
if let Ok(ref mut body) = self.body {
body.limit = limit;
}
self
}
fn err(e: PayloadError) -> Self {
MessageBody {
length: None,
timeout: ResponseTimeout::default(),
body: Err(Some(e)),
}
}
}
impl<S> Future for MessageBody<S>
where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
{
type Output = Result<Bytes, PayloadError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
match this.body {
Err(ref mut err) => Poll::Ready(Err(err.take().unwrap())),
Ok(ref mut body) => {
if let Some(len) = this.length.take() {
if len > body.limit {
return Poll::Ready(Err(PayloadError::Overflow));
}
}
this.timeout.poll_timeout(cx)?;
Pin::new(body).poll(cx)
}
}
}
}
/// Response's payload json parser, it resolves to a deserialized `T` value.
///
/// Returns error:
///
/// * content type is not `application/json`
/// * content length is greater than 64k
pub struct JsonBody<S, U> {
length: Option<usize>,
err: Option<JsonPayloadError>,
timeout: ResponseTimeout,
fut: Option<ReadBody<S>>,
_phantom: PhantomData<U>,
}
impl<S, U> JsonBody<S, U>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
U: DeserializeOwned,
{
/// Create `JsonBody` for request.
pub fn new(res: &mut ClientResponse<S>) -> Self {
// check content-type
let json = if let Ok(Some(mime)) = res.mime_type() {
mime.subtype() == mime::JSON || mime.suffix() == Some(mime::JSON)
} else {
false
};
if !json {
return JsonBody {
length: None,
fut: None,
timeout: ResponseTimeout::default(),
err: Some(JsonPayloadError::ContentType),
_phantom: PhantomData,
};
}
let mut len = None;
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)
}
}
}
JsonBody {
length: len,
err: None,
timeout: std::mem::take(&mut res.timeout),
fut: Some(ReadBody::new(res.take_payload(), 65536)),
_phantom: PhantomData,
}
}
/// Change max size of payload. By default max size is 64kB
pub fn limit(mut self, limit: usize) -> Self {
if let Some(ref mut fut) = self.fut {
fut.limit = limit;
}
self
}
}
impl<T, U> Unpin for JsonBody<T, U>
where
T: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
U: DeserializeOwned,
{
}
impl<T, U> Future for JsonBody<T, U>
where
T: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
U: DeserializeOwned,
{
type Output = Result<U, JsonPayloadError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
if let Some(err) = self.err.take() {
return Poll::Ready(Err(err));
}
if let Some(len) = self.length.take() {
if len > self.fut.as_ref().unwrap().limit {
return Poll::Ready(Err(JsonPayloadError::Payload(PayloadError::Overflow)));
}
}
self.timeout
.poll_timeout(cx)
.map_err(JsonPayloadError::Payload)?;
let body = ready!(Pin::new(&mut self.get_mut().fut.as_mut().unwrap()).poll(cx))?;
Poll::Ready(serde_json::from_slice::<U>(&body).map_err(JsonPayloadError::from))
}
}
struct ReadBody<S> {
stream: Payload<S>,
buf: BytesMut,
limit: usize,
}
impl<S> ReadBody<S> {
fn new(stream: Payload<S>, limit: usize) -> Self {
Self {
stream,
buf: BytesMut::new(),
limit,
}
}
}
impl<S> Future for ReadBody<S>
where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
{
type Output = Result<Bytes, PayloadError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
while let Some(chunk) = ready!(Pin::new(&mut this.stream).poll_next(cx)?) {
if (this.buf.len() + chunk.len()) > this.limit {
return Poll::Ready(Err(PayloadError::Overflow));
}
this.buf.extend_from_slice(&chunk);
}
Poll::Ready(Ok(this.buf.split().freeze()))
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde::{Deserialize, Serialize};
use crate::{http::header, test::TestResponse};
#[actix_rt::test]
async fn test_body() {
let mut req = TestResponse::with_header((header::CONTENT_LENGTH, "xxxx")).finish();
match req.body().await.err().unwrap() {
PayloadError::UnknownLength => {}
_ => unreachable!("error"),
}
let mut req = TestResponse::with_header((header::CONTENT_LENGTH, "10000000")).finish();
match req.body().await.err().unwrap() {
PayloadError::Overflow => {}
_ => unreachable!("error"),
}
let mut req = TestResponse::default()
.set_payload(Bytes::from_static(b"test"))
.finish();
assert_eq!(req.body().await.ok().unwrap(), Bytes::from_static(b"test"));
let mut req = TestResponse::default()
.set_payload(Bytes::from_static(b"11111111111111"))
.finish();
match req.body().limit(5).await.err().unwrap() {
PayloadError::Overflow => {}
_ => unreachable!("error"),
}
}
#[derive(Serialize, Deserialize, PartialEq, Debug)]
struct MyObject {
name: String,
}
fn json_eq(err: JsonPayloadError, other: JsonPayloadError) -> bool {
match err {
JsonPayloadError::Payload(PayloadError::Overflow) => {
matches!(other, JsonPayloadError::Payload(PayloadError::Overflow))
}
JsonPayloadError::ContentType => matches!(other, JsonPayloadError::ContentType),
_ => false,
}
}
#[actix_rt::test]
async fn test_json_body() {
let mut req = TestResponse::default().finish();
let json = JsonBody::<_, MyObject>::new(&mut req).await;
assert!(json_eq(json.err().unwrap(), JsonPayloadError::ContentType));
let mut req = TestResponse::default()
.insert_header((
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/text"),
))
.finish();
let json = JsonBody::<_, MyObject>::new(&mut req).await;
assert!(json_eq(json.err().unwrap(), JsonPayloadError::ContentType));
let mut req = TestResponse::default()
.insert_header((
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/json"),
))
.insert_header((
header::CONTENT_LENGTH,
header::HeaderValue::from_static("10000"),
))
.finish();
let json = JsonBody::<_, MyObject>::new(&mut req).limit(100).await;
assert!(json_eq(
json.err().unwrap(),
JsonPayloadError::Payload(PayloadError::Overflow)
));
let mut req = TestResponse::default()
.insert_header((
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/json"),
))
.insert_header((
header::CONTENT_LENGTH,
header::HeaderValue::from_static("16"),
))
.set_payload(Bytes::from_static(b"{\"name\": \"test\"}"))
.finish();
let json = JsonBody::<_, MyObject>::new(&mut req).await;
assert_eq!(
json.ok().unwrap(),
MyObject {
name: "test".to_owned()
}
);
}
}

192
awc/src/responses/json_body.rs Normal file
View File

@ -0,0 +1,192 @@
use std::{
future::Future,
marker::PhantomData,
mem,
pin::Pin,
task::{Context, Poll},
};
use actix_http::{error::PayloadError, header, HttpMessage};
use bytes::Bytes;
use futures_core::{ready, Stream};
use pin_project_lite::pin_project;
use serde::de::DeserializeOwned;
use super::{read_body::ReadBody, ResponseTimeout, DEFAULT_BODY_LIMIT};
use crate::{error::JsonPayloadError, ClientResponse};
pin_project! {
/// A `Future` that reads a body stream, parses JSON, resolving to a deserialized `T`.
///
/// # Errors
/// `Future` implementation returns error if:
/// - content type is not `application/json`;
/// - content length is greater than [limit](JsonBody::limit) (default: 2 MiB).
pub struct JsonBody<S, T> {
#[pin]
body: Option<ReadBody<S>>,
length: Option<usize>,
timeout: ResponseTimeout,
err: Option<JsonPayloadError>,
_phantom: PhantomData<T>,
}
}
impl<S, T> JsonBody<S, T>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
T: DeserializeOwned,
{
/// Creates a JSON body stream reader from a response by taking its payload.
pub fn new(res: &mut ClientResponse<S>) -> Self {
// check content-type
let json = if let Ok(Some(mime)) = res.mime_type() {
mime.subtype() == mime::JSON || mime.suffix() == Some(mime::JSON)
} else {
false
};
if !json {
return JsonBody {
length: None,
body: None,
timeout: ResponseTimeout::default(),
err: Some(JsonPayloadError::ContentType),
_phantom: PhantomData,
};
}
let length = res
.headers()
.get(&header::CONTENT_LENGTH)
.and_then(|len_hdr| len_hdr.to_str().ok())
.and_then(|len_str| len_str.parse::<usize>().ok());
JsonBody {
body: Some(ReadBody::new(res.take_payload(), DEFAULT_BODY_LIMIT)),
length,
timeout: mem::take(&mut res.timeout),
err: None,
_phantom: PhantomData,
}
}
/// Change max size of payload. Default limit is 2 MiB.
pub fn limit(mut self, limit: usize) -> Self {
if let Some(ref mut fut) = self.body {
fut.limit = limit;
}
self
}
}
impl<S, T> Future for JsonBody<S, T>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
T: DeserializeOwned,
{
type Output = Result<T, JsonPayloadError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
if let Some(err) = this.err.take() {
return Poll::Ready(Err(err));
}
if let Some(len) = this.length.take() {
let body = Option::as_ref(&this.body).unwrap();
if len > body.limit {
return Poll::Ready(Err(JsonPayloadError::Payload(PayloadError::Overflow)));
}
}
this.timeout
.poll_timeout(cx)
.map_err(JsonPayloadError::Payload)?;
let body = ready!(this.body.as_pin_mut().unwrap().poll(cx))?;
Poll::Ready(serde_json::from_slice::<T>(&body).map_err(JsonPayloadError::from))
}
}
#[cfg(test)]
mod tests {
use actix_http::BoxedPayloadStream;
use serde::{Deserialize, Serialize};
use static_assertions::assert_impl_all;
use super::*;
use crate::{http::header, test::TestResponse};
assert_impl_all!(JsonBody<BoxedPayloadStream, String>: Unpin);
#[derive(Serialize, Deserialize, PartialEq, Debug)]
struct MyObject {
name: String,
}
fn json_eq(err: JsonPayloadError, other: JsonPayloadError) -> bool {
match err {
JsonPayloadError::Payload(PayloadError::Overflow) => {
matches!(other, JsonPayloadError::Payload(PayloadError::Overflow))
}
JsonPayloadError::ContentType => matches!(other, JsonPayloadError::ContentType),
_ => false,
}
}
#[actix_rt::test]
async fn read_json_body() {
let mut req = TestResponse::default().finish();
let json = JsonBody::<_, MyObject>::new(&mut req).await;
assert!(json_eq(json.err().unwrap(), JsonPayloadError::ContentType));
let mut req = TestResponse::default()
.insert_header((
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/text"),
))
.finish();
let json = JsonBody::<_, MyObject>::new(&mut req).await;
assert!(json_eq(json.err().unwrap(), JsonPayloadError::ContentType));
let mut req = TestResponse::default()
.insert_header((
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/json"),
))
.insert_header((
header::CONTENT_LENGTH,
header::HeaderValue::from_static("10000"),
))
.finish();
let json = JsonBody::<_, MyObject>::new(&mut req).limit(100).await;
assert!(json_eq(
json.err().unwrap(),
JsonPayloadError::Payload(PayloadError::Overflow)
));
let mut req = TestResponse::default()
.insert_header((
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/json"),
))
.insert_header((
header::CONTENT_LENGTH,
header::HeaderValue::from_static("16"),
))
.set_payload(Bytes::from_static(b"{\"name\": \"test\"}"))
.finish();
let json = JsonBody::<_, MyObject>::new(&mut req).await;
assert_eq!(
json.ok().unwrap(),
MyObject {
name: "test".to_owned()
}
);
}
}

49
awc/src/responses/mod.rs Normal file
View File

@ -0,0 +1,49 @@
use std::{future::Future, io, pin::Pin, task::Context};
use actix_http::error::PayloadError;
use actix_rt::time::Sleep;
mod json_body;
mod read_body;
mod response;
mod response_body;
pub use self::json_body::JsonBody;
pub use self::response::ClientResponse;
#[allow(deprecated)]
pub use self::response_body::{MessageBody, ResponseBody};
/// Default body size limit: 2 MiB
const DEFAULT_BODY_LIMIT: usize = 2 * 1024 * 1024;
/// Helper enum with reusable sleep passed from `SendClientResponse`.
///
/// See [`ClientResponse::_timeout`] for reason.
pub(crate) enum ResponseTimeout {
Disabled(Option<Pin<Box<Sleep>>>),
Enabled(Pin<Box<Sleep>>),
}
impl Default for ResponseTimeout {
fn default() -> Self {
Self::Disabled(None)
}
}
impl ResponseTimeout {
fn poll_timeout(&mut self, cx: &mut Context<'_>) -> Result<(), PayloadError> {
match *self {
Self::Enabled(ref mut timeout) => {
if timeout.as_mut().poll(cx).is_ready() {
Err(PayloadError::Io(io::Error::new(
io::ErrorKind::TimedOut,
"Response Payload IO timed out",
)))
} else {
Ok(())
}
}
Self::Disabled(_) => Ok(()),
}
}
}

61
awc/src/responses/read_body.rs Normal file
View File

@ -0,0 +1,61 @@
use std::{
future::Future,
pin::Pin,
task::{Context, Poll},
};
use actix_http::{error::PayloadError, Payload};
use bytes::{Bytes, BytesMut};
use futures_core::{ready, Stream};
use pin_project_lite::pin_project;
pin_project! {
pub(crate) struct ReadBody<S> {
#[pin]
pub(crate) stream: Payload<S>,
pub(crate) buf: BytesMut,
pub(crate) limit: usize,
}
}
impl<S> ReadBody<S> {
pub(crate) fn new(stream: Payload<S>, limit: usize) -> Self {
Self {
stream,
buf: BytesMut::new(),
limit,
}
}
}
impl<S> Future for ReadBody<S>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
{
type Output = Result<Bytes, PayloadError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
while let Some(chunk) = ready!(this.stream.as_mut().poll_next(cx)?) {
if (this.buf.len() + chunk.len()) > *this.limit {
return Poll::Ready(Err(PayloadError::Overflow));
}
this.buf.extend_from_slice(&chunk);
}
Poll::Ready(Ok(this.buf.split().freeze()))
}
}
#[cfg(test)]
mod tests {
use static_assertions::assert_impl_all;
use super::*;
use crate::any_body::AnyBody;
assert_impl_all!(ReadBody<()>: Unpin);
assert_impl_all!(ReadBody<AnyBody>: Unpin);
}

257
awc/src/responses/response.rs Normal file
View File

@ -0,0 +1,257 @@
use std::{
cell::{Ref, RefMut},
fmt, mem,
pin::Pin,
task::{Context, Poll},
time::{Duration, Instant},
};
use actix_http::{
error::PayloadError, header, header::HeaderMap, BoxedPayloadStream, Extensions,
HttpMessage, Payload, ResponseHead, StatusCode, Version,
};
use actix_rt::time::{sleep, Sleep};
use bytes::Bytes;
use futures_core::Stream;
use pin_project_lite::pin_project;
use serde::de::DeserializeOwned;
#[cfg(feature = "cookies")]
use crate::cookie::{Cookie, ParseError as CookieParseError};
use super::{JsonBody, ResponseBody, ResponseTimeout};
pin_project! {
/// Client Response
pub struct ClientResponse<S = BoxedPayloadStream> {
pub(crate) head: ResponseHead,
#[pin]
pub(crate) payload: Payload<S>,
pub(crate) timeout: ResponseTimeout,
}
}
impl<S> ClientResponse<S> {
/// Create new Request instance
pub(crate) fn new(head: ResponseHead, payload: Payload<S>) -> Self {
ClientResponse {
head,
payload,
timeout: ResponseTimeout::default(),
}
}
#[inline]
pub(crate) fn head(&self) -> &ResponseHead {
&self.head
}
/// Read the Request Version.
#[inline]
pub fn version(&self) -> Version {
self.head().version
}
/// Get the status from the server.
#[inline]
pub fn status(&self) -> StatusCode {
self.head().status
}
#[inline]
/// Returns request's headers.
pub fn headers(&self) -> &HeaderMap {
&self.head().headers
}
/// Set a body and return previous body value
pub fn map_body<F, U>(mut self, f: F) -> ClientResponse<U>
where
F: FnOnce(&mut ResponseHead, Payload<S>) -> Payload<U>,
{
let payload = f(&mut self.head, self.payload);
ClientResponse {
payload,
head: self.head,
timeout: self.timeout,
}
}
/// Set a timeout duration for [`ClientResponse`](self::ClientResponse).
///
/// This duration covers the duration of processing the response body stream
/// and would end it as timeout error when deadline met.
///
/// Disabled by default.
pub fn timeout(self, dur: Duration) -> Self {
let timeout = match self.timeout {
ResponseTimeout::Disabled(Some(mut timeout))
| ResponseTimeout::Enabled(mut timeout) => match Instant::now().checked_add(dur) {
Some(deadline) => {
timeout.as_mut().reset(deadline.into());
ResponseTimeout::Enabled(timeout)
}
None => ResponseTimeout::Enabled(Box::pin(sleep(dur))),
},
_ => ResponseTimeout::Enabled(Box::pin(sleep(dur))),
};
Self {
payload: self.payload,
head: self.head,
timeout,
}
}
/// This method does not enable timeout. It's used to pass the boxed `Sleep` from
/// `SendClientRequest` and reuse it's heap allocation together with it's slot in
/// timer wheel.
pub(crate) fn _timeout(mut self, timeout: Option<Pin<Box<Sleep>>>) -> Self {
self.timeout = ResponseTimeout::Disabled(timeout);
self
}
/// Load request cookies.
#[cfg(feature = "cookies")]
pub 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(&header::SET_COOKIE) {
let s = std::str::from_utf8(hdr.as_bytes()).map_err(CookieParseError::from)?;
cookies.push(Cookie::parse_encoded(s)?.into_owned());
}
self.extensions_mut().insert(Cookies(cookies));
}
Ok(Ref::map(self.extensions(), |ext| {
&ext.get::<Cookies>().unwrap().0
}))
}
/// Return request cookie.
#[cfg(feature = "cookies")]
pub fn cookie(&self, name: &str) -> Option<Cookie<'static>> {
if let Ok(cookies) = self.cookies() {
for cookie in cookies.iter() {
if cookie.name() == name {
return Some(cookie.to_owned());
}
}
}
None
}
}
impl<S> ClientResponse<S>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
{
/// Returns a [`Future`] that consumes the body stream and resolves to [`Bytes`].
///
/// # Errors
/// `Future` implementation returns error if:
/// - content type is not `application/json`
/// - content length is greater than [limit](JsonBody::limit) (default: 2 MiB)
///
/// # Examples
/// ```no_run
/// # use awc::Client;
/// # use bytes::Bytes;
/// # #[actix_rt::main]
/// # async fn async_ctx() -> Result<(), Box<dyn std::error::Error>> {
/// let client = Client::default();
/// let mut res = client.get("https://httpbin.org/robots.txt").send().await?;
/// let body: Bytes = res.body().await?;
/// # Ok(())
/// # }
/// ```
///
/// [`Future`]: std::future::Future
pub fn body(&mut self) -> ResponseBody<S> {
ResponseBody::new(self)
}
/// Returns a [`Future`] consumes the body stream, parses JSON, and resolves to a deserialized
/// `T` value.
///
/// # Errors
/// Future returns error if:
/// - content type is not `application/json`;
/// - content length is greater than [limit](JsonBody::limit) (default: 2 MiB).
///
/// # Examples
/// ```no_run
/// # use awc::Client;
/// # #[actix_rt::main]
/// # async fn async_ctx() -> Result<(), Box<dyn std::error::Error>> {
/// let client = Client::default();
/// let mut res = client.get("https://httpbin.org/json").send().await?;
/// let val = res.json::<serde_json::Value>().await?;
/// assert!(val.is_object());
/// # Ok(())
/// # }
/// ```
///
/// [`Future`]: std::future::Future
pub fn json<T: DeserializeOwned>(&mut self) -> JsonBody<S, T> {
JsonBody::new(self)
}
}
impl<S> fmt::Debug for ClientResponse<S> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(f, "\nClientResponse {:?} {}", self.version(), self.status(),)?;
writeln!(f, " headers:")?;
for (key, val) in self.headers().iter() {
writeln!(f, " {:?}: {:?}", key, val)?;
}
Ok(())
}
}
impl<S> HttpMessage for ClientResponse<S> {
type Stream = S;
fn headers(&self) -> &HeaderMap {
&self.head.headers
}
fn take_payload(&mut self) -> Payload<S> {
mem::replace(&mut self.payload, Payload::None)
}
fn extensions(&self) -> Ref<'_, Extensions> {
self.head.extensions()
}
fn extensions_mut(&self) -> RefMut<'_, Extensions> {
self.head.extensions_mut()
}
}
impl<S> Stream for ClientResponse<S>
where
S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
{
type Item = Result<Bytes, PayloadError>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.project();
this.timeout.poll_timeout(cx)?;
this.payload.poll_next(cx)
}
}
#[cfg(test)]
mod tests {
use static_assertions::assert_impl_all;
use super::*;
use crate::any_body::AnyBody;
assert_impl_all!(ClientResponse: Unpin);
assert_impl_all!(ClientResponse<()>: Unpin);
assert_impl_all!(ClientResponse<AnyBody>: Unpin);
}

144
awc/src/responses/response_body.rs Normal file
View File

@ -0,0 +1,144 @@
use std::{
future::Future,
mem,
pin::Pin,
task::{Context, Poll},
};
use actix_http::{error::PayloadError, header, HttpMessage};
use bytes::Bytes;
use futures_core::Stream;
use pin_project_lite::pin_project;
use super::{read_body::ReadBody, ResponseTimeout, DEFAULT_BODY_LIMIT};
use crate::ClientResponse;
pin_project! {
/// A `Future` that reads a body stream, resolving as [`Bytes`].
///
/// # Errors
/// `Future` implementation returns error if:
/// - content type is not `application/json`;
/// - content length is greater than [limit](JsonBody::limit) (default: 2 MiB).
pub struct ResponseBody<S> {
#[pin]
body: Option<ReadBody<S>>,
length: Option<usize>,
timeout: ResponseTimeout,
err: Option<PayloadError>,
}
}
#[deprecated(since = "3.0.0", note = "Renamed to `ResponseBody`.")]
pub type MessageBody<B> = ResponseBody<B>;
impl<S> ResponseBody<S>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
{
/// Creates a body stream reader from a response by taking its payload.
pub fn new(res: &mut ClientResponse<S>) -> ResponseBody<S> {
let length = match res.headers().get(&header::CONTENT_LENGTH) {
Some(value) => {
let len = value.to_str().ok().and_then(|s| s.parse::<usize>().ok());
match len {
None => return Self::err(PayloadError::UnknownLength),
len => len,
}
}
None => None,
};
ResponseBody {
body: Some(ReadBody::new(res.take_payload(), DEFAULT_BODY_LIMIT)),
length,
timeout: mem::take(&mut res.timeout),
err: None,
}
}
/// Change max size limit of payload.
///
/// The default limit is 2 MiB.
pub fn limit(mut self, limit: usize) -> Self {
if let Some(ref mut body) = self.body {
body.limit = limit;
}
self
}
fn err(err: PayloadError) -> Self {
ResponseBody {
body: None,
length: None,
timeout: ResponseTimeout::default(),
err: Some(err),
}
}
}
impl<S> Future for ResponseBody<S>
where
S: Stream<Item = Result<Bytes, PayloadError>>,
{
type Output = Result<Bytes, PayloadError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
if let Some(err) = this.err.take() {
return Poll::Ready(Err(err));
}
if let Some(len) = this.length.take() {
let body = Option::as_ref(&this.body).unwrap();
if len > body.limit {
return Poll::Ready(Err(PayloadError::Overflow));
}
}
this.timeout.poll_timeout(cx)?;
this.body.as_pin_mut().unwrap().poll(cx)
}
}
#[cfg(test)]
mod tests {
use static_assertions::assert_impl_all;
use super::*;
use crate::{http::header, test::TestResponse};
assert_impl_all!(ResponseBody<()>: Unpin);
#[actix_rt::test]
async fn read_body() {
let mut req = TestResponse::with_header((header::CONTENT_LENGTH, "xxxx")).finish();
match req.body().await.err().unwrap() {
PayloadError::UnknownLength => {}
_ => unreachable!("error"),
}
let mut req = TestResponse::with_header((header::CONTENT_LENGTH, "10000000")).finish();
match req.body().await.err().unwrap() {
PayloadError::Overflow => {}
_ => unreachable!("error"),
}
let mut req = TestResponse::default()
.set_payload(Bytes::from_static(b"test"))
.finish();
assert_eq!(req.body().await.ok().unwrap(), Bytes::from_static(b"test"));
let mut req = TestResponse::default()
.set_payload(Bytes::from_static(b"11111111111111"))
.finish();
match req.body().limit(5).await.err().unwrap() {
PayloadError::Overflow => {}
_ => unreachable!("error"),
}
}
}

40
awc/src/sender.rs
View File

@ -8,7 +8,7 @@ use std::{
}; };
use actix_http::{ use actix_http::{
body::BodyStream, body::{BodyStream, MessageBody},
error::HttpError, error::HttpError,
header::{self, HeaderMap, HeaderName, TryIntoHeaderValue}, header::{self, HeaderMap, HeaderName, TryIntoHeaderValue},
RequestHead, RequestHeadType, RequestHead, RequestHeadType,
@ -20,12 +20,13 @@ use futures_core::Stream;
use serde::Serialize; use serde::Serialize;
#[cfg(feature = "__compress")] #[cfg(feature = "__compress")]
use actix_http::{encoding::Decoder, header::ContentEncoding, Payload, PayloadStream}; use actix_http::{encoding::Decoder, header::ContentEncoding, Payload};
use crate::{ use crate::{
any_body::AnyBody, any_body::AnyBody,
client::ClientConfig,
error::{FreezeRequestError, InvalidUrl, SendRequestError}, error::{FreezeRequestError, InvalidUrl, SendRequestError},
BoxError, ClientConfig, ClientResponse, ConnectRequest, ConnectResponse, BoxError, ClientResponse, ConnectRequest, ConnectResponse,
}; };
#[derive(Debug, From)] #[derive(Debug, From)]
@ -91,7 +92,7 @@ impl SendClientRequest {
#[cfg(feature = "__compress")] #[cfg(feature = "__compress")]
impl Future for SendClientRequest { impl Future for SendClientRequest {
type Output = Result<ClientResponse<Decoder<Payload<PayloadStream>>>, SendRequestError>; type Output = Result<ClientResponse<Decoder<Payload>>, SendRequestError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut(); let this = self.get_mut();
@ -108,12 +109,13 @@ impl Future for SendClientRequest {
res.into_client_response()._timeout(delay.take()).map_body( res.into_client_response()._timeout(delay.take()).map_body(
|head, payload| { |head, payload| {
if *response_decompress { if *response_decompress {
Payload::Stream(Decoder::from_headers(payload, &head.headers)) Payload::Stream {
payload: Decoder::from_headers(payload, &head.headers),
}
} else { } else {
Payload::Stream(Decoder::new( Payload::Stream {
payload, payload: Decoder::new(payload, ContentEncoding::Identity),
ContentEncoding::Identity, }
))
} }
}, },
) )
@ -188,15 +190,17 @@ impl RequestSender {
body: B, body: B,
) -> SendClientRequest ) -> SendClientRequest
where where
B: Into<AnyBody>, B: MessageBody + 'static,
{ {
let req = match self { let req = match self {
RequestSender::Owned(head) => { RequestSender::Owned(head) => ConnectRequest::Client(
ConnectRequest::Client(RequestHeadType::Owned(head), body.into(), addr) RequestHeadType::Owned(head),
} AnyBody::from_message_body(body).into_boxed(),
addr,
),
RequestSender::Rc(head, extra_headers) => ConnectRequest::Client( RequestSender::Rc(head, extra_headers) => ConnectRequest::Client(
RequestHeadType::Rc(head, extra_headers), RequestHeadType::Rc(head, extra_headers),
body.into(), AnyBody::from_message_body(body).into_boxed(),
addr, addr,
), ),
}; };
@ -228,9 +232,7 @@ impl RequestSender {
response_decompress, response_decompress,
timeout, timeout,
config, config,
AnyBody::Bytes { AnyBody::from_message_body(body.into_bytes()),
body: Bytes::from(body),
},
) )
} }
@ -259,9 +261,7 @@ impl RequestSender {
response_decompress, response_decompress,
timeout, timeout,
config, config,
AnyBody::Bytes { AnyBody::from_message_body(body.into_bytes()),
body: Bytes::from(body),
},
) )
} }

5
awc/src/test.rs
View File

@ -65,7 +65,7 @@ impl TestResponse {
/// Set response's payload /// Set response's payload
pub fn set_payload<B: Into<Bytes>>(mut self, data: B) -> Self { pub fn set_payload<B: Into<Bytes>>(mut self, data: B) -> Self {
let mut payload = h1::Payload::empty(); let (_, mut payload) = h1::Payload::create(true);
payload.unread_data(data.into()); payload.unread_data(data.into());
self.payload = Some(payload.into()); self.payload = Some(payload.into());
self self
@ -90,7 +90,8 @@ impl TestResponse {
if let Some(pl) = self.payload { if let Some(pl) = self.payload {
ClientResponse::new(head, pl) ClientResponse::new(head, pl)
} else { } else {
ClientResponse::new(head, h1::Payload::empty().into()) let (_, payload) = h1::Payload::create(true);
ClientResponse::new(head, payload.into())
} }
} }
} }

6
awc/src/ws.rs
View File

@ -31,19 +31,19 @@ use std::{convert::TryFrom, fmt, net::SocketAddr, str};
use actix_codec::Framed; use actix_codec::Framed;
use actix_http::{ws, Payload, RequestHead}; use actix_http::{ws, Payload, RequestHead};
use actix_rt::time::timeout; use actix_rt::time::timeout;
use actix_service::Service; use actix_service::Service as _;
pub use actix_http::ws::{CloseCode, CloseReason, Codec, Frame, Message}; pub use actix_http::ws::{CloseCode, CloseReason, Codec, Frame, Message};
use crate::{ use crate::{
client::ClientConfig,
connect::{BoxedSocket, ConnectRequest}, connect::{BoxedSocket, ConnectRequest},
error::{HttpError, InvalidUrl, SendRequestError, WsClientError}, error::{HttpError, InvalidUrl, SendRequestError, WsClientError},
http::{ http::{
header::{self, HeaderName, HeaderValue, TryIntoHeaderValue, AUTHORIZATION}, header::{self, HeaderName, HeaderValue, TryIntoHeaderValue, AUTHORIZATION},
ConnectionType, Method, StatusCode, Uri, Version, ConnectionType, Method, StatusCode, Uri, Version,
}, },
response::ClientResponse, ClientResponse,
ClientConfig,
}; };
#[cfg(feature = "cookies")] #[cfg(feature = "cookies")]

2
src/dev.rs
View File

@ -14,7 +14,7 @@ pub use crate::types::form::UrlEncoded;
pub use crate::types::json::JsonBody; pub use crate::types::json::JsonBody;
pub use crate::types::readlines::Readlines; pub use crate::types::readlines::Readlines;
pub use actix_http::{Extensions, Payload, PayloadStream, RequestHead, Response, ResponseHead}; pub use actix_http::{Extensions, Payload, RequestHead, Response, ResponseHead};
pub use actix_router::{Path, ResourceDef, ResourcePath, Url}; pub use actix_router::{Path, ResourceDef, ResourcePath, Url};
pub use actix_server::{Server, ServerHandle}; pub use actix_server::{Server, ServerHandle};
pub use actix_service::{ pub use actix_service::{

2
src/guard.rs
View File

@ -270,13 +270,11 @@ impl Guard for HeaderGuard {
/// ``` /// ```
/// use actix_web::{web, guard::Host, App, HttpResponse}; /// use actix_web::{web, guard::Host, App, HttpResponse};
/// ///
/// fn main() {
/// App::new().service( /// App::new().service(
/// web::resource("/index.html") /// web::resource("/index.html")
/// .guard(Host("www.rust-lang.org")) /// .guard(Host("www.rust-lang.org"))
/// .to(|| HttpResponse::MethodNotAllowed()) /// .to(|| HttpResponse::MethodNotAllowed())
/// ); /// );
/// }
/// ``` /// ```
pub fn Host<H: AsRef<str>>(host: H) -> HostGuard { pub fn Host<H: AsRef<str>>(host: H) -> HostGuard {
HostGuard(host.as_ref().to_string(), None) HostGuard(host.as_ref().to_string(), None)

1
src/http/mod.rs
View File

@ -2,4 +2,5 @@
pub mod header; pub mod header;
// TODO: figure out how best to expose http::Error vs actix_http::Error
pub use actix_http::{uri, ConnectionType, Error, Method, StatusCode, Uri, Version}; pub use actix_http::{uri, ConnectionType, Error, Method, StatusCode, Uri, Version};

32
src/response/builder.rs
View File

@ -429,9 +429,12 @@ mod tests {
use actix_http::body; use actix_http::body;
use super::*; use super::*;
use crate::http::{ use crate::{
http::{
header::{self, HeaderValue, CONTENT_TYPE}, header::{self, HeaderValue, CONTENT_TYPE},
StatusCode, StatusCode,
},
test::assert_body_eq,
}; };
#[test] #[test]
@ -472,32 +475,23 @@ mod tests {
#[actix_rt::test] #[actix_rt::test]
async fn test_json() { async fn test_json() {
let resp = HttpResponse::Ok().json(vec!["v1", "v2", "v3"]); let res = HttpResponse::Ok().json(vec!["v1", "v2", "v3"]);
let ct = resp.headers().get(CONTENT_TYPE).unwrap(); let ct = res.headers().get(CONTENT_TYPE).unwrap();
assert_eq!(ct, HeaderValue::from_static("application/json")); assert_eq!(ct, HeaderValue::from_static("application/json"));
assert_eq!( assert_body_eq!(res, br#"["v1","v2","v3"]"#);
body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
br#"["v1","v2","v3"]"#
);
let resp = HttpResponse::Ok().json(&["v1", "v2", "v3"]); let res = HttpResponse::Ok().json(&["v1", "v2", "v3"]);
let ct = resp.headers().get(CONTENT_TYPE).unwrap(); let ct = res.headers().get(CONTENT_TYPE).unwrap();
assert_eq!(ct, HeaderValue::from_static("application/json")); assert_eq!(ct, HeaderValue::from_static("application/json"));
assert_eq!( assert_body_eq!(res, br#"["v1","v2","v3"]"#);
body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
br#"["v1","v2","v3"]"#
);
// content type override // content type override
let resp = HttpResponse::Ok() let res = HttpResponse::Ok()
.insert_header((CONTENT_TYPE, "text/json")) .insert_header((CONTENT_TYPE, "text/json"))
.json(&vec!["v1", "v2", "v3"]); .json(&vec!["v1", "v2", "v3"]);
let ct = resp.headers().get(CONTENT_TYPE).unwrap(); let ct = res.headers().get(CONTENT_TYPE).unwrap();
assert_eq!(ct, HeaderValue::from_static("text/json")); assert_eq!(ct, HeaderValue::from_static("text/json"));
assert_eq!( assert_body_eq!(res, br#"["v1","v2","v3"]"#);
body::to_bytes(resp.into_body()).await.unwrap().as_ref(),
br#"["v1","v2","v3"]"#
);
} }
#[actix_rt::test] #[actix_rt::test]

4
src/service.rs
View File

@ -7,7 +7,7 @@ use std::{
use actix_http::{ use actix_http::{
body::{BoxBody, EitherBody, MessageBody}, body::{BoxBody, EitherBody, MessageBody},
header::HeaderMap, header::HeaderMap,
Extensions, HttpMessage, Method, Payload, PayloadStream, RequestHead, Response, BoxedPayloadStream, Extensions, HttpMessage, Method, Payload, RequestHead, Response,
ResponseHead, StatusCode, Uri, Version, ResponseHead, StatusCode, Uri, Version,
}; };
use actix_router::{IntoPatterns, Path, Patterns, Resource, ResourceDef, Url}; use actix_router::{IntoPatterns, Path, Patterns, Resource, ResourceDef, Url};
@ -293,7 +293,7 @@ impl Resource<Url> for ServiceRequest {
} }
impl HttpMessage for ServiceRequest { impl HttpMessage for ServiceRequest {
type Stream = PayloadStream; type Stream = BoxedPayloadStream;
#[inline] #[inline]
/// Returns Request's headers. /// Returns Request's headers.

21
src/test/test_request.rs
View File

@ -174,25 +174,28 @@ impl TestRequest {
} }
/// Set request payload. /// Set request payload.
pub fn set_payload<B: Into<Bytes>>(mut self, data: B) -> Self { pub fn set_payload(mut self, data: impl Into<Bytes>) -> Self {
self.req.set_payload(data); self.req.set_payload(data);
self self
} }
/// Serialize `data` to a URL encoded form and set it as the request payload. The `Content-Type` /// Serialize `data` to a URL encoded form and set it as the request payload.
/// header is set to `application/x-www-form-urlencoded`. ///
pub fn set_form<T: Serialize>(mut self, data: &T) -> Self { /// The `Content-Type` header is set to `application/x-www-form-urlencoded`.
let bytes = serde_urlencoded::to_string(data) pub fn set_form(mut self, data: impl Serialize) -> Self {
let bytes = serde_urlencoded::to_string(&data)
.expect("Failed to serialize test data as a urlencoded form"); .expect("Failed to serialize test data as a urlencoded form");
self.req.set_payload(bytes); self.req.set_payload(bytes);
self.req.insert_header(ContentType::form_url_encoded()); self.req.insert_header(ContentType::form_url_encoded());
self self
} }
/// Serialize `data` to JSON and set it as the request payload. The `Content-Type` header is /// Serialize `data` to JSON and set it as the request payload.
/// set to `application/json`. ///
pub fn set_json<T: Serialize>(mut self, data: &T) -> Self { /// The `Content-Type` header is set to `application/json`.
let bytes = serde_json::to_string(data).expect("Failed to serialize test data to json"); pub fn set_json(mut self, data: impl Serialize) -> Self {
let bytes =
serde_json::to_string(&data).expect("Failed to serialize test data to json");
self.req.set_payload(bytes); self.req.set_payload(bytes);
self.req.insert_header(ContentType::json()); self.req.insert_header(ContentType::json());
self self