ntex 3.7.2

Framework for composable network services
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

use crate::{Service, ServiceCtx};

#[cfg(feature = "compress")]
use crate::http::Payload;
#[cfg(feature = "compress")]
use crate::http::encoding::Decoder;

use crate::http::body::MessageBody;

use super::connector::ConnectorService;
use super::error::ClientError;
use super::{ClientConfig, ClientRawRequest, Connect, ServiceRequest, ServiceResponse};

#[derive(Debug)]
pub struct Sender {
    config: ClientConfig,
    connector: ConnectorService,
}

impl Sender {
    pub(crate) fn new(connector: ConnectorService, config: ClientConfig) -> Self {
        Self { config, connector }
    }
}

#[allow(unused_variables)]
impl Service<ServiceRequest> for Sender {
    type Response = ServiceResponse;
    type Error = ClientError;

    crate::forward_ready!(connector);
    crate::forward_poll!(connector);
    crate::forward_shutdown!(connector);

    async fn call(
        &self,
        req: ServiceRequest,
        ctx: ServiceCtx<'_, Self>,
    ) -> Result<Self::Response, Self::Error> {
        let ServiceRequest {
            head,
            addr,
            body,
            headers,
            mut timeout,
            response_decompress,
        } = req;

        let con = ctx
            .call(
                &self.connector,
                Connect {
                    addr,
                    uri: head.uri.clone(),
                },
            )
            .await?;

        if timeout.is_zero() {
            timeout = self.config.timeout();
        }

        let req = ClientRawRequest {
            head,
            headers,
            size: body.size(),
        };

        let (head, payload) = con.send_request(req, body, timeout).await?;

        #[cfg(feature = "compress")]
        if response_decompress {
            let payload =
                Payload::from_stream(Decoder::from_headers(payload, &head.headers));
            return Ok(ServiceResponse {
                head,
                payload,
                config: self.config.clone(),
            });
        }

        Ok(ServiceResponse {
            head,
            payload,
            config: self.config.clone(),
        })
    }
}