ntex_server/net/
socket.rs

1use std::{fmt, io, net};
2
3use ntex_io::Io;
4use ntex_service::cfg::SharedCfg;
5
6use super::Token;
7
8/// Accepted socket stream.
9#[derive(Debug)]
10pub enum Stream {
11    /// TCP stream.
12    Tcp(net::TcpStream),
13    /// Unix domain socket stream.
14    #[cfg(unix)]
15    Uds(std::os::unix::net::UnixStream),
16}
17
18impl Stream {
19    pub(crate) fn convert(self, cfg: SharedCfg) -> Result<Io, io::Error> {
20        match self {
21            Stream::Tcp(stream) => ntex_net::from_tcp_stream(stream, cfg),
22            #[cfg(unix)]
23            Stream::Uds(stream) => ntex_net::from_unix_stream(stream, cfg),
24        }
25    }
26}
27
28/// Accepted connection dispatched to a worker.
29#[derive(Debug)]
30pub struct Connection {
31    pub(crate) io: Stream,
32    pub(crate) token: Token,
33}
34
35impl Connection {
36    /// Returns the CPU that handled the connection's incoming packets.
37    pub(crate) fn incoming_cpu(&self) -> Option<usize> {
38        match self.io {
39            Stream::Tcp(ref stream) => crate::affinity::incoming_cpu(stream),
40            #[cfg(unix)]
41            Stream::Uds(_) => None,
42        }
43    }
44}
45
46pub enum Listener {
47    Tcp(net::TcpListener),
48    #[cfg(unix)]
49    Uds(std::os::unix::net::UnixListener),
50}
51
52impl fmt::Debug for Listener {
53    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
54        match *self {
55            Listener::Tcp(ref lst) => write!(f, "{lst:?}"),
56            #[cfg(unix)]
57            Listener::Uds(ref lst) => write!(f, "{lst:?}"),
58        }
59    }
60}
61
62impl fmt::Display for Listener {
63    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
64        match *self {
65            Listener::Tcp(ref lst) => write!(f, "{}", lst.local_addr().ok().unwrap()),
66            #[cfg(unix)]
67            Listener::Uds(ref lst) => {
68                write!(f, "{:?}", lst.local_addr().ok().unwrap())
69            }
70        }
71    }
72}
73
74pub(crate) enum SocketAddr {
75    Tcp(net::SocketAddr),
76    #[cfg(unix)]
77    Uds(std::os::unix::net::SocketAddr),
78}
79
80impl fmt::Display for SocketAddr {
81    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
82        match *self {
83            SocketAddr::Tcp(ref addr) => write!(f, "{addr}"),
84            #[cfg(unix)]
85            SocketAddr::Uds(ref addr) => write!(f, "{addr:?}"),
86        }
87    }
88}
89
90impl fmt::Debug for SocketAddr {
91    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
92        match *self {
93            SocketAddr::Tcp(ref addr) => write!(f, "{addr:?}"),
94            #[cfg(unix)]
95            SocketAddr::Uds(ref addr) => write!(f, "{addr:?}"),
96        }
97    }
98}
99
100impl Listener {
101    pub(super) fn from_tcp(lst: net::TcpListener) -> Self {
102        let _ = lst.set_nonblocking(true);
103        Listener::Tcp(lst)
104    }
105
106    #[cfg(unix)]
107    pub(super) fn from_uds(lst: std::os::unix::net::UnixListener) -> Self {
108        let _ = lst.set_nonblocking(true);
109        Listener::Uds(lst)
110    }
111
112    pub(crate) fn local_addr(&self) -> SocketAddr {
113        match self {
114            Listener::Tcp(lst) => SocketAddr::Tcp(lst.local_addr().unwrap()),
115            #[cfg(unix)]
116            Listener::Uds(lst) => SocketAddr::Uds(lst.local_addr().unwrap()),
117        }
118    }
119
120    pub(crate) fn accept(&self) -> io::Result<Option<Stream>> {
121        match *self {
122            Listener::Tcp(ref lst) => lst.accept().map(|(stream, _)| Some(Stream::Tcp(stream))),
123            #[cfg(unix)]
124            Listener::Uds(ref lst) => lst.accept().map(|(stream, _)| Some(Stream::Uds(stream))),
125        }
126    }
127
128    pub(crate) fn remove_source(&self) {
129        match *self {
130            Listener::Tcp(_) => (),
131            #[cfg(unix)]
132            Listener::Uds(ref lst) => {
133                // cleanup file path
134                if let Ok(addr) = lst.local_addr()
135                    && let Some(path) = addr.as_pathname()
136                {
137                    let _ = std::fs::remove_file(path);
138                }
139            }
140        }
141    }
142}
143
144#[cfg(unix)]
145mod listener_impl {
146    use std::os::fd::{AsFd, BorrowedFd};
147    use std::os::unix::io::{AsRawFd, RawFd};
148
149    impl AsFd for super::Listener {
150        fn as_fd(&self) -> BorrowedFd<'_> {
151            match *self {
152                super::Listener::Tcp(ref lst) => lst.as_fd(),
153                super::Listener::Uds(ref lst) => lst.as_fd(),
154            }
155        }
156    }
157
158    impl AsRawFd for super::Listener {
159        fn as_raw_fd(&self) -> RawFd {
160            match *self {
161                super::Listener::Tcp(ref lst) => lst.as_raw_fd(),
162                super::Listener::Uds(ref lst) => lst.as_raw_fd(),
163            }
164        }
165    }
166}
167
168#[cfg(windows)]
169mod listener_impl {
170    use std::os::windows::io::{AsRawSocket, AsSocket, BorrowedSocket, RawSocket};
171
172    impl AsSocket for super::Listener {
173        fn as_socket(&self) -> BorrowedSocket<'_> {
174            match *self {
175                super::Listener::Tcp(ref lst) => lst.as_socket(),
176            }
177        }
178    }
179
180    impl AsRawSocket for super::Listener {
181        fn as_raw_socket(&self) -> RawSocket {
182            match *self {
183                super::Listener::Tcp(ref lst) => lst.as_raw_socket(),
184            }
185        }
186    }
187}
188
189#[cfg(test)]
190mod tests {
191    use super::*;
192
193    #[test]
194    fn socket_addr() {
195        use socket2::{Domain, SockAddr, Socket, Type};
196
197        let addr = SocketAddr::Tcp("127.0.0.1:8080".parse().unwrap());
198        assert!(format!("{addr:?}").contains("127.0.0.1:8080"));
199        assert_eq!(format!("{addr}"), "127.0.0.1:8080");
200
201        let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
202        let socket = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap();
203        socket.set_reuse_address(true).unwrap();
204        socket.bind(&SockAddr::from(addr)).unwrap();
205        let lst = Listener::Tcp(net::TcpListener::from(socket));
206        assert!(format!("{lst:?}").contains("TcpListener"));
207        assert!(format!("{lst}").contains("127.0.0.1"));
208    }
209
210    #[test]
211    #[cfg(unix)]
212    fn uds() {
213        use std::os::unix::net::UnixListener;
214
215        let _ = std::fs::remove_file("/tmp/sock.xxxxx");
216        if let Ok(lst) = UnixListener::bind("/tmp/sock.xxxxx") {
217            let addr = lst.local_addr().expect("Couldn't get local address");
218            let a = SocketAddr::Uds(addr);
219            assert!(format!("{a:?}").contains("/tmp/sock.xxxxx"));
220            assert!(format!("{a}").contains("/tmp/sock.xxxxx"));
221
222            let lst = Listener::Uds(lst);
223            assert!(format!("{lst:?}").contains("/tmp/sock.xxxxx"));
224            assert!(format!("{lst}").contains("/tmp/sock.xxxxx"));
225        }
226    }
227}
ntex_server/net/socket.rs

ntex_server/net/
socket.rs
