mirror of
https://github.com/fafhrd91/actix-web
synced 2025-02-22 04:03:17 +01:00
Improved readability and logic
This commit is contained in:
joshbenaron
parent
58904b9ebc
commit
22e40613e8
@ -60,8 +60,8 @@ impl Retry {
|
|||||||
/// .finish();
|
/// .finish();
|
||||||
///```
|
///```
|
||||||
pub fn policy<T>(mut self, p: T) -> Self
|
pub fn policy<T>(mut self, p: T) -> Self
|
||||||
where
|
where
|
||||||
T: IntoRetryPolicy,
|
T: IntoRetryPolicy,
|
||||||
{
|
{
|
||||||
self.0.policies.push(p.into_policy());
|
self.0.policies.push(p.into_policy());
|
||||||
self
|
self
|
||||||
@ -79,8 +79,8 @@ pub trait IntoRetryPolicy {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl<T> IntoRetryPolicy for T
|
impl<T> IntoRetryPolicy for T
|
||||||
where
|
where
|
||||||
T: for<'a> Fn(StatusCode, &'a HeaderMap) -> bool + 'static,
|
T: for<'a> Fn(StatusCode, &'a HeaderMap) -> bool + 'static,
|
||||||
{
|
{
|
||||||
fn into_policy(self) -> RetryPolicy {
|
fn into_policy(self) -> RetryPolicy {
|
||||||
RetryPolicy::Custom(Box::new(self))
|
RetryPolicy::Custom(Box::new(self))
|
||||||
@ -94,8 +94,8 @@ impl IntoRetryPolicy for Vec<StatusCode> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl<S> Transform<S, ConnectRequest> for Retry
|
impl<S> Transform<S, ConnectRequest> for Retry
|
||||||
where
|
where
|
||||||
S: Service<ConnectRequest, Response = ConnectResponse, Error = SendRequestError> + 'static,
|
S: Service<ConnectRequest, Response=ConnectResponse, Error=SendRequestError> + 'static,
|
||||||
{
|
{
|
||||||
type Transform = RetryService<S>;
|
type Transform = RetryService<S>;
|
||||||
|
|
||||||
@ -116,8 +116,8 @@ pub struct RetryService<S> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl<S> Service<ConnectRequest> for RetryService<S>
|
impl<S> Service<ConnectRequest> for RetryService<S>
|
||||||
where
|
where
|
||||||
S: Service<ConnectRequest, Response = ConnectResponse, Error = SendRequestError> + 'static,
|
S: Service<ConnectRequest, Response=ConnectResponse, Error=SendRequestError> + 'static,
|
||||||
{
|
{
|
||||||
type Response = S::Response;
|
type Response = S::Response;
|
||||||
type Error = S::Error;
|
type Error = S::Error;
|
||||||
@ -133,237 +133,87 @@ where
|
|||||||
let max_retry = self.max_retry;
|
let max_retry = self.max_retry;
|
||||||
|
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
let mut tries = 0;
|
|
||||||
match req {
|
match req {
|
||||||
ConnectRequest::Client(head, body, addr) => {
|
ConnectRequest::Client(head, body, addr) => {
|
||||||
match body {
|
for _ in 1..max_retry {
|
||||||
Body::Bytes(b) => {
|
let h = clone_request_head_type(&head);
|
||||||
loop {
|
|
||||||
let h = clone_request_head_type(&head);
|
|
||||||
|
|
||||||
match connector
|
let result = connector
|
||||||
.call(ConnectRequest::Client(
|
.call(ConnectRequest::Client(h, body_to_retry_body(&body), addr))
|
||||||
h,
|
.await;
|
||||||
Body::Bytes(b.clone()),
|
|
||||||
addr,
|
|
||||||
))
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(res) => {
|
|
||||||
// ConnectResponse
|
|
||||||
match &res {
|
|
||||||
ConnectResponse::Client(ref r) => {
|
|
||||||
if is_valid_response(
|
|
||||||
policies.as_ref(),
|
|
||||||
r.status(),
|
|
||||||
r.headers(),
|
|
||||||
) {
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
if tries == max_retry {
|
if let Ok(res) = result {
|
||||||
return Ok(res);
|
match &res {
|
||||||
}
|
ConnectResponse::Client(ref r) => {
|
||||||
|
if is_valid_response(
|
||||||
tries += 1;
|
policies.as_ref(),
|
||||||
}
|
r.status(),
|
||||||
ConnectResponse::Tunnel(ref head, _) => {
|
r.headers(),
|
||||||
if is_valid_response(
|
) {
|
||||||
policies.as_ref(),
|
return Ok(res);
|
||||||
head.status,
|
|
||||||
head.headers(),
|
|
||||||
) {
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
if tries == max_retry {
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// SendRequestError
|
|
||||||
Err(e) => {
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Err(e);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
ConnectResponse::Tunnel(ref head, _) => {
|
||||||
}
|
if is_valid_response(
|
||||||
Body::Empty => {
|
policies.as_ref(),
|
||||||
loop {
|
head.status,
|
||||||
let h = clone_request_head_type(&head);
|
head.headers(),
|
||||||
|
) {
|
||||||
match connector
|
return Ok(res);
|
||||||
.call(ConnectRequest::Client(h, Body::Empty, addr))
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(res) => {
|
|
||||||
// ConnectResponse
|
|
||||||
match &res {
|
|
||||||
ConnectResponse::Client(ref r) => {
|
|
||||||
if is_valid_response(
|
|
||||||
policies.as_ref(),
|
|
||||||
r.status(),
|
|
||||||
r.headers(),
|
|
||||||
) {
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
ConnectResponse::Tunnel(ref head, _) => {
|
|
||||||
if is_valid_response(
|
|
||||||
policies.as_ref(),
|
|
||||||
head.status,
|
|
||||||
head.headers(),
|
|
||||||
) {
|
|
||||||
return Ok(res);
|
|
||||||
} else {
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!(
|
|
||||||
"Request max retry reached"
|
|
||||||
);
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// SendRequestError
|
|
||||||
Err(e) => {
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Err(e);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_ => {
|
|
||||||
log::debug!(
|
|
||||||
"Non cloneable body type given - defaulting to `Body::None`"
|
|
||||||
);
|
|
||||||
loop {
|
|
||||||
let h = clone_request_head_type(&head);
|
|
||||||
|
|
||||||
match connector
|
|
||||||
.call(ConnectRequest::Client(h, Body::None, addr))
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
Ok(res) => {
|
|
||||||
// ConnectResponse
|
|
||||||
match &res {
|
|
||||||
ConnectResponse::Client(ref r) => {
|
|
||||||
if is_valid_response(
|
|
||||||
policies.as_ref(),
|
|
||||||
r.status(),
|
|
||||||
r.headers(),
|
|
||||||
) {
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
ConnectResponse::Tunnel(ref head, _) => {
|
|
||||||
if is_valid_response(
|
|
||||||
policies.as_ref(),
|
|
||||||
head.status,
|
|
||||||
head.headers(),
|
|
||||||
) {
|
|
||||||
return Ok(res);
|
|
||||||
} else {
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!(
|
|
||||||
"Request max retry reached"
|
|
||||||
);
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// SendRequestError
|
|
||||||
Err(e) => {
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Err(e);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Exceed max retry so just return whatever response is received
|
||||||
|
log::debug!("Request max retry reached");
|
||||||
|
connector.call(ConnectRequest::Client(
|
||||||
|
head,
|
||||||
|
body,
|
||||||
|
addr,
|
||||||
|
))
|
||||||
|
.await
|
||||||
}
|
}
|
||||||
ConnectRequest::Tunnel(head, addr) => loop {
|
ConnectRequest::Tunnel(head, addr) => {
|
||||||
let h = clone_request_head(&head);
|
for _ in 1..max_retry {
|
||||||
|
let h = clone_request_head(&head);
|
||||||
|
|
||||||
match connector.call(ConnectRequest::Tunnel(h, addr)).await {
|
let result = connector.call(ConnectRequest::Tunnel(h, addr)).await;
|
||||||
Ok(res) => match &res {
|
|
||||||
ConnectResponse::Client(r) => {
|
if let Ok(res) = result {
|
||||||
if is_valid_response(&policies, r.status(), r.headers()) {
|
match &res {
|
||||||
return Ok(res);
|
ConnectResponse::Client(r) => {
|
||||||
|
if is_valid_response(&policies, r.status(), r.headers()) {
|
||||||
|
return Ok(res);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
ConnectResponse::Tunnel(head, _) => {
|
||||||
if tries == max_retry {
|
if is_valid_response(&policies, head.status, head.headers()) {
|
||||||
log::debug!("Request max retry reached");
|
return Ok(res);
|
||||||
return Ok(res);
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
}
|
||||||
ConnectResponse::Tunnel(head, _) => {
|
|
||||||
if is_valid_response(&policies, head.status, head.headers()) {
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Ok(res);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
|
||||||
},
|
|
||||||
Err(e) => {
|
|
||||||
if tries == max_retry {
|
|
||||||
log::debug!("Request max retry reached");
|
|
||||||
return Err(e);
|
|
||||||
}
|
|
||||||
|
|
||||||
tries += 1;
|
|
||||||
}
|
}
|
||||||
}
|
};
|
||||||
},
|
|
||||||
|
// Exceed max retry so just return whatever response is received
|
||||||
|
log::debug!("Request max retry reached");
|
||||||
|
connector.call(ConnectRequest::Tunnel(head, addr)).await
|
||||||
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn body_to_retry_body(body: &Body) -> Body {
|
||||||
|
match body {
|
||||||
|
Body::Empty => Body::Empty,
|
||||||
|
Body::Bytes(b) => Body::Bytes(b.clone()),
|
||||||
|
_ => Body::None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[doc(hidden)]
|
#[doc(hidden)]
|
||||||
/// Clones [RequestHeadType] except for the extensions (not required for this middleware)
|
/// Clones [RequestHeadType] except for the extensions (not required for this middleware)
|
||||||
fn clone_request_head_type(head_type: &RequestHeadType) -> RequestHeadType {
|
fn clone_request_head_type(head_type: &RequestHeadType) -> RequestHeadType {
|
||||||
@ -430,6 +280,9 @@ mod tests {
|
|||||||
|
|
||||||
#[actix_rt::test]
|
#[actix_rt::test]
|
||||||
async fn test_basic_policy() {
|
async fn test_basic_policy() {
|
||||||
|
std::env::set_var("RUST_LOG", "RUST_LOG=debug,a=debug");
|
||||||
|
env_logger::init();
|
||||||
|
|
||||||
let client = ClientBuilder::new()
|
let client = ClientBuilder::new()
|
||||||
.disable_redirects()
|
.disable_redirects()
|
||||||
.wrap(Retry::new(3).policy(vec![StatusCode::INTERNAL_SERVER_ERROR]))
|
.wrap(Retry::new(3).policy(vec![StatusCode::INTERNAL_SERVER_ERROR]))
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user