actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-06-25 06:39:22 +02:00
Code Issues Releases Wiki Activity

Added FrozenClientRequest for easier retrying HTTP calls (#1064)

Browse Source 
* Initial commit

* Added extra_headers

* Added freeze() method to ClientRequest which produces a 'read-only' copy of a request suitable for retrying the send operation

* Additional methods for FrozenClientRequest

* Fix

* Increased crates versions

* Fixed a unit test. Added one more unit test.

* Added RequestHeaderWrapper

* Small fixes

* Renamed RequestHeadWrapper->RequestHeadType

* Updated CHANGES.md files

* Small fix

* Small changes

* Removed *_extra methods from Connection trait

* Added FrozenSendBuilder

* Added FrozenSendBuilder

* Minor fix

* Replaced impl Future with concrete Future implementation

* Small renaming

* Renamed Send->SendBody
This commit is contained in:
Dmitry Pypin
2019-09-09 21:29:32 -07:00
committed by Nikolay Kim
parent 5e8f1c338c
commit 8873e9b39e
14 changed files with 828 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
actix-http/CHANGES.md
View File

@ -1,5 +1,12 @@
# Changes
##
### Added
* Add support for sending HTTP requests with `Rc<RequestHead>` in addition to sending HTTP requests with `RequestHead`
## [0.2.10] - 2019-09-xx
### Fixed

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

@ -8,7 +8,7 @@ use h2::client::SendRequest;
use crate::body::MessageBody;
use crate::h1::ClientCodec;
use crate::message::{RequestHead, ResponseHead};
use crate::message::{RequestHeadType, ResponseHead};
use crate::payload::Payload;
use super::error::SendRequestError;
@ -27,9 +27,9 @@ pub trait Connection {
fn protocol(&self) -> Protocol;
/// Send request and body
fn send_request<B: MessageBody + 'static>(
fn send_request<B: MessageBody + 'static, H: Into<RequestHeadType>>(
self,
head: RequestHead,
head: H,
body: B,
) -> Self::Future;
@ -39,7 +39,7 @@ pub trait Connection {
>;
/// Send request, returns Response and Framed
fn open_tunnel(self, head: RequestHead) -> Self::TunnelFuture;
fn open_tunnel<H: Into<RequestHeadType>>(self, head: H) -> Self::TunnelFuture;
}
pub(crate) trait ConnectionLifetime: AsyncRead + AsyncWrite + 'static {
@ -105,22 +105,22 @@ where
}
}
fn send_request<B: MessageBody + 'static>(
fn send_request<B: MessageBody + 'static, H: Into<RequestHeadType>>(
mut self,
head: RequestHead,
head: H,
body: B,
) -> Self::Future {
match self.io.take().unwrap() {
ConnectionType::H1(io) => Box::new(h1proto::send_request(
io,
head,
head.into(),
body,
self.created,
self.pool,
)),
ConnectionType::H2(io) => Box::new(h2proto::send_request(
io,
head,
head.into(),
body,
self.created,
self.pool,
@ -139,10 +139,10 @@ where
>;
/// Send request, returns Response and Framed
fn open_tunnel(mut self, head: RequestHead) -> Self::TunnelFuture {
fn open_tunnel<H: Into<RequestHeadType>>(mut self, head: H) -> Self::TunnelFuture {
match self.io.take().unwrap() {
ConnectionType::H1(io) => {
Either::A(Box::new(h1proto::open_tunnel(io, head)))
Either::A(Box::new(h1proto::open_tunnel(io, head.into())))
}
ConnectionType::H2(io) => {
if let Some(mut pool) = self.pool.take() {
@ -180,9 +180,9 @@ where
}
}
fn send_request<RB: MessageBody + 'static>(
fn send_request<RB: MessageBody + 'static, H: Into<RequestHeadType>>(
self,
head: RequestHead,
head: H,
body: RB,
) -> Self::Future {
match self {
@ -199,7 +199,7 @@ where
>;
/// Send request, returns Response and Framed
fn open_tunnel(self, head: RequestHead) -> Self::TunnelFuture {
fn open_tunnel<H: Into<RequestHeadType>>(self, head: H) -> Self::TunnelFuture {
match self {
EitherConnection::A(con) => Box::new(
con.open_tunnel(head)

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

@ -128,3 +128,23 @@ impl ResponseError for SendRequestError {
.into()
}
}
/// A set of errors that can occur during freezing a request
#[derive(Debug, Display, From)]
pub enum FreezeRequestError {
/// Invalid URL
#[display(fmt = "Invalid URL: {}", _0)]
Url(InvalidUrl),
/// Http error
#[display(fmt = "{}", _0)]
Http(HttpError),
}
impl From<FreezeRequestError> for SendRequestError {
fn from(e: FreezeRequestError) -> Self {
match e {
FreezeRequestError::Url(e) => e.into(),
FreezeRequestError::Http(e) => e.into(),
}
}
}

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

@ -9,8 +9,9 @@ use futures::{Async, Future, Poll, Sink, Stream};
use crate::error::PayloadError;
use crate::h1;
use crate::http::header::{IntoHeaderValue, HOST};
use crate::message::{RequestHead, ResponseHead};
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};
@ -19,7 +20,7 @@ use crate::body::{BodySize, MessageBody};
pub(crate) fn send_request<T, B>(
io: T,
mut head: RequestHead,
mut head: RequestHeadType,
body: B,
created: time::Instant,
pool: Option<Acquired<T>>,
@ -29,21 +30,29 @@ where
B: MessageBody,
{
// set request host header
if !head.headers.contains_key(HOST) {
if let Some(host) = head.uri.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();
let _ = match head.uri.port_u16() {
let _ = match head.as_ref().uri.port_u16() {
None | Some(80) | Some(443) => write!(wrt, "{}", host),
Some(port) => write!(wrt, "{}:{}", host, port),
};
match wrt.get_mut().take().freeze().try_into() {
Ok(value) => {
head.headers.insert(HOST, 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);
log::error!("Can not set HOST header {}", e)
}
}
}
@ -57,7 +66,7 @@ where
let len = body.size();
// create Framed and send reqest
// create Framed and send request
Framed::new(io, h1::ClientCodec::default())
.send((head, len).into())
.from_err()
@ -95,12 +104,12 @@ where
pub(crate) fn open_tunnel<T>(
io: T,
head: RequestHead,
head: RequestHeadType,
) -> impl Future<Item = (ResponseHead, Framed<T, h1::ClientCodec>), Error = SendRequestError>
where
T: AsyncRead + AsyncWrite + 'static,
{
// create Framed and send reqest
// create Framed and send request
Framed::new(io, h1::ClientCodec::default())
.send((head, BodySize::None).into())
.from_err()

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

@ -9,8 +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::message::{RequestHead, ResponseHead};
use crate::message::{RequestHeadType, ResponseHead};
use crate::payload::Payload;
use crate::header::HeaderMap;
use super::connection::{ConnectionType, IoConnection};
use super::error::SendRequestError;
@ -18,7 +19,7 @@ use super::pool::Acquired;
pub(crate) fn send_request<T, B>(
io: SendRequest<Bytes>,
head: RequestHead,
head: RequestHeadType,
body: B,
created: time::Instant,
pool: Option<Acquired<T>>,
@ -28,7 +29,7 @@ where
B: MessageBody,
{
trace!("Sending client request: {:?} {:?}", head, body.size());
let head_req = head.method == Method::HEAD;
let head_req = head.as_ref().method == Method::HEAD;
let length = body.size();
let eof = match length {
BodySize::None | BodySize::Empty | BodySize::Sized(0) => true,
@ -39,8 +40,8 @@ where
.map_err(SendRequestError::from)
.and_then(move |mut io| {
let mut req = Request::new(());
*req.uri_mut() = head.uri;
*req.method_mut() = head.method;
*req.uri_mut() = head.as_ref().uri.clone();
*req.method_mut() = head.as_ref().method.clone();
*req.version_mut() = Version::HTTP_2;
let mut skip_len = true;
@ -66,8 +67,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())),
};
// merging headers from head and extra headers.
let headers = head.as_ref().headers.iter()
.filter(|(name, _)| {
!extra_headers.contains_key(*name)
})
.chain(extra_headers.iter());
// copy headers
for (key, value) in head.headers.iter() {
for (key, value) in headers {
match *key {
CONNECTION | TRANSFER_ENCODING => continue, // http2 specific
CONTENT_LENGTH if skip_len => continue,

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

@ -10,7 +10,7 @@ mod pool;
pub use self::connection::Connection;
pub use self::connector::Connector;
pub use self::error::{ConnectError, InvalidUrl, SendRequestError};
pub use self::error::{ConnectError, InvalidUrl, SendRequestError, FreezeRequestError};
pub use self::pool::Protocol;
#[derive(Clone)]

18
actix-http/src/h1/client.rs
View File

@ -1,5 +1,6 @@
#![allow(unused_imports, unused_variables, dead_code)]
use std::io::{self, Write};
use std::rc::Rc;
use actix_codec::{Decoder, Encoder};
use bitflags::bitflags;
@ -16,7 +17,8 @@ use crate::body::BodySize;
use crate::config::ServiceConfig;
use crate::error::{ParseError, PayloadError};
use crate::helpers;
use crate::message::{ConnectionType, Head, MessagePool, RequestHead, ResponseHead};
use crate::message::{ConnectionType, Head, MessagePool, RequestHead, RequestHeadType, ResponseHead};
use crate::header::HeaderMap;
bitflags! {
struct Flags: u8 {
@ -48,7 +50,7 @@ struct ClientCodecInner {
// encoder part
flags: Flags,
headers_size: u32,
encoder: encoder::MessageEncoder<RequestHead>,
encoder: encoder::MessageEncoder<RequestHeadType>,
}
impl Default for ClientCodec {
@ -183,7 +185,7 @@ impl Decoder for ClientPayloadCodec {
}
impl Encoder for ClientCodec {
type Item = Message<(RequestHead, BodySize)>;
type Item = Message<(RequestHeadType, BodySize)>;
type Error = io::Error;
fn encode(
@ -192,13 +194,13 @@ impl Encoder for ClientCodec {
dst: &mut BytesMut,
) -> Result<(), Self::Error> {
match item {
Message::Item((mut msg, length)) => {
Message::Item((mut head, length)) => {
let inner = &mut self.inner;
inner.version = msg.version;
inner.flags.set(Flags::HEAD, msg.method == Method::HEAD);
inner.version = head.as_ref().version;
inner.flags.set(Flags::HEAD, head.as_ref().method == Method::HEAD);
// connection status
inner.ctype = match msg.connection_type() {
inner.ctype = match head.as_ref().connection_type() {
ConnectionType::KeepAlive => {
if inner.flags.contains(Flags::KEEPALIVE_ENABLED) {
ConnectionType::KeepAlive
@ -212,7 +214,7 @@ impl Encoder for ClientCodec {
inner.encoder.encode(
dst,
&mut msg,
&mut head,
false,
false,
inner.version,

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

@ -4,6 +4,7 @@ use std::io::Write;
use std::marker::PhantomData;
use std::str::FromStr;
use std::{cmp, fmt, io, mem};
use std::rc::Rc;
use bytes::{BufMut, Bytes, BytesMut};
@ -15,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};
use crate::message::{ConnectionType, Head, RequestHead, ResponseHead, RequestHeadType};
use crate::request::Request;
use crate::response::Response;
@ -43,6 +44,8 @@ pub(crate) trait MessageType: Sized {
fn headers(&self) -> &HeaderMap;
fn extra_headers(&self) -> Option<&HeaderMap>;
fn camel_case(&self) -> bool {
false
}
@ -128,12 +131,21 @@ 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)
})
.chain(extra_headers.inner.iter());
// write headers
let mut pos = 0;
let mut has_date = false;
let mut remaining = dst.remaining_mut();
let mut buf = unsafe { &mut *(dst.bytes_mut() as *mut [u8]) };
for (key, value) in self.headers().inner.iter() {
for (key, value) in headers {
match *key {
CONNECTION => continue,
TRANSFER_ENCODING | CONTENT_LENGTH if skip_len => continue,
@ -235,6 +247,10 @@ impl MessageType for Response<()> {
&self.head().headers
}
fn extra_headers(&self) -> Option<&HeaderMap> {
None
}
fn encode_status(&mut self, dst: &mut BytesMut) -> io::Result<()> {
let head = self.head();
let reason = head.reason().as_bytes();
@ -247,31 +263,36 @@ impl MessageType for Response<()> {
}
}
impl MessageType for RequestHead {
impl MessageType for RequestHeadType {
fn status(&self) -> Option<StatusCode> {
None
}
fn chunked(&self) -> bool {
self.chunked()
self.as_ref().chunked()
}
fn camel_case(&self) -> bool {
RequestHead::camel_case_headers(self)
self.as_ref().camel_case_headers()
}
fn headers(&self) -> &HeaderMap {
&self.headers
self.as_ref().headers()
}
fn extra_headers(&self) -> Option<&HeaderMap> {
self.extra_headers()
}
fn encode_status(&mut self, dst: &mut BytesMut) -> io::Result<()> {
dst.reserve(256 + self.headers.len() * AVERAGE_HEADER_SIZE);
let head = self.as_ref();
dst.reserve(256 + head.headers.len() * AVERAGE_HEADER_SIZE);
write!(
Writer(dst),
"{} {} {}",
self.method,
self.uri.path_and_query().map(|u| u.as_str()).unwrap_or("/"),
match self.version {
head.method,
head.uri.path_and_query().map(|u| u.as_str()).unwrap_or("/"),
match head.version {
Version::HTTP_09 => "HTTP/0.9",
Version::HTTP_10 => "HTTP/1.0",
Version::HTTP_11 => "HTTP/1.1",
@ -488,9 +509,11 @@ fn write_camel_case(value: &[u8], buffer: &mut [u8]) {
#[cfg(test)]
mod tests {
use bytes::Bytes;
//use std::rc::Rc;
use super::*;
use crate::http::header::{HeaderValue, CONTENT_TYPE};
use http::header::AUTHORIZATION;
#[test]
fn test_chunked_te() {
@ -515,6 +538,8 @@ mod tests {
head.headers
.insert(CONTENT_TYPE, HeaderValue::from_static("plain/text"));
let mut head = RequestHeadType::Owned(head);
let _ = head.encode_headers(
&mut bytes,
Version::HTTP_11,
@ -551,21 +576,16 @@ mod tests {
Bytes::from_static(b"\r\nContent-Length: 100\r\nDate: date\r\nContent-Type: plain/text\r\n\r\n")
);
let mut head = RequestHead::default();
head.set_camel_case_headers(false);
head.headers.insert(DATE, HeaderValue::from_static("date"));
head.headers
.insert(CONTENT_TYPE, HeaderValue::from_static("plain/text"));
head.headers
.append(CONTENT_TYPE, HeaderValue::from_static("xml"));
let _ = head.encode_headers(
&mut bytes,
Version::HTTP_11,
BodySize::Stream,
ConnectionType::KeepAlive,
&ServiceConfig::default(),
);
assert_eq!(
bytes.take().freeze(),
Bytes::from_static(b"\r\nTransfer-Encoding: chunked\r\nDate: date\r\nContent-Type: xml\r\nContent-Type: plain/text\r\n\r\n")
);
head.set_camel_case_headers(false);
let mut head = RequestHeadType::Owned(head);
let _ = head.encode_headers(
&mut bytes,
Version::HTTP_11,
@ -578,4 +598,30 @@ mod tests {
Bytes::from_static(b"\r\ntransfer-encoding: chunked\r\ndate: date\r\ncontent-type: xml\r\ncontent-type: plain/text\r\n\r\n")
);
}
#[test]
fn test_extra_headers() {
let mut bytes = BytesMut::with_capacity(2048);
let mut head = RequestHead::default();
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(DATE, HeaderValue::from_static("date"));
let mut head = RequestHeadType::Rc(Rc::new(head), Some(extra_headers));
let _ = head.encode_headers(
&mut bytes,
Version::HTTP_11,
BodySize::Empty,
ConnectionType::Close,
&ServiceConfig::default(),
);
assert_eq!(
bytes.take().freeze(),
Bytes::from_static(b"\r\ncontent-length: 0\r\nconnection: close\r\nauthorization: another authorization\r\ndate: date\r\n\r\n")
);
}
}

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

@ -39,7 +39,7 @@ pub use self::config::{KeepAlive, ServiceConfig};
pub use self::error::{Error, ResponseError, Result};
pub use self::extensions::Extensions;
pub use self::httpmessage::HttpMessage;
pub use self::message::{Message, RequestHead, ResponseHead};
pub use self::message::{Message, RequestHead, RequestHeadType, ResponseHead};
pub use self::payload::{Payload, PayloadStream};
pub use self::request::Request;
pub use self::response::{Response, ResponseBuilder};

30
actix-http/src/message.rs
View File

@ -181,6 +181,36 @@ impl RequestHead {
}
}
#[derive(Debug)]
pub enum RequestHeadType {
Owned(RequestHead),
Rc(Rc<RequestHead>, Option<HeaderMap>),
}
impl RequestHeadType {
pub fn extra_headers(&self) -> Option<&HeaderMap> {
match self {
RequestHeadType::Owned(_) => None,
RequestHeadType::Rc(_, headers) => headers.as_ref(),
}
}
}
impl AsRef<RequestHead> for RequestHeadType {
fn as_ref(&self) -> &RequestHead {
match self {
RequestHeadType::Owned(head) => &head,
RequestHeadType::Rc(head, _) => head.as_ref(),
}
}
}
impl From<RequestHead> for RequestHeadType {
fn from(head: RequestHead) -> Self {
RequestHeadType::Owned(head)
}
}
#[derive(Debug)]
pub struct ResponseHead {
pub version: Version,