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

Replaced impl Future with concrete Future implementation

This commit is contained in:
Dmitry Pypin 2019-09-09 15:44:50 -07:00
parent 3ea7be5221
commit 2589298c27
1 changed files with 162 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

305
awc/src/request.rs
View File

@ -1,16 +1,15 @@
use std::fmt::Write as FmtWrite; use std::fmt::Write as FmtWrite;
use std::io::Write; use std::io::Write;
use std::rc::Rc; use std::rc::Rc;
use std::time::Duration; use std::time::{Duration, Instant};
use std::{fmt, net}; use std::{fmt, net};
use bytes::{BufMut, Bytes, BytesMut}; use bytes::{BufMut, Bytes, BytesMut};
use futures::future::{err, Either}; use futures::{Async, Future, Poll, Stream, try_ready};
use futures::{Future, Stream};
use percent_encoding::percent_encode; use percent_encoding::percent_encode;
use serde::Serialize; use serde::Serialize;
use serde_json; use serde_json;
use tokio_timer::Timeout; use tokio_timer::Delay;
use derive_more::From; use derive_more::From;
use actix_http::body::{Body, BodyStream}; use actix_http::body::{Body, BodyStream};
@ -21,9 +20,9 @@ use actix_http::http::{
uri, ConnectionType, Error as HttpError, HeaderMap, HeaderName, HeaderValue, uri, ConnectionType, Error as HttpError, HeaderMap, HeaderName, HeaderValue,
HttpTryFrom, Method, Uri, Version, HttpTryFrom, Method, Uri, Version,
}; };
use actix_http::{Error, Payload, RequestHead}; use actix_http::{Error, Payload, PayloadStream, RequestHead};
use crate::error::{InvalidUrl, PayloadError, SendRequestError, FreezeRequestError}; use crate::error::{InvalidUrl, SendRequestError, FreezeRequestError};
use crate::response::ClientResponse; use crate::response::ClientResponse;
use crate::ClientConfig; use crate::ClientConfig;
@ -397,35 +396,32 @@ impl ClientRequest {
pub fn send_body<B>( pub fn send_body<B>(
self, self,
body: B, body: B,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
B: Into<Body>, B: Into<Body>,
{ {
let slf = match self.prep_for_sending() { let slf = match self.prep_for_sending() {
Ok(slf) => slf, Ok(slf) => slf,
Err(e) => return Either::A(err(e.into())), Err(e) => return e.into(),
}; };
Either::B(RequestSender::Owned(slf.head).send_body(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), body)) RequestSender::Owned(slf.head)
.send_body(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), body)
} }
/// Set a JSON body and generate `ClientRequest` /// Set a JSON body and generate `ClientRequest`
pub fn send_json<T: Serialize>( pub fn send_json<T: Serialize>(
self, self,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
let slf = match self.prep_for_sending() { let slf = match self.prep_for_sending() {
Ok(slf) => slf, Ok(slf) => slf,
Err(e) => return Either::A(err(e.into())), Err(e) => return e.into(),
}; };
Either::B(RequestSender::Owned(slf.head).send_json(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), value)) RequestSender::Owned(slf.head)
.send_json(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), value)
} }
/// Set a urlencoded body and generate `ClientRequest` /// Set a urlencoded body and generate `ClientRequest`
@ -434,51 +430,47 @@ impl ClientRequest {
pub fn send_form<T: Serialize>( pub fn send_form<T: Serialize>(
self, self,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
let slf = match self.prep_for_sending() { let slf = match self.prep_for_sending() {
Ok(slf) => slf, Ok(slf) => slf,
Err(e) => return Either::A(err(e.into())), Err(e) => return e.into(),
}; };
Either::B(RequestSender::Owned(slf.head).send_form(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), value)) RequestSender::Owned(slf.head)
.send_form(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), value)
} }
/// Set an streaming body and generate `ClientRequest`. /// Set an streaming body and generate `ClientRequest`.
pub fn send_stream<S, E>( pub fn send_stream<S, E>(
self, self,
stream: S, stream: S,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
S: Stream<Item = Bytes, Error = E> + 'static, S: Stream<Item = Bytes, Error = E> + 'static,
E: Into<Error> + 'static, E: Into<Error> + 'static,
{ {
let slf = match self.prep_for_sending() { let slf = match self.prep_for_sending() {
Ok(slf) => slf, Ok(slf) => slf,
Err(e) => return Either::A(err(e.into())), Err(e) => return e.into(),
}; };
Either::B(RequestSender::Owned(slf.head).send_stream(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), stream)) RequestSender::Owned(slf.head)
.send_stream(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), stream)
} }
/// Set an empty body and generate `ClientRequest`. /// Set an empty body and generate `ClientRequest`.
pub fn send( pub fn send(
self, self,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
let slf = match self.prep_for_sending() { let slf = match self.prep_for_sending() {
Ok(slf) => slf, Ok(slf) => slf,
Err(e) => return Either::A(err(e.into())), Err(e) => return e.into(),
}; };
Either::B(RequestSender::Owned(slf.head).send(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref())) RequestSender::Owned(slf.head)
.send(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref())
} }
fn prep_for_sending(mut self) -> Result<Self, PrepForSendingError> { fn prep_for_sending(mut self) -> Result<Self, PrepForSendingError> {
@ -592,10 +584,7 @@ impl FrozenClientRequest {
pub fn send_body<B>( pub fn send_body<B>(
&self, &self,
body: B, body: B,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
B: Into<Body>, B: Into<Body>,
{ {
@ -607,10 +596,8 @@ impl FrozenClientRequest {
pub fn send_json<T: Serialize>( pub fn send_json<T: Serialize>(
&self, &self,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
RequestSender::Rc(self.head.clone(), None) RequestSender::Rc(self.head.clone(), None)
.send_json(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), value) .send_json(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), value)
} }
@ -619,10 +606,8 @@ impl FrozenClientRequest {
pub fn send_form<T: Serialize>( pub fn send_form<T: Serialize>(
&self, &self,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
RequestSender::Rc(self.head.clone(), None) RequestSender::Rc(self.head.clone(), None)
.send_form(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), value) .send_form(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), value)
} }
@ -631,10 +616,7 @@ impl FrozenClientRequest {
pub fn send_stream<S, E>( pub fn send_stream<S, E>(
&self, &self,
stream: S, stream: S,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
S: Stream<Item = Bytes, Error = E> + 'static, S: Stream<Item = Bytes, Error = E> + 'static,
E: Into<Error> + 'static, E: Into<Error> + 'static,
@ -646,10 +628,8 @@ impl FrozenClientRequest {
/// Send an empty body. /// Send an empty body.
pub fn send( pub fn send(
&self, &self,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
RequestSender::Rc(self.head.clone(), None) RequestSender::Rc(self.head.clone(), None)
.send(self.addr, self.response_decompress, self.timeout, self.config.as_ref()) .send(self.addr, self.response_decompress, self.timeout, self.config.as_ref())
} }
@ -704,86 +684,74 @@ impl FrozenSendBuilder {
pub fn send_body<B>( pub fn send_body<B>(
self, self,
body: B, body: B,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
B: Into<Body>, B: Into<Body>,
{ {
if let Some(e) = self.err { if let Some(e) = self.err {
return Either::A(err(e.into())) return e.into()
} }
Either::B(RequestSender::Rc(self.req.head, Some(self.extra_headers)) RequestSender::Rc(self.req.head, Some(self.extra_headers))
.send_body(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), body)) .send_body(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), body)
} }
/// Complete request construction and send a json body. /// Complete request construction and send a json body.
pub fn send_json<T: Serialize>( pub fn send_json<T: Serialize>(
self, self,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
if let Some(e) = self.err { if let Some(e) = self.err {
return Either::A(err(e.into())) return e.into()
} }
Either::B(RequestSender::Rc(self.req.head, Some(self.extra_headers)) RequestSender::Rc(self.req.head, Some(self.extra_headers))
.send_json(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), value)) .send_json(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), value)
} }
/// Complete request construction and send an urlencoded body. /// Complete request construction and send an urlencoded body.
pub fn send_form<T: Serialize>( pub fn send_form<T: Serialize>(
self, self,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
if let Some(e) = self.err { if let Some(e) = self.err {
return Either::A(err(e.into())) return e.into()
} }
Either::B(RequestSender::Rc(self.req.head, Some(self.extra_headers)) RequestSender::Rc(self.req.head, Some(self.extra_headers))
.send_form(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), value)) .send_form(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), value)
} }
/// Complete request construction and send a streaming body. /// Complete request construction and send a streaming body.
pub fn send_stream<S, E>( pub fn send_stream<S, E>(
self, self,
stream: S, stream: S,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
S: Stream<Item = Bytes, Error = E> + 'static, S: Stream<Item = Bytes, Error = E> + 'static,
E: Into<Error> + 'static, E: Into<Error> + 'static,
{ {
if let Some(e) = self.err { if let Some(e) = self.err {
return Either::A(err(e.into())) return e.into()
} }
Either::B(RequestSender::Rc(self.req.head, Some(self.extra_headers)) RequestSender::Rc(self.req.head, Some(self.extra_headers))
.send_stream(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), stream)) .send_stream(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref(), stream)
} }
/// Complete request construction and send an empty body. /// Complete request construction and send an empty body.
pub fn send( pub fn send(
self, self,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
if let Some(e) = self.err { if let Some(e) = self.err {
return Either::A(err(e.into())) return e.into()
} }
Either::B(RequestSender::Rc(self.req.head, Some(self.extra_headers)) RequestSender::Rc(self.req.head, Some(self.extra_headers))
.send(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref())) .send(self.req.addr, self.req.response_decompress, self.req.timeout, self.req.config.as_ref())
} }
} }
@ -811,6 +779,91 @@ impl Into<SendRequestError> for PrepForSendingError {
} }
} }
pub enum Send
{
Fut(Box<dyn Future<Item = ClientResponse, Error = SendRequestError>>, Option<Delay>, bool),
Err(Option<SendRequestError>),
}
impl Send
{
pub fn new(
send: Box<dyn Future<Item = ClientResponse, Error = SendRequestError>>,
response_decompress: bool,
timeout: Option<Duration>,
) -> Send
{
let delay = timeout.map(|t| Delay::new(Instant::now() + t));
Send::Fut(send, delay, response_decompress)
}
}
impl Future for Send
{
type Item = ClientResponse<Decoder<Payload<PayloadStream>>>;
type Error = SendRequestError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self {
Send::Fut(send, delay, response_decompress) => {
if delay.is_some() {
match delay.poll() {
Ok(Async::NotReady) => (),
_ => return Err(SendRequestError::Timeout),
}
}
let res = try_ready!(send.poll())
.map_body(|head, payload| {
if *response_decompress {
Payload::Stream(Decoder::from_headers(payload, &head.headers))
} else {
Payload::Stream(Decoder::new(payload, ContentEncoding::Identity))
}
});
Ok(Async::Ready(res))
},
Send::Err(ref mut e) => {
match e.take() {
Some(e) => Err(e.into()),
None => panic!("Attempting to call completed future"),
}
}
}
}
}
impl From<SendRequestError> for Send
{
fn from(e: SendRequestError) -> Self {
Send::Err(Some(e))
}
}
impl From<Error> for Send
{
fn from(e: Error) -> Self {
Send::Err(Some(e.into()))
}
}
impl From<HttpError> for Send
{
fn from(e: HttpError) -> Self {
Send::Err(Some(e.into()))
}
}
impl From<PrepForSendingError> for Send
{
fn from(e: PrepForSendingError) -> Self {
Send::Err(Some(e.into()))
}
}
#[derive(Debug)]
enum RequestSender { enum RequestSender {
Owned(RequestHead), Owned(RequestHead),
Rc(Rc<RequestHead>, Option<HeaderMap>), Rc(Rc<RequestHead>, Option<HeaderMap>),
@ -824,10 +877,7 @@ impl RequestSender {
timeout: Option<Duration>, timeout: Option<Duration>,
config: &ClientConfig, config: &ClientConfig,
body: B, body: B,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>,
Error = SendRequestError,
>
where where
B: Into<Body>, B: Into<Body>,
{ {
@ -838,29 +888,7 @@ impl RequestSender {
RequestSender::Rc(head, extra_headers) => connector.send_request_extra(head, extra_headers, body.into(), addr), RequestSender::Rc(head, extra_headers) => connector.send_request_extra(head, extra_headers, body.into(), addr),
}; };
let fut = fut Send::new(fut, response_decompress, timeout.or_else(|| config.timeout.clone()))
.map(move |res| {
res.map_body(|head, payload| {
if response_decompress {
Payload::Stream(Decoder::from_headers(payload, &head.headers))
} else {
Payload::Stream(Decoder::new(payload, ContentEncoding::Identity))
}
})
});
// set request timeout
if let Some(timeout) = timeout.or_else(|| config.timeout.clone()) {
Either::A(Timeout::new(fut, timeout).map_err(|e| {
if let Some(e) = e.into_inner() {
e
} else {
SendRequestError::Timeout
}
}))
} else {
Either::B(fut)
}
} }
pub fn send_json<T: Serialize>( pub fn send_json<T: Serialize>(
@ -870,20 +898,18 @@ impl RequestSender {
timeout: Option<Duration>, timeout: Option<Duration>,
config: &ClientConfig, config: &ClientConfig,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
let body = match serde_json::to_string(value) { let body = match serde_json::to_string(value) {
Ok(body) => body, Ok(body) => body,
Err(e) => return Either::A(err(Error::from(e).into())), Err(e) => return Error::from(e).into(),
}; };
if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/json") { if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/json") {
return Either::A(err(e.into())); return e.into();
} }
Either::B(self.send_body(addr, response_decompress, timeout, config, Body::Bytes(Bytes::from(body)))) self.send_body(addr, response_decompress, timeout, config, Body::Bytes(Bytes::from(body)))
} }
pub fn send_form<T: Serialize>( pub fn send_form<T: Serialize>(
@ -893,21 +919,19 @@ impl RequestSender {
timeout: Option<Duration>, timeout: Option<Duration>,
config: &ClientConfig, config: &ClientConfig,
value: &T, value: &T,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
let body = match serde_urlencoded::to_string(value) { let body = match serde_urlencoded::to_string(value) {
Ok(body) => body, Ok(body) => body,
Err(e) => return Either::A(err(Error::from(e).into())), Err(e) => return Error::from(e).into(),
}; };
// set content-type // set content-type
if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/x-www-form-urlencoded") { if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/x-www-form-urlencoded") {
return Either::A(err(e.into())); return e.into();
} }
Either::B(self.send_body(addr, response_decompress, timeout, config, Body::Bytes(Bytes::from(body)))) self.send_body(addr, response_decompress, timeout, config, Body::Bytes(Bytes::from(body)))
} }
pub fn send_stream<S, E>( pub fn send_stream<S, E>(
@ -917,13 +941,10 @@ impl RequestSender {
timeout: Option<Duration>, timeout: Option<Duration>,
config: &ClientConfig, config: &ClientConfig,
stream: S, stream: S,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, where
Error = SendRequestError, S: Stream<Item = Bytes, Error = E> + 'static,
> E: Into<Error> + 'static,
where
S: Stream<Item = Bytes, Error = E> + 'static,
E: Into<Error> + 'static,
{ {
self.send_body(addr, response_decompress, timeout, config, Body::from_message(BodyStream::new(stream))) self.send_body(addr, response_decompress, timeout, config, Body::from_message(BodyStream::new(stream)))
} }
@ -934,10 +955,8 @@ impl RequestSender {
response_decompress: bool, response_decompress: bool,
timeout: Option<Duration>, timeout: Option<Duration>,
config: &ClientConfig, config: &ClientConfig,
) -> impl Future< ) -> Send
Item = ClientResponse<impl Stream<Item = Bytes, Error = PayloadError>>, {
Error = SendRequestError,
> {
self.send_body(addr, response_decompress, timeout, config, Body::Empty) self.send_body(addr, response_decompress, timeout, config, Body::Empty)
} }