mirror of
https://github.com/fafhrd91/actix-web
synced 2025-07-04 01:51:30 +02:00
Compare commits
4 Commits
http-v0.2.
...
multipart-
Author | SHA1 | Date | |
---|---|---|---|
e35d930ef9 | |||
60b7aebd0a | |||
45d2fd4299 | |||
71f8577713 |
@ -66,7 +66,7 @@ hashbrown = "0.5.0"
|
||||
h2 = "0.1.16"
|
||||
http = "0.1.17"
|
||||
httparse = "1.3"
|
||||
indexmap = "1.0"
|
||||
indexmap = "1.2"
|
||||
lazy_static = "1.0"
|
||||
language-tags = "0.2"
|
||||
log = "0.4"
|
||||
|
@ -8,10 +8,10 @@ use futures::{Async, Future, Poll, Sink, Stream};
|
||||
|
||||
use crate::error::PayloadError;
|
||||
use crate::h1;
|
||||
use crate::header::HeaderMap;
|
||||
use crate::http::header::{IntoHeaderValue, HOST};
|
||||
use crate::message::{RequestHeadType, ResponseHead};
|
||||
use crate::payload::{Payload, PayloadStream};
|
||||
use crate::header::HeaderMap;
|
||||
|
||||
use super::connection::{ConnectionLifetime, ConnectionType, IoConnection};
|
||||
use super::error::{ConnectError, SendRequestError};
|
||||
@ -30,7 +30,9 @@ where
|
||||
B: MessageBody,
|
||||
{
|
||||
// set request host header
|
||||
if !head.as_ref().headers.contains_key(HOST) && !head.extra_headers().iter().any(|h| h.contains_key(HOST)) {
|
||||
if !head.as_ref().headers.contains_key(HOST)
|
||||
&& !head.extra_headers().iter().any(|h| h.contains_key(HOST))
|
||||
{
|
||||
if let Some(host) = head.as_ref().uri.host() {
|
||||
let mut wrt = BytesMut::with_capacity(host.len() + 5).writer();
|
||||
|
||||
@ -40,20 +42,16 @@ where
|
||||
};
|
||||
|
||||
match wrt.get_mut().take().freeze().try_into() {
|
||||
Ok(value) => {
|
||||
match head {
|
||||
Ok(value) => match head {
|
||||
RequestHeadType::Owned(ref mut head) => {
|
||||
head.headers.insert(HOST, value)
|
||||
},
|
||||
}
|
||||
RequestHeadType::Rc(_, ref mut extra_headers) => {
|
||||
let headers = extra_headers.get_or_insert(HeaderMap::new());
|
||||
headers.insert(HOST, value)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
log::error!("Can not set HOST header {}", e)
|
||||
}
|
||||
Err(e) => log::error!("Can not set HOST header {}", e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -9,9 +9,9 @@ use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, TRANSFER_ENCODING};
|
||||
use http::{request::Request, HttpTryFrom, Method, Version};
|
||||
|
||||
use crate::body::{BodySize, MessageBody};
|
||||
use crate::header::HeaderMap;
|
||||
use crate::message::{RequestHeadType, ResponseHead};
|
||||
use crate::payload::Payload;
|
||||
use crate::header::HeaderMap;
|
||||
|
||||
use super::connection::{ConnectionType, IoConnection};
|
||||
use super::error::SendRequestError;
|
||||
@ -69,15 +69,21 @@ where
|
||||
|
||||
// Extracting extra headers from RequestHeadType. HeaderMap::new() does not allocate.
|
||||
let (head, extra_headers) = match head {
|
||||
RequestHeadType::Owned(head) => (RequestHeadType::Owned(head), HeaderMap::new()),
|
||||
RequestHeadType::Rc(head, extra_headers) => (RequestHeadType::Rc(head, None), extra_headers.unwrap_or(HeaderMap::new())),
|
||||
RequestHeadType::Owned(head) => {
|
||||
(RequestHeadType::Owned(head), HeaderMap::new())
|
||||
}
|
||||
RequestHeadType::Rc(head, extra_headers) => (
|
||||
RequestHeadType::Rc(head, None),
|
||||
extra_headers.unwrap_or_else(HeaderMap::new),
|
||||
),
|
||||
};
|
||||
|
||||
// merging headers from head and extra headers.
|
||||
let headers = head.as_ref().headers.iter()
|
||||
.filter(|(name, _)| {
|
||||
!extra_headers.contains_key(*name)
|
||||
})
|
||||
let headers = head
|
||||
.as_ref()
|
||||
.headers
|
||||
.iter()
|
||||
.filter(|(name, _)| !extra_headers.contains_key(*name))
|
||||
.chain(extra_headers.iter());
|
||||
|
||||
// copy headers
|
||||
|
@ -10,7 +10,7 @@ mod pool;
|
||||
|
||||
pub use self::connection::Connection;
|
||||
pub use self::connector::Connector;
|
||||
pub use self::error::{ConnectError, InvalidUrl, SendRequestError, FreezeRequestError};
|
||||
pub use self::error::{ConnectError, FreezeRequestError, InvalidUrl, SendRequestError};
|
||||
pub use self::pool::Protocol;
|
||||
|
||||
#[derive(Clone)]
|
||||
|
@ -326,7 +326,7 @@ impl<Io> Inner<Io> {
|
||||
|
||||
fn release_waiter(&mut self, key: &Key, token: usize) {
|
||||
self.waiters.remove(token);
|
||||
self.waiters_queue.remove(&(key.clone(), token));
|
||||
let _ = self.waiters_queue.shift_remove(&(key.clone(), token));
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -16,9 +16,11 @@ use super::{Message, MessageType};
|
||||
use crate::body::BodySize;
|
||||
use crate::config::ServiceConfig;
|
||||
use crate::error::{ParseError, PayloadError};
|
||||
use crate::helpers;
|
||||
use crate::message::{ConnectionType, Head, MessagePool, RequestHead, RequestHeadType, ResponseHead};
|
||||
use crate::header::HeaderMap;
|
||||
use crate::helpers;
|
||||
use crate::message::{
|
||||
ConnectionType, Head, MessagePool, RequestHead, RequestHeadType, ResponseHead,
|
||||
};
|
||||
|
||||
bitflags! {
|
||||
struct Flags: u8 {
|
||||
@ -197,7 +199,9 @@ impl Encoder for ClientCodec {
|
||||
Message::Item((mut head, length)) => {
|
||||
let inner = &mut self.inner;
|
||||
inner.version = head.as_ref().version;
|
||||
inner.flags.set(Flags::HEAD, head.as_ref().method == Method::HEAD);
|
||||
inner
|
||||
.flags
|
||||
.set(Flags::HEAD, head.as_ref().method == Method::HEAD);
|
||||
|
||||
// connection status
|
||||
inner.ctype = match head.as_ref().connection_type() {
|
||||
|
@ -2,9 +2,9 @@
|
||||
use std::fmt::Write as FmtWrite;
|
||||
use std::io::Write;
|
||||
use std::marker::PhantomData;
|
||||
use std::rc::Rc;
|
||||
use std::str::FromStr;
|
||||
use std::{cmp, fmt, io, mem};
|
||||
use std::rc::Rc;
|
||||
|
||||
use bytes::{BufMut, Bytes, BytesMut};
|
||||
|
||||
@ -16,7 +16,7 @@ use crate::http::header::{
|
||||
HeaderValue, ACCEPT_ENCODING, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING,
|
||||
};
|
||||
use crate::http::{HeaderMap, Method, StatusCode, Version};
|
||||
use crate::message::{ConnectionType, Head, RequestHead, ResponseHead, RequestHeadType};
|
||||
use crate::message::{ConnectionType, Head, RequestHead, RequestHeadType, ResponseHead};
|
||||
use crate::request::Request;
|
||||
use crate::response::Response;
|
||||
|
||||
@ -134,10 +134,11 @@ pub(crate) trait MessageType: Sized {
|
||||
// merging headers from head and extra headers. HeaderMap::new() does not allocate.
|
||||
let empty_headers = HeaderMap::new();
|
||||
let extra_headers = self.extra_headers().unwrap_or(&empty_headers);
|
||||
let headers = self.headers().inner.iter()
|
||||
.filter(|(name, _)| {
|
||||
!extra_headers.contains_key(*name)
|
||||
})
|
||||
let headers = self
|
||||
.headers()
|
||||
.inner
|
||||
.iter()
|
||||
.filter(|(name, _)| !extra_headers.contains_key(*name))
|
||||
.chain(extra_headers.inner.iter());
|
||||
|
||||
// write headers
|
||||
@ -604,10 +605,16 @@ mod tests {
|
||||
let mut bytes = BytesMut::with_capacity(2048);
|
||||
|
||||
let mut head = RequestHead::default();
|
||||
head.headers.insert(AUTHORIZATION, HeaderValue::from_static("some authorization"));
|
||||
head.headers.insert(
|
||||
AUTHORIZATION,
|
||||
HeaderValue::from_static("some authorization"),
|
||||
);
|
||||
|
||||
let mut extra_headers = HeaderMap::new();
|
||||
extra_headers.insert(AUTHORIZATION,HeaderValue::from_static("another authorization"));
|
||||
extra_headers.insert(
|
||||
AUTHORIZATION,
|
||||
HeaderValue::from_static("another authorization"),
|
||||
);
|
||||
extra_headers.insert(DATE, HeaderValue::from_static("date"));
|
||||
|
||||
let mut head = RequestHeadType::Rc(Rc::new(head), Some(extra_headers));
|
||||
|
@ -1,5 +1,6 @@
|
||||
# Changes
|
||||
## [0.1.4] - 2019-xx-xx
|
||||
|
||||
## [0.1.4] - 2019-09-12
|
||||
|
||||
* Multipart handling now parses requests which do not end in CRLF #1038
|
||||
|
||||
|
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "actix-multipart"
|
||||
version = "0.1.3"
|
||||
version = "0.1.4"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Multipart support for actix web framework."
|
||||
readme = "README.md"
|
||||
|
@ -178,13 +178,15 @@ impl InnerMultipart {
|
||||
Some(chunk) => {
|
||||
if chunk.len() < boundary.len() + 4
|
||||
|| &chunk[..2] != b"--"
|
||||
|| &chunk[2..boundary.len() + 2] != boundary.as_bytes() {
|
||||
|| &chunk[2..boundary.len() + 2] != boundary.as_bytes()
|
||||
{
|
||||
Err(MultipartError::Boundary)
|
||||
} else if &chunk[boundary.len() + 2..] == b"\r\n" {
|
||||
Ok(Some(false))
|
||||
} else if &chunk[boundary.len() + 2..boundary.len() + 4] == b"--"
|
||||
&& (chunk.len() == boundary.len() + 4
|
||||
|| &chunk[boundary.len() + 4..] == b"\r\n") {
|
||||
|| &chunk[boundary.len() + 4..] == b"\r\n")
|
||||
{
|
||||
Ok(Some(true))
|
||||
} else {
|
||||
Err(MultipartError::Boundary)
|
||||
@ -781,8 +783,10 @@ impl PayloadBuffer {
|
||||
/// Read bytes until new line delimiter or eof
|
||||
pub fn readline_or_eof(&mut self) -> Result<Option<Bytes>, MultipartError> {
|
||||
match self.readline() {
|
||||
Err(MultipartError::Incomplete) if self.eof => Ok(Some(self.buf.take().freeze())),
|
||||
line => line
|
||||
Err(MultipartError::Incomplete) if self.eof => {
|
||||
Ok(Some(self.buf.take().freeze()))
|
||||
}
|
||||
line => line,
|
||||
}
|
||||
}
|
||||
|
||||
@ -866,7 +870,7 @@ mod tests {
|
||||
--abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
|
||||
Content-Type: text/plain; charset=utf-8\r\nContent-Length: 4\r\n\r\n\
|
||||
data\r\n\
|
||||
--abbc761f78ff4d7cb7573b5a23f96ef0--\r\n"
|
||||
--abbc761f78ff4d7cb7573b5a23f96ef0--\r\n",
|
||||
);
|
||||
let mut headers = HeaderMap::new();
|
||||
headers.insert(
|
||||
|
@ -1,14 +1,18 @@
|
||||
# Changes
|
||||
|
||||
##
|
||||
## [0.2.6] - 2019-09-12
|
||||
|
||||
### Added
|
||||
|
||||
* Export frozen request related types.
|
||||
|
||||
|
||||
## [0.2.5] - 2019-09-11
|
||||
|
||||
### Added
|
||||
|
||||
* Add `FrozenClientRequest` to support retries for sending HTTP requests
|
||||
|
||||
|
||||
## [0.2.5] - 2019-09-06
|
||||
|
||||
### Changed
|
||||
|
||||
* Ensure that the `Host` header is set when initiating a WebSocket client connection.
|
||||
|
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "awc"
|
||||
version = "0.2.4"
|
||||
version = "0.2.6"
|
||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||
description = "Actix http client."
|
||||
readme = "README.md"
|
||||
@ -44,7 +44,7 @@ flate2-rust = ["actix-http/flate2-rust"]
|
||||
[dependencies]
|
||||
actix-codec = "0.1.2"
|
||||
actix-service = "0.4.1"
|
||||
actix-http = "0.2.9"
|
||||
actix-http = "0.2.10"
|
||||
base64 = "0.10.1"
|
||||
bytes = "0.4"
|
||||
derive_more = "0.15.0"
|
||||
@ -63,7 +63,7 @@ rustls = { version = "0.15.2", optional = true }
|
||||
[dev-dependencies]
|
||||
actix-rt = "0.2.2"
|
||||
actix-web = { version = "1.0.0", features=["ssl"] }
|
||||
actix-http = { version = "0.2.4", features=["ssl"] }
|
||||
actix-http = { version = "0.2.10", features=["ssl"] }
|
||||
actix-http-test = { version = "0.2.0", features=["ssl"] }
|
||||
actix-utils = "0.4.1"
|
||||
actix-server = { version = "0.6.0", features=["ssl", "rust-tls"] }
|
||||
|
@ -1,5 +1,5 @@
|
||||
use std::{fmt, io, net};
|
||||
use std::rc::Rc;
|
||||
use std::{fmt, io, net};
|
||||
|
||||
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||
use actix_http::body::Body;
|
||||
@ -7,8 +7,8 @@ use actix_http::client::{
|
||||
Connect as ClientConnect, ConnectError, Connection, SendRequestError,
|
||||
};
|
||||
use actix_http::h1::ClientCodec;
|
||||
use actix_http::{RequestHead, RequestHeadType, ResponseHead};
|
||||
use actix_http::http::HeaderMap;
|
||||
use actix_http::{RequestHead, RequestHeadType, ResponseHead};
|
||||
use actix_service::Service;
|
||||
use futures::{Future, Poll};
|
||||
|
||||
@ -82,7 +82,9 @@ where
|
||||
})
|
||||
.from_err()
|
||||
// send request
|
||||
.and_then(move |connection| connection.send_request(RequestHeadType::from(head), body))
|
||||
.and_then(move |connection| {
|
||||
connection.send_request(RequestHeadType::from(head), body)
|
||||
})
|
||||
.map(|(head, payload)| ClientResponse::new(head, payload)),
|
||||
)
|
||||
}
|
||||
@ -103,7 +105,10 @@ where
|
||||
})
|
||||
.from_err()
|
||||
// send request
|
||||
.and_then(move |connection| connection.send_request(RequestHeadType::Rc(head, extra_headers), body))
|
||||
.and_then(move |connection| {
|
||||
connection
|
||||
.send_request(RequestHeadType::Rc(head, extra_headers), body)
|
||||
})
|
||||
.map(|(head, payload)| ClientResponse::new(head, payload)),
|
||||
)
|
||||
}
|
||||
@ -127,7 +132,9 @@ where
|
||||
})
|
||||
.from_err()
|
||||
// send request
|
||||
.and_then(move |connection| connection.open_tunnel(RequestHeadType::from(head)))
|
||||
.and_then(move |connection| {
|
||||
connection.open_tunnel(RequestHeadType::from(head))
|
||||
})
|
||||
.map(|(head, framed)| {
|
||||
let framed = framed.map_io(|io| BoxedSocket(Box::new(Socket(io))));
|
||||
(head, framed)
|
||||
@ -155,7 +162,9 @@ where
|
||||
})
|
||||
.from_err()
|
||||
// send request
|
||||
.and_then(move |connection| connection.open_tunnel(RequestHeadType::Rc(head, extra_headers)))
|
||||
.and_then(move |connection| {
|
||||
connection.open_tunnel(RequestHeadType::Rc(head, extra_headers))
|
||||
})
|
||||
.map(|(head, framed)| {
|
||||
let framed = framed.map_io(|io| BoxedSocket(Box::new(Socket(io))));
|
||||
(head, framed)
|
||||
|
@ -1,5 +1,7 @@
|
||||
//! Http client errors
|
||||
pub use actix_http::client::{ConnectError, InvalidUrl, SendRequestError, FreezeRequestError};
|
||||
pub use actix_http::client::{
|
||||
ConnectError, FreezeRequestError, InvalidUrl, SendRequestError,
|
||||
};
|
||||
pub use actix_http::error::PayloadError;
|
||||
pub use actix_http::ws::HandshakeError as WsHandshakeError;
|
||||
pub use actix_http::ws::ProtocolError as WsProtocolError;
|
||||
|
235
awc/src/frozen.rs
Normal file
235
awc/src/frozen.rs
Normal file
@ -0,0 +1,235 @@
|
||||
use std::net;
|
||||
use std::rc::Rc;
|
||||
use std::time::Duration;
|
||||
|
||||
use bytes::Bytes;
|
||||
use futures::Stream;
|
||||
use serde::Serialize;
|
||||
|
||||
use actix_http::body::Body;
|
||||
use actix_http::http::header::IntoHeaderValue;
|
||||
use actix_http::http::{
|
||||
Error as HttpError, HeaderMap, HeaderName, HttpTryFrom, Method, Uri,
|
||||
};
|
||||
use actix_http::{Error, RequestHead};
|
||||
|
||||
use crate::sender::{RequestSender, SendClientRequest};
|
||||
use crate::ClientConfig;
|
||||
|
||||
/// `FrozenClientRequest` struct represents clonable client request.
|
||||
/// It could be used to send same request multiple times.
|
||||
#[derive(Clone)]
|
||||
pub struct FrozenClientRequest {
|
||||
pub(crate) head: Rc<RequestHead>,
|
||||
pub(crate) addr: Option<net::SocketAddr>,
|
||||
pub(crate) response_decompress: bool,
|
||||
pub(crate) timeout: Option<Duration>,
|
||||
pub(crate) config: Rc<ClientConfig>,
|
||||
}
|
||||
|
||||
impl FrozenClientRequest {
|
||||
/// Get HTTP URI of request
|
||||
pub fn get_uri(&self) -> &Uri {
|
||||
&self.head.uri
|
||||
}
|
||||
|
||||
/// Get HTTP method of this request
|
||||
pub fn get_method(&self) -> &Method {
|
||||
&self.head.method
|
||||
}
|
||||
|
||||
/// Returns request's headers.
|
||||
pub fn headers(&self) -> &HeaderMap {
|
||||
&self.head.headers
|
||||
}
|
||||
|
||||
/// Send a body.
|
||||
pub fn send_body<B>(&self, body: B) -> SendClientRequest
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None).send_body(
|
||||
self.addr,
|
||||
self.response_decompress,
|
||||
self.timeout,
|
||||
self.config.as_ref(),
|
||||
body,
|
||||
)
|
||||
}
|
||||
|
||||
/// Send a json body.
|
||||
pub fn send_json<T: Serialize>(&self, value: &T) -> SendClientRequest {
|
||||
RequestSender::Rc(self.head.clone(), None).send_json(
|
||||
self.addr,
|
||||
self.response_decompress,
|
||||
self.timeout,
|
||||
self.config.as_ref(),
|
||||
value,
|
||||
)
|
||||
}
|
||||
|
||||
/// Send an urlencoded body.
|
||||
pub fn send_form<T: Serialize>(&self, value: &T) -> SendClientRequest {
|
||||
RequestSender::Rc(self.head.clone(), None).send_form(
|
||||
self.addr,
|
||||
self.response_decompress,
|
||||
self.timeout,
|
||||
self.config.as_ref(),
|
||||
value,
|
||||
)
|
||||
}
|
||||
|
||||
/// Send a streaming body.
|
||||
pub fn send_stream<S, E>(&self, stream: S) -> SendClientRequest
|
||||
where
|
||||
S: Stream<Item = Bytes, Error = E> + 'static,
|
||||
E: Into<Error> + 'static,
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None).send_stream(
|
||||
self.addr,
|
||||
self.response_decompress,
|
||||
self.timeout,
|
||||
self.config.as_ref(),
|
||||
stream,
|
||||
)
|
||||
}
|
||||
|
||||
/// Send an empty body.
|
||||
pub fn send(&self) -> SendClientRequest {
|
||||
RequestSender::Rc(self.head.clone(), None).send(
|
||||
self.addr,
|
||||
self.response_decompress,
|
||||
self.timeout,
|
||||
self.config.as_ref(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Create a `FrozenSendBuilder` with extra headers
|
||||
pub fn extra_headers(&self, extra_headers: HeaderMap) -> FrozenSendBuilder {
|
||||
FrozenSendBuilder::new(self.clone(), extra_headers)
|
||||
}
|
||||
|
||||
/// Create a `FrozenSendBuilder` with an extra header
|
||||
pub fn extra_header<K, V>(&self, key: K, value: V) -> FrozenSendBuilder
|
||||
where
|
||||
HeaderName: HttpTryFrom<K>,
|
||||
V: IntoHeaderValue,
|
||||
{
|
||||
self.extra_headers(HeaderMap::new())
|
||||
.extra_header(key, value)
|
||||
}
|
||||
}
|
||||
|
||||
/// Builder that allows to modify extra headers.
|
||||
pub struct FrozenSendBuilder {
|
||||
req: FrozenClientRequest,
|
||||
extra_headers: HeaderMap,
|
||||
err: Option<HttpError>,
|
||||
}
|
||||
|
||||
impl FrozenSendBuilder {
|
||||
pub(crate) fn new(req: FrozenClientRequest, extra_headers: HeaderMap) -> Self {
|
||||
Self {
|
||||
req,
|
||||
extra_headers,
|
||||
err: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Insert a header, it overrides existing header in `FrozenClientRequest`.
|
||||
pub fn extra_header<K, V>(mut self, key: K, value: V) -> Self
|
||||
where
|
||||
HeaderName: HttpTryFrom<K>,
|
||||
V: IntoHeaderValue,
|
||||
{
|
||||
match HeaderName::try_from(key) {
|
||||
Ok(key) => match value.try_into() {
|
||||
Ok(value) => self.extra_headers.insert(key, value),
|
||||
Err(e) => self.err = Some(e.into()),
|
||||
},
|
||||
Err(e) => self.err = Some(e.into()),
|
||||
}
|
||||
self
|
||||
}
|
||||
|
||||
/// Complete request construction and send a body.
|
||||
pub fn send_body<B>(self, body: B) -> SendClientRequest
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
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,
|
||||
)
|
||||
}
|
||||
|
||||
/// Complete request construction and send a json body.
|
||||
pub fn send_json<T: Serialize>(self, value: &T) -> SendClientRequest {
|
||||
if let Some(e) = self.err {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
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,
|
||||
)
|
||||
}
|
||||
|
||||
/// Complete request construction and send an urlencoded body.
|
||||
pub fn send_form<T: Serialize>(self, value: &T) -> SendClientRequest {
|
||||
if let Some(e) = self.err {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
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,
|
||||
)
|
||||
}
|
||||
|
||||
/// Complete request construction and send a streaming body.
|
||||
pub fn send_stream<S, E>(self, stream: S) -> SendClientRequest
|
||||
where
|
||||
S: Stream<Item = Bytes, Error = E> + 'static,
|
||||
E: Into<Error> + 'static,
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
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,
|
||||
)
|
||||
}
|
||||
|
||||
/// Complete request construction and send an empty body.
|
||||
pub fn send(self) -> SendClientRequest {
|
||||
if let Some(e) = self.err {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
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(),
|
||||
)
|
||||
}
|
||||
}
|
@ -33,15 +33,19 @@ use actix_http::RequestHead;
|
||||
mod builder;
|
||||
mod connect;
|
||||
pub mod error;
|
||||
mod frozen;
|
||||
mod request;
|
||||
mod response;
|
||||
mod sender;
|
||||
pub mod test;
|
||||
pub mod ws;
|
||||
|
||||
pub use self::builder::ClientBuilder;
|
||||
pub use self::connect::BoxedSocket;
|
||||
pub use self::frozen::{FrozenClientRequest, FrozenSendBuilder};
|
||||
pub use self::request::ClientRequest;
|
||||
pub use self::response::{ClientResponse, JsonBody, MessageBody};
|
||||
pub use self::sender::SendClientRequest;
|
||||
|
||||
use self::connect::{Connect, ConnectorWrapper};
|
||||
|
||||
|
@ -1,29 +1,26 @@
|
||||
use std::fmt::Write as FmtWrite;
|
||||
use std::io::Write;
|
||||
use std::rc::Rc;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::time::Duration;
|
||||
use std::{fmt, net};
|
||||
|
||||
use bytes::{BufMut, Bytes, BytesMut};
|
||||
use futures::{Async, Future, Poll, Stream, try_ready};
|
||||
use futures::Stream;
|
||||
use percent_encoding::percent_encode;
|
||||
use serde::Serialize;
|
||||
use serde_json;
|
||||
use tokio_timer::Delay;
|
||||
use derive_more::From;
|
||||
|
||||
use actix_http::body::{Body, BodyStream};
|
||||
use actix_http::body::Body;
|
||||
use actix_http::cookie::{Cookie, CookieJar, USERINFO};
|
||||
use actix_http::encoding::Decoder;
|
||||
use actix_http::http::header::{self, ContentEncoding, Header, IntoHeaderValue};
|
||||
use actix_http::http::header::{self, Header, IntoHeaderValue};
|
||||
use actix_http::http::{
|
||||
uri, ConnectionType, Error as HttpError, HeaderMap, HeaderName, HeaderValue,
|
||||
HttpTryFrom, Method, Uri, Version,
|
||||
};
|
||||
use actix_http::{Error, Payload, PayloadStream, RequestHead};
|
||||
use actix_http::{Error, RequestHead};
|
||||
|
||||
use crate::error::{InvalidUrl, SendRequestError, FreezeRequestError};
|
||||
use crate::response::ClientResponse;
|
||||
use crate::error::{FreezeRequestError, InvalidUrl};
|
||||
use crate::frozen::FrozenClientRequest;
|
||||
use crate::sender::{PrepForSendingError, RequestSender, SendClientRequest};
|
||||
use crate::ClientConfig;
|
||||
|
||||
#[cfg(any(feature = "brotli", feature = "flate2-zlib", feature = "flate2-rust"))]
|
||||
@ -375,6 +372,8 @@ impl ClientRequest {
|
||||
}
|
||||
}
|
||||
|
||||
/// Freeze request builder and construct `FrozenClientRequest`,
|
||||
/// which could be used for sending same request multiple times.
|
||||
pub fn freeze(self) -> Result<FrozenClientRequest, FreezeRequestError> {
|
||||
let slf = match self.prep_for_sending() {
|
||||
Ok(slf) => slf,
|
||||
@ -393,10 +392,7 @@ impl ClientRequest {
|
||||
}
|
||||
|
||||
/// Complete request construction and send body.
|
||||
pub fn send_body<B>(
|
||||
self,
|
||||
body: B,
|
||||
) -> SendBody
|
||||
pub fn send_body<B>(self, body: B) -> SendClientRequest
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
@ -405,47 +401,51 @@ impl ClientRequest {
|
||||
Err(e) => return e.into(),
|
||||
};
|
||||
|
||||
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`
|
||||
pub fn send_json<T: Serialize>(
|
||||
self,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
pub fn send_json<T: Serialize>(self, value: &T) -> SendClientRequest {
|
||||
let slf = match self.prep_for_sending() {
|
||||
Ok(slf) => slf,
|
||||
Err(e) => return e.into(),
|
||||
};
|
||||
|
||||
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`
|
||||
///
|
||||
/// `ClientRequestBuilder` can not be used after this call.
|
||||
pub fn send_form<T: Serialize>(
|
||||
self,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
pub fn send_form<T: Serialize>(self, value: &T) -> SendClientRequest {
|
||||
let slf = match self.prep_for_sending() {
|
||||
Ok(slf) => slf,
|
||||
Err(e) => return e.into(),
|
||||
};
|
||||
|
||||
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`.
|
||||
pub fn send_stream<S, E>(
|
||||
self,
|
||||
stream: S,
|
||||
) -> SendBody
|
||||
pub fn send_stream<S, E>(self, stream: S) -> SendClientRequest
|
||||
where
|
||||
S: Stream<Item = Bytes, Error = E> + 'static,
|
||||
E: Into<Error> + 'static,
|
||||
@ -455,22 +455,28 @@ impl ClientRequest {
|
||||
Err(e) => return e.into(),
|
||||
};
|
||||
|
||||
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`.
|
||||
pub fn send(
|
||||
self,
|
||||
) -> SendBody
|
||||
{
|
||||
pub fn send(self) -> SendClientRequest {
|
||||
let slf = match self.prep_for_sending() {
|
||||
Ok(slf) => slf,
|
||||
Err(e) => return e.into(),
|
||||
};
|
||||
|
||||
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> {
|
||||
@ -555,441 +561,6 @@ impl fmt::Debug for ClientRequest {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct FrozenClientRequest {
|
||||
pub(crate) head: Rc<RequestHead>,
|
||||
pub(crate) addr: Option<net::SocketAddr>,
|
||||
pub(crate) response_decompress: bool,
|
||||
pub(crate) timeout: Option<Duration>,
|
||||
pub(crate) config: Rc<ClientConfig>,
|
||||
}
|
||||
|
||||
impl FrozenClientRequest {
|
||||
/// Get HTTP URI of request
|
||||
pub fn get_uri(&self) -> &Uri {
|
||||
&self.head.uri
|
||||
}
|
||||
|
||||
/// Get HTTP method of this request
|
||||
pub fn get_method(&self) -> &Method {
|
||||
&self.head.method
|
||||
}
|
||||
|
||||
/// Returns request's headers.
|
||||
pub fn headers(&self) -> &HeaderMap {
|
||||
&self.head.headers
|
||||
}
|
||||
|
||||
/// Send a body.
|
||||
pub fn send_body<B>(
|
||||
&self,
|
||||
body: B,
|
||||
) -> SendBody
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None)
|
||||
.send_body(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), body)
|
||||
}
|
||||
|
||||
/// Send a json body.
|
||||
pub fn send_json<T: Serialize>(
|
||||
&self,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None)
|
||||
.send_json(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), value)
|
||||
}
|
||||
|
||||
/// Send an urlencoded body.
|
||||
pub fn send_form<T: Serialize>(
|
||||
&self,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None)
|
||||
.send_form(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), value)
|
||||
}
|
||||
|
||||
/// Send a streaming body.
|
||||
pub fn send_stream<S, E>(
|
||||
&self,
|
||||
stream: S,
|
||||
) -> SendBody
|
||||
where
|
||||
S: Stream<Item = Bytes, Error = E> + 'static,
|
||||
E: Into<Error> + 'static,
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None)
|
||||
.send_stream(self.addr, self.response_decompress, self.timeout, self.config.as_ref(), stream)
|
||||
}
|
||||
|
||||
/// Send an empty body.
|
||||
pub fn send(
|
||||
&self,
|
||||
) -> SendBody
|
||||
{
|
||||
RequestSender::Rc(self.head.clone(), None)
|
||||
.send(self.addr, self.response_decompress, self.timeout, self.config.as_ref())
|
||||
}
|
||||
|
||||
/// Create a `FrozenSendBuilder` with extra headers
|
||||
pub fn extra_headers(&self, extra_headers: HeaderMap) -> FrozenSendBuilder {
|
||||
FrozenSendBuilder::new(self.clone(), extra_headers)
|
||||
}
|
||||
|
||||
/// Create a `FrozenSendBuilder` with an extra header
|
||||
pub fn extra_header<K, V>(&self, key: K, value: V) -> FrozenSendBuilder
|
||||
where
|
||||
HeaderName: HttpTryFrom<K>,
|
||||
V: IntoHeaderValue,
|
||||
{
|
||||
self.extra_headers(HeaderMap::new()).extra_header(key, value)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct FrozenSendBuilder {
|
||||
req: FrozenClientRequest,
|
||||
extra_headers: HeaderMap,
|
||||
err: Option<HttpError>,
|
||||
}
|
||||
|
||||
impl FrozenSendBuilder {
|
||||
pub(crate) fn new(req: FrozenClientRequest, extra_headers: HeaderMap) -> Self {
|
||||
Self {
|
||||
req,
|
||||
extra_headers,
|
||||
err: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Insert a header, it overrides existing header in `FrozenClientRequest`.
|
||||
pub fn extra_header<K, V>(mut self, key: K, value: V) -> Self
|
||||
where
|
||||
HeaderName: HttpTryFrom<K>,
|
||||
V: IntoHeaderValue,
|
||||
{
|
||||
match HeaderName::try_from(key) {
|
||||
Ok(key) => match value.try_into() {
|
||||
Ok(value) => self.extra_headers.insert(key, value),
|
||||
Err(e) => self.err = Some(e.into()),
|
||||
},
|
||||
Err(e) => self.err = Some(e.into()),
|
||||
}
|
||||
self
|
||||
}
|
||||
|
||||
/// Complete request construction and send a body.
|
||||
pub fn send_body<B>(
|
||||
self,
|
||||
body: B,
|
||||
) -> SendBody
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into()
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
/// Complete request construction and send a json body.
|
||||
pub fn send_json<T: Serialize>(
|
||||
self,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into()
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
/// Complete request construction and send an urlencoded body.
|
||||
pub fn send_form<T: Serialize>(
|
||||
self,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into()
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
/// Complete request construction and send a streaming body.
|
||||
pub fn send_stream<S, E>(
|
||||
self,
|
||||
stream: S,
|
||||
) -> SendBody
|
||||
where
|
||||
S: Stream<Item = Bytes, Error = E> + 'static,
|
||||
E: Into<Error> + 'static,
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into()
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
/// Complete request construction and send an empty body.
|
||||
pub fn send(
|
||||
self,
|
||||
) -> SendBody
|
||||
{
|
||||
if let Some(e) = self.err {
|
||||
return e.into()
|
||||
}
|
||||
|
||||
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())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, From)]
|
||||
enum PrepForSendingError {
|
||||
Url(InvalidUrl),
|
||||
Http(HttpError),
|
||||
}
|
||||
|
||||
impl Into<FreezeRequestError> for PrepForSendingError {
|
||||
fn into(self) -> FreezeRequestError {
|
||||
match self {
|
||||
PrepForSendingError::Url(e) => FreezeRequestError::Url(e),
|
||||
PrepForSendingError::Http(e) => FreezeRequestError::Http(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Into<SendRequestError> for PrepForSendingError {
|
||||
fn into(self) -> SendRequestError {
|
||||
match self {
|
||||
PrepForSendingError::Url(e) => SendRequestError::Url(e),
|
||||
PrepForSendingError::Http(e) => SendRequestError::Http(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub enum SendBody
|
||||
{
|
||||
Fut(Box<dyn Future<Item = ClientResponse, Error = SendRequestError>>, Option<Delay>, bool),
|
||||
Err(Option<SendRequestError>),
|
||||
}
|
||||
|
||||
impl SendBody
|
||||
{
|
||||
pub fn new(
|
||||
send: Box<dyn Future<Item = ClientResponse, Error = SendRequestError>>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
) -> SendBody
|
||||
{
|
||||
let delay = timeout.map(|t| Delay::new(Instant::now() + t));
|
||||
SendBody::Fut(send, delay, response_decompress)
|
||||
}
|
||||
}
|
||||
|
||||
impl Future for SendBody
|
||||
{
|
||||
type Item = ClientResponse<Decoder<Payload<PayloadStream>>>;
|
||||
type Error = SendRequestError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
match self {
|
||||
SendBody::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))
|
||||
},
|
||||
SendBody::Err(ref mut e) => {
|
||||
match e.take() {
|
||||
Some(e) => Err(e.into()),
|
||||
None => panic!("Attempting to call completed future"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
impl From<SendRequestError> for SendBody
|
||||
{
|
||||
fn from(e: SendRequestError) -> Self {
|
||||
SendBody::Err(Some(e))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Error> for SendBody
|
||||
{
|
||||
fn from(e: Error) -> Self {
|
||||
SendBody::Err(Some(e.into()))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<HttpError> for SendBody
|
||||
{
|
||||
fn from(e: HttpError) -> Self {
|
||||
SendBody::Err(Some(e.into()))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<PrepForSendingError> for SendBody
|
||||
{
|
||||
fn from(e: PrepForSendingError) -> Self {
|
||||
SendBody::Err(Some(e.into()))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum RequestSender {
|
||||
Owned(RequestHead),
|
||||
Rc(Rc<RequestHead>, Option<HeaderMap>),
|
||||
}
|
||||
|
||||
impl RequestSender {
|
||||
pub fn send_body<B>(
|
||||
self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
body: B,
|
||||
) -> SendBody
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
let mut connector = config.connector.borrow_mut();
|
||||
|
||||
let fut = match self {
|
||||
RequestSender::Owned(head) => connector.send_request(head, body.into(), addr),
|
||||
RequestSender::Rc(head, extra_headers) => connector.send_request_extra(head, extra_headers, body.into(), addr),
|
||||
};
|
||||
|
||||
SendBody::new(fut, response_decompress, timeout.or_else(|| config.timeout.clone()))
|
||||
}
|
||||
|
||||
pub fn send_json<T: Serialize>(
|
||||
mut self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
let body = match serde_json::to_string(value) {
|
||||
Ok(body) => body,
|
||||
Err(e) => return Error::from(e).into(),
|
||||
};
|
||||
|
||||
if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/json") {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
self.send_body(addr, response_decompress, timeout, config, Body::Bytes(Bytes::from(body)))
|
||||
}
|
||||
|
||||
pub fn send_form<T: Serialize>(
|
||||
mut self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
value: &T,
|
||||
) -> SendBody
|
||||
{
|
||||
let body = match serde_urlencoded::to_string(value) {
|
||||
Ok(body) => body,
|
||||
Err(e) => return Error::from(e).into(),
|
||||
};
|
||||
|
||||
// set content-type
|
||||
if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/x-www-form-urlencoded") {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
self.send_body(addr, response_decompress, timeout, config, Body::Bytes(Bytes::from(body)))
|
||||
}
|
||||
|
||||
pub fn send_stream<S, E>(
|
||||
self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
stream: S,
|
||||
) -> SendBody
|
||||
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)))
|
||||
}
|
||||
|
||||
pub fn send(
|
||||
self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
) -> SendBody
|
||||
{
|
||||
self.send_body(addr, response_decompress, timeout, config, Body::Empty)
|
||||
}
|
||||
|
||||
fn set_header_if_none<V>(&mut self, key: HeaderName, value: V) -> Result<(), HttpError>
|
||||
where
|
||||
V: IntoHeaderValue,
|
||||
{
|
||||
match self {
|
||||
RequestSender::Owned(head) => {
|
||||
if !head.headers.contains_key(&key) {
|
||||
match value.try_into() {
|
||||
Ok(value) => head.headers.insert(key, value),
|
||||
Err(e) => return Err(e.into()),
|
||||
}
|
||||
}
|
||||
},
|
||||
RequestSender::Rc(head, extra_headers) => {
|
||||
if !head.headers.contains_key(&key) && !extra_headers.iter().any(|h| h.contains_key(&key)) {
|
||||
match value.try_into(){
|
||||
Ok(v) => {
|
||||
let h = extra_headers.get_or_insert(HeaderMap::new());
|
||||
h.insert(key, v)
|
||||
},
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::time::SystemTime;
|
||||
|
282
awc/src/sender.rs
Normal file
282
awc/src/sender.rs
Normal file
@ -0,0 +1,282 @@
|
||||
use std::net;
|
||||
use std::rc::Rc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use bytes::Bytes;
|
||||
use derive_more::From;
|
||||
use futures::{try_ready, Async, Future, Poll, Stream};
|
||||
use serde::Serialize;
|
||||
use serde_json;
|
||||
use tokio_timer::Delay;
|
||||
|
||||
use actix_http::body::{Body, BodyStream};
|
||||
use actix_http::encoding::Decoder;
|
||||
use actix_http::http::header::{self, ContentEncoding, IntoHeaderValue};
|
||||
use actix_http::http::{Error as HttpError, HeaderMap, HeaderName};
|
||||
use actix_http::{Error, Payload, PayloadStream, RequestHead};
|
||||
|
||||
use crate::error::{FreezeRequestError, InvalidUrl, SendRequestError};
|
||||
use crate::response::ClientResponse;
|
||||
use crate::ClientConfig;
|
||||
|
||||
#[derive(Debug, From)]
|
||||
pub(crate) enum PrepForSendingError {
|
||||
Url(InvalidUrl),
|
||||
Http(HttpError),
|
||||
}
|
||||
|
||||
impl Into<FreezeRequestError> for PrepForSendingError {
|
||||
fn into(self) -> FreezeRequestError {
|
||||
match self {
|
||||
PrepForSendingError::Url(e) => FreezeRequestError::Url(e),
|
||||
PrepForSendingError::Http(e) => FreezeRequestError::Http(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Into<SendRequestError> for PrepForSendingError {
|
||||
fn into(self) -> SendRequestError {
|
||||
match self {
|
||||
PrepForSendingError::Url(e) => SendRequestError::Url(e),
|
||||
PrepForSendingError::Http(e) => SendRequestError::Http(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Future that sends request's payload and resolves to a server response.
|
||||
#[must_use = "futures do nothing unless polled"]
|
||||
pub enum SendClientRequest {
|
||||
Fut(
|
||||
Box<dyn Future<Item = ClientResponse, Error = SendRequestError>>,
|
||||
Option<Delay>,
|
||||
bool,
|
||||
),
|
||||
Err(Option<SendRequestError>),
|
||||
}
|
||||
|
||||
impl SendClientRequest {
|
||||
pub(crate) fn new(
|
||||
send: Box<dyn Future<Item = ClientResponse, Error = SendRequestError>>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
) -> SendClientRequest {
|
||||
let delay = timeout.map(|t| Delay::new(Instant::now() + t));
|
||||
SendClientRequest::Fut(send, delay, response_decompress)
|
||||
}
|
||||
}
|
||||
|
||||
impl Future for SendClientRequest {
|
||||
type Item = ClientResponse<Decoder<Payload<PayloadStream>>>;
|
||||
type Error = SendRequestError;
|
||||
|
||||
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
|
||||
match self {
|
||||
SendClientRequest::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))
|
||||
}
|
||||
SendClientRequest::Err(ref mut e) => match e.take() {
|
||||
Some(e) => Err(e),
|
||||
None => panic!("Attempting to call completed future"),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<SendRequestError> for SendClientRequest {
|
||||
fn from(e: SendRequestError) -> Self {
|
||||
SendClientRequest::Err(Some(e))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Error> for SendClientRequest {
|
||||
fn from(e: Error) -> Self {
|
||||
SendClientRequest::Err(Some(e.into()))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<HttpError> for SendClientRequest {
|
||||
fn from(e: HttpError) -> Self {
|
||||
SendClientRequest::Err(Some(e.into()))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<PrepForSendingError> for SendClientRequest {
|
||||
fn from(e: PrepForSendingError) -> Self {
|
||||
SendClientRequest::Err(Some(e.into()))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum RequestSender {
|
||||
Owned(RequestHead),
|
||||
Rc(Rc<RequestHead>, Option<HeaderMap>),
|
||||
}
|
||||
|
||||
impl RequestSender {
|
||||
pub(crate) fn send_body<B>(
|
||||
self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
body: B,
|
||||
) -> SendClientRequest
|
||||
where
|
||||
B: Into<Body>,
|
||||
{
|
||||
let mut connector = config.connector.borrow_mut();
|
||||
|
||||
let fut = match self {
|
||||
RequestSender::Owned(head) => {
|
||||
connector.send_request(head, body.into(), addr)
|
||||
}
|
||||
RequestSender::Rc(head, extra_headers) => {
|
||||
connector.send_request_extra(head, extra_headers, body.into(), addr)
|
||||
}
|
||||
};
|
||||
|
||||
SendClientRequest::new(
|
||||
fut,
|
||||
response_decompress,
|
||||
timeout.or_else(|| config.timeout),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn send_json<T: Serialize>(
|
||||
mut self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
value: &T,
|
||||
) -> SendClientRequest {
|
||||
let body = match serde_json::to_string(value) {
|
||||
Ok(body) => body,
|
||||
Err(e) => return Error::from(e).into(),
|
||||
};
|
||||
|
||||
if let Err(e) = self.set_header_if_none(header::CONTENT_TYPE, "application/json")
|
||||
{
|
||||
return e.into();
|
||||
}
|
||||
|
||||
self.send_body(
|
||||
addr,
|
||||
response_decompress,
|
||||
timeout,
|
||||
config,
|
||||
Body::Bytes(Bytes::from(body)),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn send_form<T: Serialize>(
|
||||
mut self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
value: &T,
|
||||
) -> SendClientRequest {
|
||||
let body = match serde_urlencoded::to_string(value) {
|
||||
Ok(body) => body,
|
||||
Err(e) => return Error::from(e).into(),
|
||||
};
|
||||
|
||||
// set content-type
|
||||
if let Err(e) = self.set_header_if_none(
|
||||
header::CONTENT_TYPE,
|
||||
"application/x-www-form-urlencoded",
|
||||
) {
|
||||
return e.into();
|
||||
}
|
||||
|
||||
self.send_body(
|
||||
addr,
|
||||
response_decompress,
|
||||
timeout,
|
||||
config,
|
||||
Body::Bytes(Bytes::from(body)),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn send_stream<S, E>(
|
||||
self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
stream: S,
|
||||
) -> SendClientRequest
|
||||
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)),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn send(
|
||||
self,
|
||||
addr: Option<net::SocketAddr>,
|
||||
response_decompress: bool,
|
||||
timeout: Option<Duration>,
|
||||
config: &ClientConfig,
|
||||
) -> SendClientRequest {
|
||||
self.send_body(addr, response_decompress, timeout, config, Body::Empty)
|
||||
}
|
||||
|
||||
fn set_header_if_none<V>(
|
||||
&mut self,
|
||||
key: HeaderName,
|
||||
value: V,
|
||||
) -> Result<(), HttpError>
|
||||
where
|
||||
V: IntoHeaderValue,
|
||||
{
|
||||
match self {
|
||||
RequestSender::Owned(head) => {
|
||||
if !head.headers.contains_key(&key) {
|
||||
match value.try_into() {
|
||||
Ok(value) => head.headers.insert(key, value),
|
||||
Err(e) => return Err(e.into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
RequestSender::Rc(head, extra_headers) => {
|
||||
if !head.headers.contains_key(&key)
|
||||
&& !extra_headers.iter().any(|h| h.contains_key(&key))
|
||||
{
|
||||
match value.try_into() {
|
||||
Ok(v) => {
|
||||
let h = extra_headers.get_or_insert(HeaderMap::new());
|
||||
h.insert(key, v)
|
||||
}
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
@ -234,7 +234,10 @@ impl WebsocketsRequest {
|
||||
}
|
||||
|
||||
if !self.head.headers.contains_key(header::HOST) {
|
||||
self.head.headers.insert(header::HOST, HeaderValue::from_str(uri.host().unwrap()).unwrap());
|
||||
self.head.headers.insert(
|
||||
header::HOST,
|
||||
HeaderValue::from_str(uri.host().unwrap()).unwrap(),
|
||||
);
|
||||
}
|
||||
|
||||
// set cookies
|
||||
|
@ -169,7 +169,7 @@ where
|
||||
match self.data_factories_fut[idx].poll()? {
|
||||
Async::Ready(f) => {
|
||||
self.data_factories.push(f);
|
||||
self.data_factories_fut.remove(idx);
|
||||
let _ = self.data_factories_fut.remove(idx);
|
||||
}
|
||||
Async::NotReady => idx += 1,
|
||||
}
|
||||
|
@ -2,13 +2,13 @@
|
||||
mod compress;
|
||||
pub use self::compress::{BodyEncoding, Compress};
|
||||
|
||||
mod condition;
|
||||
mod defaultheaders;
|
||||
pub mod errhandlers;
|
||||
mod logger;
|
||||
mod normalize;
|
||||
mod condition;
|
||||
|
||||
pub use self::condition::Condition;
|
||||
pub use self::defaultheaders::DefaultHeaders;
|
||||
pub use self::logger::Logger;
|
||||
pub use self::normalize::NormalizePath;
|
||||
pub use self::condition::Condition;
|
||||
|
Reference in New Issue
Block a user