mirror of
https://github.com/fafhrd91/actix-web
synced 2025-07-08 11:56:32 +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"
|
h2 = "0.1.16"
|
||||||
http = "0.1.17"
|
http = "0.1.17"
|
||||||
httparse = "1.3"
|
httparse = "1.3"
|
||||||
indexmap = "1.0"
|
indexmap = "1.2"
|
||||||
lazy_static = "1.0"
|
lazy_static = "1.0"
|
||||||
language-tags = "0.2"
|
language-tags = "0.2"
|
||||||
log = "0.4"
|
log = "0.4"
|
||||||
|
@ -8,10 +8,10 @@ use futures::{Async, Future, Poll, Sink, Stream};
|
|||||||
|
|
||||||
use crate::error::PayloadError;
|
use crate::error::PayloadError;
|
||||||
use crate::h1;
|
use crate::h1;
|
||||||
|
use crate::header::HeaderMap;
|
||||||
use crate::http::header::{IntoHeaderValue, HOST};
|
use crate::http::header::{IntoHeaderValue, HOST};
|
||||||
use crate::message::{RequestHeadType, ResponseHead};
|
use crate::message::{RequestHeadType, ResponseHead};
|
||||||
use crate::payload::{Payload, PayloadStream};
|
use crate::payload::{Payload, PayloadStream};
|
||||||
use crate::header::HeaderMap;
|
|
||||||
|
|
||||||
use super::connection::{ConnectionLifetime, ConnectionType, IoConnection};
|
use super::connection::{ConnectionLifetime, ConnectionType, IoConnection};
|
||||||
use super::error::{ConnectError, SendRequestError};
|
use super::error::{ConnectError, SendRequestError};
|
||||||
@ -30,7 +30,9 @@ where
|
|||||||
B: MessageBody,
|
B: MessageBody,
|
||||||
{
|
{
|
||||||
// set request host header
|
// 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() {
|
if let Some(host) = head.as_ref().uri.host() {
|
||||||
let mut wrt = BytesMut::with_capacity(host.len() + 5).writer();
|
let mut wrt = BytesMut::with_capacity(host.len() + 5).writer();
|
||||||
|
|
||||||
@ -40,20 +42,16 @@ where
|
|||||||
};
|
};
|
||||||
|
|
||||||
match wrt.get_mut().take().freeze().try_into() {
|
match wrt.get_mut().take().freeze().try_into() {
|
||||||
Ok(value) => {
|
Ok(value) => match head {
|
||||||
match head {
|
RequestHeadType::Owned(ref mut head) => {
|
||||||
RequestHeadType::Owned(ref mut head) => {
|
head.headers.insert(HOST, value)
|
||||||
head.headers.insert(HOST, value)
|
|
||||||
},
|
|
||||||
RequestHeadType::Rc(_, ref mut extra_headers) => {
|
|
||||||
let headers = extra_headers.get_or_insert(HeaderMap::new());
|
|
||||||
headers.insert(HOST, value)
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
}
|
RequestHeadType::Rc(_, ref mut extra_headers) => {
|
||||||
Err(e) => {
|
let headers = extra_headers.get_or_insert(HeaderMap::new());
|
||||||
log::error!("Can not set HOST header {}", e)
|
headers.insert(HOST, value)
|
||||||
}
|
}
|
||||||
|
},
|
||||||
|
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 http::{request::Request, HttpTryFrom, Method, Version};
|
||||||
|
|
||||||
use crate::body::{BodySize, MessageBody};
|
use crate::body::{BodySize, MessageBody};
|
||||||
|
use crate::header::HeaderMap;
|
||||||
use crate::message::{RequestHeadType, ResponseHead};
|
use crate::message::{RequestHeadType, ResponseHead};
|
||||||
use crate::payload::Payload;
|
use crate::payload::Payload;
|
||||||
use crate::header::HeaderMap;
|
|
||||||
|
|
||||||
use super::connection::{ConnectionType, IoConnection};
|
use super::connection::{ConnectionType, IoConnection};
|
||||||
use super::error::SendRequestError;
|
use super::error::SendRequestError;
|
||||||
@ -69,15 +69,21 @@ where
|
|||||||
|
|
||||||
// Extracting extra headers from RequestHeadType. HeaderMap::new() does not allocate.
|
// Extracting extra headers from RequestHeadType. HeaderMap::new() does not allocate.
|
||||||
let (head, extra_headers) = match head {
|
let (head, extra_headers) = match head {
|
||||||
RequestHeadType::Owned(head) => (RequestHeadType::Owned(head), HeaderMap::new()),
|
RequestHeadType::Owned(head) => {
|
||||||
RequestHeadType::Rc(head, extra_headers) => (RequestHeadType::Rc(head, None), extra_headers.unwrap_or(HeaderMap::new())),
|
(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.
|
// merging headers from head and extra headers.
|
||||||
let headers = head.as_ref().headers.iter()
|
let headers = head
|
||||||
.filter(|(name, _)| {
|
.as_ref()
|
||||||
!extra_headers.contains_key(*name)
|
.headers
|
||||||
})
|
.iter()
|
||||||
|
.filter(|(name, _)| !extra_headers.contains_key(*name))
|
||||||
.chain(extra_headers.iter());
|
.chain(extra_headers.iter());
|
||||||
|
|
||||||
// copy headers
|
// copy headers
|
||||||
|
@ -10,7 +10,7 @@ mod pool;
|
|||||||
|
|
||||||
pub use self::connection::Connection;
|
pub use self::connection::Connection;
|
||||||
pub use self::connector::Connector;
|
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;
|
pub use self::pool::Protocol;
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
|
@ -326,7 +326,7 @@ impl<Io> Inner<Io> {
|
|||||||
|
|
||||||
fn release_waiter(&mut self, key: &Key, token: usize) {
|
fn release_waiter(&mut self, key: &Key, token: usize) {
|
||||||
self.waiters.remove(token);
|
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::body::BodySize;
|
||||||
use crate::config::ServiceConfig;
|
use crate::config::ServiceConfig;
|
||||||
use crate::error::{ParseError, PayloadError};
|
use crate::error::{ParseError, PayloadError};
|
||||||
use crate::helpers;
|
|
||||||
use crate::message::{ConnectionType, Head, MessagePool, RequestHead, RequestHeadType, ResponseHead};
|
|
||||||
use crate::header::HeaderMap;
|
use crate::header::HeaderMap;
|
||||||
|
use crate::helpers;
|
||||||
|
use crate::message::{
|
||||||
|
ConnectionType, Head, MessagePool, RequestHead, RequestHeadType, ResponseHead,
|
||||||
|
};
|
||||||
|
|
||||||
bitflags! {
|
bitflags! {
|
||||||
struct Flags: u8 {
|
struct Flags: u8 {
|
||||||
@ -197,7 +199,9 @@ impl Encoder for ClientCodec {
|
|||||||
Message::Item((mut head, length)) => {
|
Message::Item((mut head, length)) => {
|
||||||
let inner = &mut self.inner;
|
let inner = &mut self.inner;
|
||||||
inner.version = head.as_ref().version;
|
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
|
// connection status
|
||||||
inner.ctype = match head.as_ref().connection_type() {
|
inner.ctype = match head.as_ref().connection_type() {
|
||||||
|
@ -2,9 +2,9 @@
|
|||||||
use std::fmt::Write as FmtWrite;
|
use std::fmt::Write as FmtWrite;
|
||||||
use std::io::Write;
|
use std::io::Write;
|
||||||
use std::marker::PhantomData;
|
use std::marker::PhantomData;
|
||||||
|
use std::rc::Rc;
|
||||||
use std::str::FromStr;
|
use std::str::FromStr;
|
||||||
use std::{cmp, fmt, io, mem};
|
use std::{cmp, fmt, io, mem};
|
||||||
use std::rc::Rc;
|
|
||||||
|
|
||||||
use bytes::{BufMut, Bytes, BytesMut};
|
use bytes::{BufMut, Bytes, BytesMut};
|
||||||
|
|
||||||
@ -16,7 +16,7 @@ use crate::http::header::{
|
|||||||
HeaderValue, ACCEPT_ENCODING, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING,
|
HeaderValue, ACCEPT_ENCODING, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING,
|
||||||
};
|
};
|
||||||
use crate::http::{HeaderMap, Method, StatusCode, Version};
|
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::request::Request;
|
||||||
use crate::response::Response;
|
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.
|
// merging headers from head and extra headers. HeaderMap::new() does not allocate.
|
||||||
let empty_headers = HeaderMap::new();
|
let empty_headers = HeaderMap::new();
|
||||||
let extra_headers = self.extra_headers().unwrap_or(&empty_headers);
|
let extra_headers = self.extra_headers().unwrap_or(&empty_headers);
|
||||||
let headers = self.headers().inner.iter()
|
let headers = self
|
||||||
.filter(|(name, _)| {
|
.headers()
|
||||||
!extra_headers.contains_key(*name)
|
.inner
|
||||||
})
|
.iter()
|
||||||
|
.filter(|(name, _)| !extra_headers.contains_key(*name))
|
||||||
.chain(extra_headers.inner.iter());
|
.chain(extra_headers.inner.iter());
|
||||||
|
|
||||||
// write headers
|
// write headers
|
||||||
@ -604,10 +605,16 @@ mod tests {
|
|||||||
let mut bytes = BytesMut::with_capacity(2048);
|
let mut bytes = BytesMut::with_capacity(2048);
|
||||||
|
|
||||||
let mut head = RequestHead::default();
|
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();
|
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"));
|
extra_headers.insert(DATE, HeaderValue::from_static("date"));
|
||||||
|
|
||||||
let mut head = RequestHeadType::Rc(Rc::new(head), Some(extra_headers));
|
let mut head = RequestHeadType::Rc(Rc::new(head), Some(extra_headers));
|
||||||
|
@ -1,5 +1,6 @@
|
|||||||
# Changes
|
# 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
|
* Multipart handling now parses requests which do not end in CRLF #1038
|
||||||
|
|
||||||
|
@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-multipart"
|
name = "actix-multipart"
|
||||||
version = "0.1.3"
|
version = "0.1.4"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Multipart support for actix web framework."
|
description = "Multipart support for actix web framework."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
|
@ -178,13 +178,15 @@ impl InnerMultipart {
|
|||||||
Some(chunk) => {
|
Some(chunk) => {
|
||||||
if chunk.len() < boundary.len() + 4
|
if chunk.len() < boundary.len() + 4
|
||||||
|| &chunk[..2] != b"--"
|
|| &chunk[..2] != b"--"
|
||||||
|| &chunk[2..boundary.len() + 2] != boundary.as_bytes() {
|
|| &chunk[2..boundary.len() + 2] != boundary.as_bytes()
|
||||||
|
{
|
||||||
Err(MultipartError::Boundary)
|
Err(MultipartError::Boundary)
|
||||||
} else if &chunk[boundary.len() + 2..] == b"\r\n" {
|
} else if &chunk[boundary.len() + 2..] == b"\r\n" {
|
||||||
Ok(Some(false))
|
Ok(Some(false))
|
||||||
} else if &chunk[boundary.len() + 2..boundary.len() + 4] == b"--"
|
} else if &chunk[boundary.len() + 2..boundary.len() + 4] == b"--"
|
||||||
&& (chunk.len() == boundary.len() + 4
|
&& (chunk.len() == boundary.len() + 4
|
||||||
|| &chunk[boundary.len() + 4..] == b"\r\n") {
|
|| &chunk[boundary.len() + 4..] == b"\r\n")
|
||||||
|
{
|
||||||
Ok(Some(true))
|
Ok(Some(true))
|
||||||
} else {
|
} else {
|
||||||
Err(MultipartError::Boundary)
|
Err(MultipartError::Boundary)
|
||||||
@ -781,8 +783,10 @@ impl PayloadBuffer {
|
|||||||
/// Read bytes until new line delimiter or eof
|
/// Read bytes until new line delimiter or eof
|
||||||
pub fn readline_or_eof(&mut self) -> Result<Option<Bytes>, MultipartError> {
|
pub fn readline_or_eof(&mut self) -> Result<Option<Bytes>, MultipartError> {
|
||||||
match self.readline() {
|
match self.readline() {
|
||||||
Err(MultipartError::Incomplete) if self.eof => Ok(Some(self.buf.take().freeze())),
|
Err(MultipartError::Incomplete) if self.eof => {
|
||||||
line => line
|
Ok(Some(self.buf.take().freeze()))
|
||||||
|
}
|
||||||
|
line => line,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -859,14 +863,14 @@ mod tests {
|
|||||||
fn create_simple_request_with_header() -> (Bytes, HeaderMap) {
|
fn create_simple_request_with_header() -> (Bytes, HeaderMap) {
|
||||||
let bytes = Bytes::from(
|
let bytes = Bytes::from(
|
||||||
"testasdadsad\r\n\
|
"testasdadsad\r\n\
|
||||||
--abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
|
--abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
|
||||||
Content-Disposition: form-data; name=\"file\"; filename=\"fn.txt\"\r\n\
|
Content-Disposition: form-data; name=\"file\"; filename=\"fn.txt\"\r\n\
|
||||||
Content-Type: text/plain; charset=utf-8\r\nContent-Length: 4\r\n\r\n\
|
Content-Type: text/plain; charset=utf-8\r\nContent-Length: 4\r\n\r\n\
|
||||||
test\r\n\
|
test\r\n\
|
||||||
--abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
|
--abbc761f78ff4d7cb7573b5a23f96ef0\r\n\
|
||||||
Content-Type: text/plain; charset=utf-8\r\nContent-Length: 4\r\n\r\n\
|
Content-Type: text/plain; charset=utf-8\r\nContent-Length: 4\r\n\r\n\
|
||||||
data\r\n\
|
data\r\n\
|
||||||
--abbc761f78ff4d7cb7573b5a23f96ef0--\r\n"
|
--abbc761f78ff4d7cb7573b5a23f96ef0--\r\n",
|
||||||
);
|
);
|
||||||
let mut headers = HeaderMap::new();
|
let mut headers = HeaderMap::new();
|
||||||
headers.insert(
|
headers.insert(
|
||||||
|
@ -1,14 +1,18 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
##
|
## [0.2.6] - 2019-09-12
|
||||||
|
|
||||||
|
### Added
|
||||||
|
|
||||||
|
* Export frozen request related types.
|
||||||
|
|
||||||
|
|
||||||
|
## [0.2.5] - 2019-09-11
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
|
|
||||||
* Add `FrozenClientRequest` to support retries for sending HTTP requests
|
* Add `FrozenClientRequest` to support retries for sending HTTP requests
|
||||||
|
|
||||||
|
|
||||||
## [0.2.5] - 2019-09-06
|
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
|
||||||
* Ensure that the `Host` header is set when initiating a WebSocket client connection.
|
* Ensure that the `Host` header is set when initiating a WebSocket client connection.
|
||||||
|
@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "awc"
|
name = "awc"
|
||||||
version = "0.2.4"
|
version = "0.2.6"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix http client."
|
description = "Actix http client."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
@ -44,7 +44,7 @@ flate2-rust = ["actix-http/flate2-rust"]
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
actix-codec = "0.1.2"
|
actix-codec = "0.1.2"
|
||||||
actix-service = "0.4.1"
|
actix-service = "0.4.1"
|
||||||
actix-http = "0.2.9"
|
actix-http = "0.2.10"
|
||||||
base64 = "0.10.1"
|
base64 = "0.10.1"
|
||||||
bytes = "0.4"
|
bytes = "0.4"
|
||||||
derive_more = "0.15.0"
|
derive_more = "0.15.0"
|
||||||
@ -63,7 +63,7 @@ rustls = { version = "0.15.2", optional = true }
|
|||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-rt = "0.2.2"
|
actix-rt = "0.2.2"
|
||||||
actix-web = { version = "1.0.0", features=["ssl"] }
|
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-http-test = { version = "0.2.0", features=["ssl"] }
|
||||||
actix-utils = "0.4.1"
|
actix-utils = "0.4.1"
|
||||||
actix-server = { version = "0.6.0", features=["ssl", "rust-tls"] }
|
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::rc::Rc;
|
||||||
|
use std::{fmt, io, net};
|
||||||
|
|
||||||
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
use actix_codec::{AsyncRead, AsyncWrite, Framed};
|
||||||
use actix_http::body::Body;
|
use actix_http::body::Body;
|
||||||
@ -7,8 +7,8 @@ use actix_http::client::{
|
|||||||
Connect as ClientConnect, ConnectError, Connection, SendRequestError,
|
Connect as ClientConnect, ConnectError, Connection, SendRequestError,
|
||||||
};
|
};
|
||||||
use actix_http::h1::ClientCodec;
|
use actix_http::h1::ClientCodec;
|
||||||
use actix_http::{RequestHead, RequestHeadType, ResponseHead};
|
|
||||||
use actix_http::http::HeaderMap;
|
use actix_http::http::HeaderMap;
|
||||||
|
use actix_http::{RequestHead, RequestHeadType, ResponseHead};
|
||||||
use actix_service::Service;
|
use actix_service::Service;
|
||||||
use futures::{Future, Poll};
|
use futures::{Future, Poll};
|
||||||
|
|
||||||
@ -82,7 +82,9 @@ where
|
|||||||
})
|
})
|
||||||
.from_err()
|
.from_err()
|
||||||
// send request
|
// 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)),
|
.map(|(head, payload)| ClientResponse::new(head, payload)),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@ -103,7 +105,10 @@ where
|
|||||||
})
|
})
|
||||||
.from_err()
|
.from_err()
|
||||||
// send request
|
// 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)),
|
.map(|(head, payload)| ClientResponse::new(head, payload)),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@ -127,7 +132,9 @@ where
|
|||||||
})
|
})
|
||||||
.from_err()
|
.from_err()
|
||||||
// send request
|
// 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)| {
|
.map(|(head, framed)| {
|
||||||
let framed = framed.map_io(|io| BoxedSocket(Box::new(Socket(io))));
|
let framed = framed.map_io(|io| BoxedSocket(Box::new(Socket(io))));
|
||||||
(head, framed)
|
(head, framed)
|
||||||
@ -155,7 +162,9 @@ where
|
|||||||
})
|
})
|
||||||
.from_err()
|
.from_err()
|
||||||
// send request
|
// 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)| {
|
.map(|(head, framed)| {
|
||||||
let framed = framed.map_io(|io| BoxedSocket(Box::new(Socket(io))));
|
let framed = framed.map_io(|io| BoxedSocket(Box::new(Socket(io))));
|
||||||
(head, framed)
|
(head, framed)
|
||||||
|
@ -1,5 +1,7 @@
|
|||||||
//! Http client errors
|
//! 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::error::PayloadError;
|
||||||
pub use actix_http::ws::HandshakeError as WsHandshakeError;
|
pub use actix_http::ws::HandshakeError as WsHandshakeError;
|
||||||
pub use actix_http::ws::ProtocolError as WsProtocolError;
|
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 builder;
|
||||||
mod connect;
|
mod connect;
|
||||||
pub mod error;
|
pub mod error;
|
||||||
|
mod frozen;
|
||||||
mod request;
|
mod request;
|
||||||
mod response;
|
mod response;
|
||||||
|
mod sender;
|
||||||
pub mod test;
|
pub mod test;
|
||||||
pub mod ws;
|
pub mod ws;
|
||||||
|
|
||||||
pub use self::builder::ClientBuilder;
|
pub use self::builder::ClientBuilder;
|
||||||
pub use self::connect::BoxedSocket;
|
pub use self::connect::BoxedSocket;
|
||||||
|
pub use self::frozen::{FrozenClientRequest, FrozenSendBuilder};
|
||||||
pub use self::request::ClientRequest;
|
pub use self::request::ClientRequest;
|
||||||
pub use self::response::{ClientResponse, JsonBody, MessageBody};
|
pub use self::response::{ClientResponse, JsonBody, MessageBody};
|
||||||
|
pub use self::sender::SendClientRequest;
|
||||||
|
|
||||||
use self::connect::{Connect, ConnectorWrapper};
|
use self::connect::{Connect, ConnectorWrapper};
|
||||||
|
|
||||||
|
@ -1,29 +1,26 @@
|
|||||||
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, Instant};
|
use std::time::Duration;
|
||||||
use std::{fmt, net};
|
use std::{fmt, net};
|
||||||
|
|
||||||
use bytes::{BufMut, Bytes, BytesMut};
|
use bytes::{BufMut, Bytes, BytesMut};
|
||||||
use futures::{Async, Future, Poll, Stream, try_ready};
|
use futures::Stream;
|
||||||
use percent_encoding::percent_encode;
|
use percent_encoding::percent_encode;
|
||||||
use serde::Serialize;
|
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::cookie::{Cookie, CookieJar, USERINFO};
|
||||||
use actix_http::encoding::Decoder;
|
use actix_http::http::header::{self, Header, IntoHeaderValue};
|
||||||
use actix_http::http::header::{self, ContentEncoding, Header, IntoHeaderValue};
|
|
||||||
use actix_http::http::{
|
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, PayloadStream, RequestHead};
|
use actix_http::{Error, RequestHead};
|
||||||
|
|
||||||
use crate::error::{InvalidUrl, SendRequestError, FreezeRequestError};
|
use crate::error::{FreezeRequestError, InvalidUrl};
|
||||||
use crate::response::ClientResponse;
|
use crate::frozen::FrozenClientRequest;
|
||||||
|
use crate::sender::{PrepForSendingError, RequestSender, SendClientRequest};
|
||||||
use crate::ClientConfig;
|
use crate::ClientConfig;
|
||||||
|
|
||||||
#[cfg(any(feature = "brotli", feature = "flate2-zlib", feature = "flate2-rust"))]
|
#[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> {
|
pub fn freeze(self) -> Result<FrozenClientRequest, FreezeRequestError> {
|
||||||
let slf = match self.prep_for_sending() {
|
let slf = match self.prep_for_sending() {
|
||||||
Ok(slf) => slf,
|
Ok(slf) => slf,
|
||||||
@ -393,10 +392,7 @@ impl ClientRequest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Complete request construction and send body.
|
/// Complete request construction and send body.
|
||||||
pub fn send_body<B>(
|
pub fn send_body<B>(self, body: B) -> SendClientRequest
|
||||||
self,
|
|
||||||
body: B,
|
|
||||||
) -> SendBody
|
|
||||||
where
|
where
|
||||||
B: Into<Body>,
|
B: Into<Body>,
|
||||||
{
|
{
|
||||||
@ -405,47 +401,51 @@ impl ClientRequest {
|
|||||||
Err(e) => return e.into(),
|
Err(e) => return e.into(),
|
||||||
};
|
};
|
||||||
|
|
||||||
RequestSender::Owned(slf.head)
|
RequestSender::Owned(slf.head).send_body(
|
||||||
.send_body(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), 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, value: &T) -> SendClientRequest {
|
||||||
self,
|
|
||||||
value: &T,
|
|
||||||
) -> SendBody
|
|
||||||
{
|
|
||||||
let slf = match self.prep_for_sending() {
|
let slf = match self.prep_for_sending() {
|
||||||
Ok(slf) => slf,
|
Ok(slf) => slf,
|
||||||
Err(e) => return e.into(),
|
Err(e) => return e.into(),
|
||||||
};
|
};
|
||||||
|
|
||||||
RequestSender::Owned(slf.head)
|
RequestSender::Owned(slf.head).send_json(
|
||||||
.send_json(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), value)
|
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`
|
||||||
///
|
///
|
||||||
/// `ClientRequestBuilder` can not be used after this call.
|
/// `ClientRequestBuilder` can not be used after this call.
|
||||||
pub fn send_form<T: Serialize>(
|
pub fn send_form<T: Serialize>(self, value: &T) -> SendClientRequest {
|
||||||
self,
|
|
||||||
value: &T,
|
|
||||||
) -> SendBody
|
|
||||||
{
|
|
||||||
let slf = match self.prep_for_sending() {
|
let slf = match self.prep_for_sending() {
|
||||||
Ok(slf) => slf,
|
Ok(slf) => slf,
|
||||||
Err(e) => return e.into(),
|
Err(e) => return e.into(),
|
||||||
};
|
};
|
||||||
|
|
||||||
RequestSender::Owned(slf.head)
|
RequestSender::Owned(slf.head).send_form(
|
||||||
.send_form(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), value)
|
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, stream: S) -> SendClientRequest
|
||||||
self,
|
|
||||||
stream: S,
|
|
||||||
) -> SendBody
|
|
||||||
where
|
where
|
||||||
S: Stream<Item = Bytes, Error = E> + 'static,
|
S: Stream<Item = Bytes, Error = E> + 'static,
|
||||||
E: Into<Error> + 'static,
|
E: Into<Error> + 'static,
|
||||||
@ -455,22 +455,28 @@ impl ClientRequest {
|
|||||||
Err(e) => return e.into(),
|
Err(e) => return e.into(),
|
||||||
};
|
};
|
||||||
|
|
||||||
RequestSender::Owned(slf.head)
|
RequestSender::Owned(slf.head).send_stream(
|
||||||
.send_stream(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref(), 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) -> SendClientRequest {
|
||||||
self,
|
|
||||||
) -> SendBody
|
|
||||||
{
|
|
||||||
let slf = match self.prep_for_sending() {
|
let slf = match self.prep_for_sending() {
|
||||||
Ok(slf) => slf,
|
Ok(slf) => slf,
|
||||||
Err(e) => return e.into(),
|
Err(e) => return e.into(),
|
||||||
};
|
};
|
||||||
|
|
||||||
RequestSender::Owned(slf.head)
|
RequestSender::Owned(slf.head).send(
|
||||||
.send(slf.addr, slf.response_decompress, slf.timeout, slf.config.as_ref())
|
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> {
|
||||||
@ -528,10 +534,10 @@ impl ClientRequest {
|
|||||||
slf = slf.set_header_if_none(header::ACCEPT_ENCODING, HTTPS_ENCODING)
|
slf = slf.set_header_if_none(header::ACCEPT_ENCODING, HTTPS_ENCODING)
|
||||||
} else {
|
} else {
|
||||||
#[cfg(any(feature = "flate2-zlib", feature = "flate2-rust"))]
|
#[cfg(any(feature = "flate2-zlib", feature = "flate2-rust"))]
|
||||||
{
|
{
|
||||||
slf = slf
|
slf = slf
|
||||||
.set_header_if_none(header::ACCEPT_ENCODING, "gzip, deflate")
|
.set_header_if_none(header::ACCEPT_ENCODING, "gzip, deflate")
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use std::time::SystemTime;
|
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) {
|
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
|
// set cookies
|
||||||
|
@ -169,7 +169,7 @@ where
|
|||||||
match self.data_factories_fut[idx].poll()? {
|
match self.data_factories_fut[idx].poll()? {
|
||||||
Async::Ready(f) => {
|
Async::Ready(f) => {
|
||||||
self.data_factories.push(f);
|
self.data_factories.push(f);
|
||||||
self.data_factories_fut.remove(idx);
|
let _ = self.data_factories_fut.remove(idx);
|
||||||
}
|
}
|
||||||
Async::NotReady => idx += 1,
|
Async::NotReady => idx += 1,
|
||||||
}
|
}
|
||||||
|
@ -2,13 +2,13 @@
|
|||||||
mod compress;
|
mod compress;
|
||||||
pub use self::compress::{BodyEncoding, Compress};
|
pub use self::compress::{BodyEncoding, Compress};
|
||||||
|
|
||||||
|
mod condition;
|
||||||
mod defaultheaders;
|
mod defaultheaders;
|
||||||
pub mod errhandlers;
|
pub mod errhandlers;
|
||||||
mod logger;
|
mod logger;
|
||||||
mod normalize;
|
mod normalize;
|
||||||
mod condition;
|
|
||||||
|
|
||||||
|
pub use self::condition::Condition;
|
||||||
pub use self::defaultheaders::DefaultHeaders;
|
pub use self::defaultheaders::DefaultHeaders;
|
||||||
pub use self::logger::Logger;
|
pub use self::logger::Logger;
|
||||||
pub use self::normalize::NormalizePath;
|
pub use self::normalize::NormalizePath;
|
||||||
pub use self::condition::Condition;
|
|
||||||
|
Reference in New Issue
Block a user