ntex_net/connect/
service.rs

1use std::{collections::VecDeque, io, marker, net::SocketAddr};
2
3use ntex_error::Error;
4use ntex_io::{Io, IoConfig, types};
5use ntex_service::{Ctx, Service, cfg::SharedCfg};
6use ntex_util::{future::Either, time::timeout_checked};
7
8use super::{Address, Connect, ConnectError, resolve};
9
10#[derive(Debug)]
11/// DNS-aware TCP stream connector.
12pub struct Connector<A> {
13    _t: marker::PhantomData<A>,
14}
15
16impl<A> Connector<A> {
17    #[inline]
18    /// Creates a connector.
19    pub fn new() -> Self {
20        Connector {
21            _t: marker::PhantomData,
22        }
23    }
24}
25
26impl<A> Default for Connector<A> {
27    fn default() -> Self {
28        Connector::new()
29    }
30}
31
32impl<A> Clone for Connector<A> {
33    fn clone(&self) -> Self {
34        Connector {
35            _t: marker::PhantomData,
36        }
37    }
38}
39
40impl<A: Address> Connector<A> {
41    /// Resolves and connects to a remote host.
42    pub async fn connect<U>(&self, message: U, cfg: &SharedCfg) -> Result<Io, Error<ConnectError>>
43    where
44        Connect<A>: From<U>,
45    {
46        let timeout = cfg.get::<IoConfig>().connect_timeout();
47        timeout_checked(timeout, async {
48            // resolve first
49            let msg = resolve::lookup(message.into(), cfg.tag()).await?;
50
51            let port = msg.port();
52            let Connect { req, addr, .. } = msg;
53
54            if let Some(addr) = addr {
55                connect(req, port, addr, cfg).await
56            } else if let Some(addr) = req.addr() {
57                connect(req, addr.port(), Either::Left(addr), cfg).await
58            } else {
59                Err(Error::from(ConnectError::Unresolved))
60            }
61        })
62        .await
63        .map_err(|()| {
64            Error::from(ConnectError::Io(io::Error::new(
65                io::ErrorKind::TimedOut,
66                "Connect timeout",
67            )))
68        })
69        .and_then(|item| item)
70        .map_err(|e| e.with_service(cfg.service()))
71    }
72}
73
74impl<A: Address> Service<SharedCfg, Connect<A>> for Connector<A> {
75    type Res = Io;
76    type Error = Error<ConnectError>;
77
78    async fn call(
79        &self,
80        req: Connect<A>,
81        ctx: Ctx<'_, Self, SharedCfg>,
82    ) -> Result<Io, Self::Error> {
83        self.connect(req, ctx.st()).await
84    }
85}
86
87/// Tcp stream connector
88async fn connect<A: Address>(
89    req: A,
90    port: u16,
91    addr: Either<SocketAddr, VecDeque<SocketAddr>>,
92    cfg: &SharedCfg,
93) -> Result<Io, Error<ConnectError>> {
94    log::trace!(
95        "{}: TCP connector - connecting to {:?} addr:{addr:?} port:{port}",
96        cfg.tag(),
97        req.host(),
98    );
99
100    let io = match addr {
101        Either::Left(addr) => crate::tcp_connect(addr, cfg.clone())
102            .await
103            .map_err(ConnectError::from)?,
104        Either::Right(mut addrs) => loop {
105            let addr = addrs.pop_front().unwrap();
106
107            match crate::tcp_connect(addr, cfg.clone()).await {
108                Ok(io) => break io,
109                Err(err) => {
110                    log::trace!(
111                        "{}: TCP connector - failed to connect to {:?} port: {port} err: {err:?}",
112                        cfg.tag(),
113                        req.host(),
114                    );
115                    if addrs.is_empty() {
116                        return Err(ConnectError::from(err).into());
117                    }
118                }
119            }
120        },
121    };
122
123    log::trace!(
124        "{}: TCP connector - successfully connected to {:?} - {:?}",
125        cfg.tag(),
126        req.host(),
127        io.query::<types::PeerAddr>().get()
128    );
129    Ok(io)
130}
ntex_net/connect/service.rs

ntex_net/connect/
service.rs
