fn req_with_retries(ctx: &Context, path: String, authorization: String, body: Option<String>, try_no: u32) -> Box<Future<Item=String, Error=io::Error> + Send> {

    let ctx_clone = ctx.clone();

    let (client, seqno) = {
        let mut guard = ctx.http_client.lock().unwrap();
        let http_client = &mut *guard;
        if http_client.client.is_none() {
            // create a new http client.
            http_client.client.get_or_insert_with(|| new_client(&ctx.config));
            http_client.seqno += 1;
        }
        let cc = http_client.client.as_ref().unwrap().clone();
        (cc, http_client.seqno)
    };

    // build the uri based on the currently active webnis server.
    let server = &ctx.config.servers[seqno % ctx.config.servers.len()];
    let uri = build_uri(server, &path);
    let method = if body.is_some() { Method::POST } else { Method::GET };

    let mut builder = hyper::Request::builder();
    builder
        .uri(uri)
        .method(method)
        .header(header::AUTHORIZATION, authorization.as_str());
    if body.is_some() {
        builder.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
    }
    let request = builder
        .body(body.clone().map(|x| x.into()).unwrap_or(hyper::Body::empty()))
        .unwrap();

    let resp_body = client.request(request)
    .map_err(|e| {
        // something went very wrong. mark it with code 550 so that at the
        // end of the future chain we can detect it and retry.
        //
        // FIXME differ between real problems where we need to throw away the
        // hyper::Client and problems where we just need to switch to the next server.
        debug!("client: got error, need retry: {}", e);
        Response::error(550, &format!("GET error: {}", e))
    })
    .and_then(|res| {
        // see if response is what we expected
        let is_json = res.headers().get(header::CONTENT_TYPE).map(|h| h == "application/json").unwrap_or(false);
        if !is_json {
            if res.status().is_success() {
                future::err(Response::error(416, "expected application/json"))
            } else {
                let code = res.status().as_u16() as i64;
                future::err(Response::error(code, "HTTP error"))
            }
        } else {
            future::ok(res)
        }
    })
    .and_then(|res| {
        res
        .into_body()
        .concat2()
        .map_err(|_| Response::error(400, "GET body error"))
    });

    // add a timeout. need to have an answer in 1 second.
    let timeout = Duration::from_millis(REQUEST_TIMEOUT_MS);
    let body_tmout_wrapper = resp_body.timeout(timeout).map_err(|e| {
        debug!("got error {}", e);
        match e.into_inner() {
            Some(e) => e,
            None => Response::error(408, "request timeout"),
        }
    });

    let resp =
    body_tmout_wrapper.then(move |res| {
        let resp_body = match res {
            Ok(body) => body,
            Err(e) => {
                if !e.starts_with("401 ") &&
                   !e.starts_with("403 ") &&
                   !e.starts_with("404 ") &&
                   !ctx_clone.eof.load(Ordering::SeqCst) &&
                   try_no < MAX_TRIES {
                    {
    				    let mut guard = ctx_clone.http_client.lock().unwrap();
                        if (*guard).seqno == seqno {
                            // only do something if noone else took action.
                            debug!("invalidating server {} and scheduling retry {} because of {}",
                                   ctx_clone.config.servers[seqno % ctx_clone.config.servers.len()], try_no + 1, e);
                            if e.starts_with("550 ") {
                                // throw away hyper::Client
    				            (*guard).client.take();
                                // reset DNS resolver state just in case.
                                // see https://github.com/rust-lang/rust/issues/47955
                                #[cfg(target_env = "gnu")]
                                {
                                    extern { fn __h_errno_location() -> *mut i32; }
                                    unsafe { *__h_errno_location() = 0 }
                                }
                            } else {
                                // just switch to next server.
                                (*guard).seqno += 1;
                            }
                        } else {
                            debug!("scheduling try {} because of {}", try_no + 1, e);
                        }
                    }
					// and retry.
                    return req_with_retries(&ctx_clone, path, authorization, body, try_no + 1);
                } else {
                    return Box::new(future::ok(e));
                }
            },
        };
        Box::new(future::ok(Response::transform(resp_body)))
    });

    if try_no > 1 {
        let when = Instant::now() + Duration::from_millis(RETRY_DELAY_MS);
        Box::new(Delay::new(when).then(move |_| resp))
    } else {
        Box::new(resp)
    }
}
// This function can call itself recursively to keep on
// generating futures so as to retry.
//
// On errors (except 404) we cycle to the next server.
//
// If there is a serious error from hyper::Client that we do not reckognize,
// we throw away the current hyper::Client instance and create a new one.
//
// This guards against bugs in hyper::Client or its dependencies
// that can get a hyper::Client stuck, see:
//
// https://github.com/hyperium/hyper/issues/1422
// https://github.com/rust-lang/rust/issues/47955
//
This function can call itself recursively to keep on
generating futures so as to retry.

On errors (except 404) we cycle to the next server.

If there is a serious error from hyper::Client that we do not reckognize,
we throw away the current hyper::Client instance and create a new one.

This guards against bugs in hyper::Client or its dependencies
that can get a hyper::Client stuck, see.


This function can call itself recursively to keep on generating futures so as to retry . On errors ( except 404 ) we cycle to the next server . If there is a serious error from hyper :: Client that we do not reckognize we throw away the current hyper :: Client instance and create a new one . This guards against bugs in hyper :: Client or its dependencies that can get a hyper :: Client stuck see .
fn req_with_retries(ctx: &Context, path: String, authorization: String, body: Option<String>, try_no: u32) -> Box<Future<Item=String, Error=io::Error> + Send> {
    let ctx_clone = ctx.clone();
    let (client, seqno) = {
        let mut guard = ctx.http_client.lock().unwrap();
        let http_client = &mut *guard;
        if http_client.client.is_none() {
            // create a new http client.
            http_client.client.get_or_insert_with(|| new_client(&ctx.config));
            http_client.seqno += 1;
        }
        let cc = http_client.client.as_ref().unwrap().clone();
        (cc, http_client.seqno)
    };
    // build the uri based on the currently active webnis server.
    let server = &ctx.config.servers[seqno % ctx.config.servers.len()];
    let uri = build_uri(server, &path);
    let method = if body.is_some() { Method::POST } else { Method::GET };
    let mut builder = hyper::Request::builder();
    builder
        .uri(uri)
        .method(method)
        .header(header::AUTHORIZATION, authorization.as_str());
    if body.is_some() {
        builder.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
    }
    let request = builder
        .body(body.clone().map(|x| x.into()).unwrap_or(hyper::Body::empty()))
        .unwrap();
    let resp_body = client.request(request)
    .map_err(|e| {
        // something went very wrong. mark it with code 550 so that at the
        // end of the future chain we can detect it and retry.
        //
        // FIXME differ between real problems where we need to throw away the
        // hyper::Client and problems where we just need to switch to the next server.
        debug!("client: got error, need retry: {}", e);
        Response::error(550, &format!("GET error: {}", e))
    })
    .and_then(|res| {
        // see if response is what we expected
        let is_json = res.headers().get(header::CONTENT_TYPE).map(|h| h == "application/json").unwrap_or(false);
        if !is_json {
            if res.status().is_success() {
                future::err(Response::error(416, "expected application/json"))
            } else {
                let code = res.status().as_u16() as i64;
                future::err(Response::error(code, "HTTP error"))
            }
        } else {
            future::ok(res)
        }
    })
    .and_then(|res| {
        res
        .into_body()
        .concat2()
        .map_err(|_| Response::error(400, "GET body error"))
    });
    // add a timeout. need to have an answer in 1 second.
    let timeout = Duration::from_millis(REQUEST_TIMEOUT_MS);
    let body_tmout_wrapper = resp_body.timeout(timeout).map_err(|e| {
        debug!("got error {}", e);
        match e.into_inner() {
            Some(e) => e,
            None => Response::error(408, "request timeout"),
        }
    });
    let resp =
    body_tmout_wrapper.then(move |res| {
        let resp_body = match res {
            Ok(body) => body,
            Err(e) => {
                if !e.starts_with("401 ") &&
                   !e.starts_with("403 ") &&
                   !e.starts_with("404 ") &&
                   !ctx_clone.eof.load(Ordering::SeqCst) &&
                   try_no < MAX_TRIES {
                    {
    				    let mut guard = ctx_clone.http_client.lock().unwrap();
                        if (*guard).seqno == seqno {
                            // only do something if noone else took action.
                            debug!("invalidating server {} and scheduling retry {} because of {}",
                                   ctx_clone.config.servers[seqno % ctx_clone.config.servers.len()], try_no + 1, e);
                            if e.starts_with("550 ") {
                                // throw away hyper::Client
    				            (*guard).client.take();
                                // reset DNS resolver state just in case.
                                // see https://github.com/rust-lang/rust/issues/47955
                                #[cfg(target_env = "gnu")]
                                {
                                    extern { fn __h_errno_location() -> *mut i32; }
                                    unsafe { *__h_errno_location() = 0 }
                                }
                            } else {
                                // just switch to next server.
                                (*guard).seqno += 1;
                            }
                        } else {
                            debug!("scheduling try {} because of {}", try_no + 1, e);
                        }
                    }
					// and retry.
                    return req_with_retries(&ctx_clone, path, authorization, body, try_no + 1);
                } else {
                    return Box::new(future::ok(e));
                }
            },
        };
        Box::new(future::ok(Response::transform(resp_body)))
    });
    if try_no > 1 {
        let when = Instant::now() + Duration::from_millis(RETRY_DELAY_MS);
        Box::new(Delay::new(when).then(move |_| resp))
    } else {
        Box::new(resp)
    }
}
fn req_with_retries ( ctx : & Context , path : String , authorization : String , body : Option < String > , try_no : u32 ) -> Box < Future < Item = String , Error = io :: Error > + Send > { let ctx_clone = ctx . clone ( ) ; let ( client , seqno ) = { let mut guard = ctx . http_client . lock ( ) . unwrap ( ) ; let http_client = & mut * guard ; if http_client . client . is_none ( ) { // create a new http client. http_client . client . get_or_insert_with ( | | new_client ( & ctx . config ) ) ; http_client . seqno += 1 ; } let cc = http_client . client . as_ref ( ) . unwrap ( ) . clone ( ) ; ( cc , http_client . seqno ) } ; // build the uri based on the currently active webnis server. let server = & ctx . config . servers [ seqno % ctx . config . servers . len ( ) ] ; let uri = build_uri ( server , & path ) ; let method = if body . is_some ( ) { Method :: POST } else { Method :: GET } ; let mut builder = hyper :: Request :: builder ( ) ; builder . uri ( uri ) . method ( method ) . header ( header :: AUTHORIZATION , authorization . as_str ( ) ) ; if body . is_some ( ) { builder . header ( header :: CONTENT_TYPE , " " ) ; } let request = builder . body ( body . clone ( ) . map ( | x | x . into ( ) ) . unwrap_or ( hyper :: Body :: empty ( ) ) ) . unwrap ( ) ; let resp_body = client . request ( request ) . map_err ( | e | { // something went very wrong. mark it with code 550 so that at the // end of the future chain we can detect it and retry. // // FIXME differ between real problems where we need to throw away the // hyper::Client and problems where we just need to switch to the next server. debug ! ( " " e ) ; Response :: error ( 550 , & format ! ( " " e ) ) } ) . and_then ( | res | { // see if response is what we expected let is_json = res . headers ( ) . get ( header :: CONTENT_TYPE ) . map ( | h | h == " " ) . unwrap_or ( false ) ; if ! is_json { if res . status ( ) . is_success ( ) { future :: err ( Response :: error ( 416 , " " ) ) } else { let code = res . status ( ) . as_u16 ( ) as i64 ; future :: err ( Response :: error ( code , " " ) ) } } else { future :: ok ( res ) } } ) . and_then ( | res | { res . into_body ( ) . concat2 ( ) . map_err ( | _ | Response :: error ( 400 , " " ) ) } ) ; // add a timeout. need to have an answer in 1 second. let timeout = Duration :: from_millis ( REQUEST_TIMEOUT_MS ) ; let body_tmout_wrapper = resp_body . timeout ( timeout ) . map_err ( | e | { debug ! ( " " e ) ; match e . into_inner ( ) { Some ( e ) => e , None => Response :: error ( 408 , " " ) , } } ) ; let resp = body_tmout_wrapper . then ( move | res | { let resp_body = match res { Ok ( body ) => body , Err ( e ) => { if ! e . starts_with ( " " ) && ! e . starts_with ( " " ) && ! e . starts_with ( " " ) && ! ctx_clone . eof . load ( Ordering :: SeqCst ) && try_no < MAX_TRIES { { let mut guard = ctx_clone . http_client . lock ( ) . unwrap ( ) ; if ( * guard ) . seqno == seqno { // only do something if noone else took action. debug ! ( " " ctx_clone config servers [ seqno ctx_clone config servers len ( ) ] try_no 1 e ) ; if e . starts_with ( " " ) { // throw away hyper::Client ( * guard ) . client . take ( ) ; // reset DNS resolver state just in case. // see https://github.com/rust-lang/rust/issues/47955 # [ cfg ( target_env " " ) ] { extern { fn __h_errno_location ( ) -> * mut i32 ; } unsafe { * __h_errno_location ( ) = 0 } } } else { // just switch to next server. ( * guard ) . seqno += 1 ; } } else { debug ! ( " " try_no 1 e ) ; } } // and retry. return req_with_retries ( & ctx_clone , path , authorization , body , try_no + 1 ) ; } else { return Box :: new ( future :: ok ( e ) ) ; } } , } ; Box :: new ( future :: ok ( Response :: transform ( resp_body ) ) ) } ) ; if try_no > 1 { let when = Instant :: now ( ) + Duration :: from_millis ( RETRY_DELAY_MS ) ; Box :: new ( Delay :: new ( when ) . then ( move | _ | resp ) ) } else { Box :: new ( resp ) } }
This function can call itself recursively to keep on
generating futures so as to retry.
This function can call itself recursively to keep on generating futures so as to retry .
ctx
&Context
path
String
authorization
String
body
Option<String>
try_no
u32