volo-http 0.5.5

HTTP framework implementation of volo.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

use motore::{layer::Layer, service::Service};
use volo::context::Context;

use crate::{
    context::client::Config,
    error::ClientError,
    request::{Request, RequestPartsExt},
};

/// [`Layer`] for applying timeout from [`Config`].
///
/// This layer will be applied by default when using [`ClientBuilder::build`], without this layer,
/// timeout from [`Client`] or [`CallOpt`] will not work.
///
/// [`Client`]: crate::client::Client
/// [`ClientBuilder::build`]: crate::client::ClientBuilder::build
/// [`CallOpt`]: crate::client::CallOpt
#[derive(Clone, Debug, Default)]
pub struct Timeout;

impl<S> Layer<S> for Timeout {
    type Service = TimeoutService<S>;

    fn layer(self, inner: S) -> Self::Service {
        TimeoutService { inner }
    }
}

/// The [`Service`] generated by [`Timeout`].
///
/// See [`Timeout`] for more details.
pub struct TimeoutService<S> {
    inner: S,
}

impl<Cx, B, S> Service<Cx, Request<B>> for TimeoutService<S>
where
    Cx: Context<Config = Config> + Send,
    B: Send,
    S: Service<Cx, Request<B>, Error = ClientError> + Send + Sync,
{
    type Response = S::Response;
    type Error = S::Error;

    async fn call(&self, cx: &mut Cx, req: Request<B>) -> Result<Self::Response, Self::Error> {
        let timeout = cx.rpc_info().config().timeout().cloned();

        if let Some(duration) = timeout {
            let url = req.url();
            let fut = self.inner.call(cx, req);
            let sleep = tokio::time::sleep(duration);

            tokio::select! {
                res = fut => res,
                _ = sleep => {
                    if let Some(url) = url {
                        tracing::warn!("[Volo-HTTP] request timeout on `{url}`");
                    }
                    Err(crate::error::client::timeout().with_endpoint(cx.rpc_info().callee()))
                }
            }
        } else {
            self.inner.call(cx, req).await
        }
    }
}