karyon_net/
endpoint.rs

1use std::{
2    net::{IpAddr, SocketAddr, ToSocketAddrs},
3    path::PathBuf,
4    str::FromStr,
5    time::Duration,
6};
7
8#[cfg(all(target_family = "unix", feature = "unix"))]
9use std::os::unix::net::SocketAddr as UnixSocketAddr;
10
11use bincode::{Decode, Encode};
12use karyon_core::{async_runtime::net::lookup_host, async_util::timeout};
13use url::Url;
14
15#[cfg(feature = "serde")]
16use serde::{Deserialize, Serialize};
17
18use crate::{Error, Result};
19
20/// Port defined as a u16.
21pub type Port = u16;
22
23/// Max time to wait for a DNS lookup.
24const DNS_TIMEOUT: Duration = Duration::from_secs(10);
25
26/// Endpoint defines generic network endpoints for karyon.
27///
28/// Transport schemes (TCP, UDP, TLS, QUIC, Unix) carry just address + port.
29/// URL schemes (HTTP, HTTPS, WS, WSS) carry a full `Url` so path, query,
30/// and other URL semantics are preserved.
31///
32/// # Example
33///
34/// ```
35/// use std::net::SocketAddr;
36///
37/// use karyon_net::Endpoint;
38///
39/// let endpoint: Endpoint = "tcp://127.0.0.1:3000".parse().unwrap();
40///
41/// let socketaddr: SocketAddr = "127.0.0.1:3000".parse().unwrap();
42/// let endpoint = Endpoint::new_udp_addr(socketaddr);
43///
44/// let endpoint: Endpoint = "http://example.com:8080/rpc".parse().unwrap();
45/// ```
46#[derive(Debug, Clone, PartialEq, Eq, Hash)]
47#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
48#[cfg_attr(feature = "serde", serde(into = "String"))]
49pub enum Endpoint {
50    Udp(Addr, Port),
51    Tcp(Addr, Port),
52    Tls(Addr, Port),
53    Quic(Addr, Port),
54    Http(Url),
55    Https(Url),
56    Ws(Url),
57    Wss(Url),
58    Unix(PathBuf),
59}
60
61impl Endpoint {
62    /// Creates a new TCP endpoint from a `SocketAddr`.
63    pub fn new_tcp_addr(addr: SocketAddr) -> Endpoint {
64        Endpoint::Tcp(Addr::Ip(addr.ip()), addr.port())
65    }
66
67    /// Creates a new UDP endpoint from a `SocketAddr`.
68    pub fn new_udp_addr(addr: SocketAddr) -> Endpoint {
69        Endpoint::Udp(Addr::Ip(addr.ip()), addr.port())
70    }
71
72    /// Creates a new TLS endpoint from a `SocketAddr`.
73    pub fn new_tls_addr(addr: SocketAddr) -> Endpoint {
74        Endpoint::Tls(Addr::Ip(addr.ip()), addr.port())
75    }
76
77    /// Creates a new QUIC endpoint from a `SocketAddr`.
78    pub fn new_quic_addr(addr: SocketAddr) -> Endpoint {
79        Endpoint::Quic(Addr::Ip(addr.ip()), addr.port())
80    }
81
82    /// Creates a new Unix endpoint from a `UnixSocketAddr`.
83    pub fn new_unix_addr(addr: &std::path::Path) -> Endpoint {
84        Endpoint::Unix(addr.to_path_buf())
85    }
86
87    #[inline]
88    /// Checks if the `Endpoint` is of type `Tcp`.
89    pub fn is_tcp(&self) -> bool {
90        matches!(self, Endpoint::Tcp(..))
91    }
92
93    #[inline]
94    /// Checks if the `Endpoint` is of type `Tls`.
95    pub fn is_tls(&self) -> bool {
96        matches!(self, Endpoint::Tls(..))
97    }
98
99    #[inline]
100    /// Checks if the `Endpoint` is of type `Ws`.
101    pub fn is_ws(&self) -> bool {
102        matches!(self, Endpoint::Ws(..))
103    }
104
105    #[inline]
106    /// Checks if the `Endpoint` is of type `Wss`.
107    pub fn is_wss(&self) -> bool {
108        matches!(self, Endpoint::Wss(..))
109    }
110
111    #[inline]
112    /// Checks if the `Endpoint` is of type `Quic`.
113    pub fn is_quic(&self) -> bool {
114        matches!(self, Endpoint::Quic(..))
115    }
116
117    #[inline]
118    /// Checks if the `Endpoint` is of type `Udp`.
119    pub fn is_udp(&self) -> bool {
120        matches!(self, Endpoint::Udp(..))
121    }
122
123    #[inline]
124    /// Checks if the `Endpoint` is of type `Http`.
125    pub fn is_http(&self) -> bool {
126        matches!(self, Endpoint::Http(..))
127    }
128
129    #[inline]
130    /// Checks if the `Endpoint` is of type `Https`.
131    pub fn is_https(&self) -> bool {
132        matches!(self, Endpoint::Https(..))
133    }
134
135    #[inline]
136    /// Checks if the `Endpoint` is of type `Unix`.
137    pub fn is_unix(&self) -> bool {
138        matches!(self, Endpoint::Unix(..))
139    }
140
141    /// Returns the port of the endpoint. For URL-family endpoints the
142    /// scheme's default port is returned if the URL omits a port.
143    pub fn port(&self) -> Result<Port> {
144        match self {
145            Endpoint::Tcp(_, port)
146            | Endpoint::Udp(_, port)
147            | Endpoint::Tls(_, port)
148            | Endpoint::Quic(_, port) => Ok(*port),
149            Endpoint::Http(url) | Endpoint::Https(url) | Endpoint::Ws(url) | Endpoint::Wss(url) => {
150                url.port_or_known_default()
151                    .ok_or_else(|| Error::ParseEndpoint(format!("port missing: {url}")))
152            }
153            Endpoint::Unix(_) => Err(Error::TryFromEndpoint),
154        }
155    }
156
157    /// Returns the address of the endpoint.
158    pub fn addr(&self) -> Result<Addr> {
159        match self {
160            Endpoint::Tcp(addr, _)
161            | Endpoint::Udp(addr, _)
162            | Endpoint::Tls(addr, _)
163            | Endpoint::Quic(addr, _) => Ok(addr.clone()),
164            Endpoint::Http(url) | Endpoint::Https(url) | Endpoint::Ws(url) | Endpoint::Wss(url) => {
165                url_to_addr(url)
166            }
167            Endpoint::Unix(_) => Err(Error::TryFromEndpoint),
168        }
169    }
170}
171
172fn url_to_addr(url: &Url) -> Result<Addr> {
173    let host = url
174        .host_str()
175        .ok_or_else(|| Error::ParseEndpoint(format!("host missing: {url}")))?;
176    match host.parse::<IpAddr>() {
177        Ok(ip) => Ok(Addr::Ip(ip)),
178        Err(_) => Ok(Addr::Domain(host.to_string())),
179    }
180}
181
182impl std::fmt::Display for Endpoint {
183    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
184        match self {
185            Endpoint::Udp(ip, port) => write!(f, "udp://{ip}:{port}"),
186            Endpoint::Tcp(ip, port) => write!(f, "tcp://{ip}:{port}"),
187            Endpoint::Tls(ip, port) => write!(f, "tls://{ip}:{port}"),
188            Endpoint::Quic(ip, port) => write!(f, "quic://{ip}:{port}"),
189            Endpoint::Http(url) | Endpoint::Https(url) | Endpoint::Ws(url) | Endpoint::Wss(url) => {
190                write!(f, "{url}")
191            }
192            Endpoint::Unix(path) => write!(f, "unix:/{}", path.to_string_lossy()),
193        }
194    }
195}
196
197impl From<Endpoint> for String {
198    fn from(endpoint: Endpoint) -> String {
199        endpoint.to_string()
200    }
201}
202
203/// Converts an endpoint to a socket address.
204///
205/// For a domain endpoint this runs a blocking DNS lookup on the current
206/// thread. In async code, avoid it for untrusted or remote endpoints; the
207/// dial functions resolve domains off the executor instead.
208impl TryFrom<Endpoint> for SocketAddr {
209    type Error = Error;
210    fn try_from(endpoint: Endpoint) -> std::result::Result<SocketAddr, Self::Error> {
211        match endpoint {
212            Endpoint::Udp(addr, port)
213            | Endpoint::Tcp(addr, port)
214            | Endpoint::Tls(addr, port)
215            | Endpoint::Quic(addr, port) => resolve(addr, port),
216            Endpoint::Http(ref url)
217            | Endpoint::Https(ref url)
218            | Endpoint::Ws(ref url)
219            | Endpoint::Wss(ref url) => {
220                let addr = url_to_addr(url)?;
221                let port = endpoint.port()?;
222                resolve(addr, port)
223            }
224            Endpoint::Unix(_) => Err(Error::TryFromEndpoint),
225        }
226    }
227}
228
229/// Resolve an `Addr` + port to a `SocketAddr`. Domains are resolved
230/// through the system DNS with the given port.
231fn resolve(addr: Addr, port: Port) -> Result<SocketAddr> {
232    match addr {
233        Addr::Ip(ip) => Ok(SocketAddr::new(ip, port)),
234        Addr::Domain(d) => (d.as_str(), port)
235            .to_socket_addrs()?
236            .next()
237            .ok_or_else(|| Error::ParseEndpoint(format!("could not resolve {d}:{port}"))),
238    }
239}
240
241/// Async version of `SocketAddr::try_from(endpoint)`. Domains are resolved
242/// on a blocking thread pool under a timeout, so DNS never blocks the
243/// executor.
244pub(crate) async fn resolve_endpoint(endpoint: &Endpoint) -> Result<SocketAddr> {
245    let port = endpoint.port()?;
246    match endpoint.addr()? {
247        Addr::Ip(ip) => Ok(SocketAddr::new(ip, port)),
248        Addr::Domain(d) => timeout(DNS_TIMEOUT, lookup_host(&d, port))
249            .await??
250            .into_iter()
251            .next()
252            .ok_or_else(|| Error::ParseEndpoint(format!("could not resolve {d}:{port}"))),
253    }
254}
255
256impl TryFrom<Endpoint> for PathBuf {
257    type Error = Error;
258    fn try_from(endpoint: Endpoint) -> std::result::Result<PathBuf, Self::Error> {
259        match endpoint {
260            Endpoint::Unix(path) => Ok(PathBuf::from(&path)),
261            _ => Err(Error::TryFromEndpoint),
262        }
263    }
264}
265
266#[cfg(all(feature = "unix", target_family = "unix"))]
267impl TryFrom<Endpoint> for UnixSocketAddr {
268    type Error = Error;
269    fn try_from(endpoint: Endpoint) -> std::result::Result<UnixSocketAddr, Self::Error> {
270        match endpoint {
271            Endpoint::Unix(a) => Ok(UnixSocketAddr::from_pathname(a)?),
272            _ => Err(Error::TryFromEndpoint),
273        }
274    }
275}
276
277impl TryFrom<Endpoint> for Url {
278    type Error = Error;
279
280    fn try_from(ep: Endpoint) -> Result<Self> {
281        match ep {
282            Endpoint::Http(u) | Endpoint::Https(u) | Endpoint::Ws(u) | Endpoint::Wss(u) => Ok(u),
283            other => other
284                .to_string()
285                .parse::<Url>()
286                .map_err(|e| Error::ParseEndpoint(e.to_string())),
287        }
288    }
289}
290
291impl TryFrom<Url> for Endpoint {
292    type Error = Error;
293
294    fn try_from(url: Url) -> Result<Self> {
295        match url.scheme() {
296            "http" => Ok(Endpoint::Http(url)),
297            "https" => Ok(Endpoint::Https(url)),
298            "ws" => Ok(Endpoint::Ws(url)),
299            "wss" => Ok(Endpoint::Wss(url)),
300            "tcp" | "udp" | "tls" | "quic" => {
301                let addr = url_to_addr(&url)?;
302                let port = url
303                    .port()
304                    .ok_or_else(|| Error::ParseEndpoint(format!("port missing: {url}")))?;
305                Ok(match url.scheme() {
306                    "tcp" => Endpoint::Tcp(addr, port),
307                    "udp" => Endpoint::Udp(addr, port),
308                    "tls" => Endpoint::Tls(addr, port),
309                    "quic" => Endpoint::Quic(addr, port),
310                    _ => unreachable!(),
311                })
312            }
313            "unix" => {
314                if url.path().is_empty() {
315                    return Err(Error::UnsupportedEndpoint(url.to_string()));
316                }
317                Ok(Endpoint::Unix(url.path().into()))
318            }
319            _ => Err(Error::UnsupportedEndpoint(url.to_string())),
320        }
321    }
322}
323
324impl FromStr for Endpoint {
325    type Err = Error;
326
327    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
328        let url: Url = s
329            .parse()
330            .map_err(|err: url::ParseError| Error::ParseEndpoint(err.to_string()))?;
331        Endpoint::try_from(url)
332    }
333}
334
335/// Addr defines a type for an address, either IP or domain.
336#[derive(Debug, Clone, PartialEq, Eq, Hash, Encode, Decode)]
337#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
338pub enum Addr {
339    Ip(IpAddr),
340    Domain(String),
341}
342
343impl TryFrom<Addr> for IpAddr {
344    type Error = std::io::Error;
345    fn try_from(addr: Addr) -> std::result::Result<IpAddr, Self::Error> {
346        match addr {
347            Addr::Ip(ip) => Ok(ip),
348            Addr::Domain(_) => Err(std::io::Error::new(
349                std::io::ErrorKind::InvalidInput,
350                "Addr::Domain cannot be converted to IpAddr without a port; \
351                 use SocketAddr::try_from(Endpoint) instead",
352            )),
353        }
354    }
355}
356
357impl std::fmt::Display for Addr {
358    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
359        match self {
360            Addr::Ip(ip) => write!(f, "{ip}"),
361            Addr::Domain(d) => write!(f, "{d}"),
362        }
363    }
364}
365
366pub trait ToEndpoint {
367    fn to_endpoint(&self) -> Result<Endpoint>;
368}
369
370impl ToEndpoint for String {
371    fn to_endpoint(&self) -> Result<Endpoint> {
372        Endpoint::from_str(self)
373    }
374}
375
376impl ToEndpoint for Endpoint {
377    fn to_endpoint(&self) -> Result<Endpoint> {
378        Ok(self.clone())
379    }
380}
381
382impl ToEndpoint for &str {
383    fn to_endpoint(&self) -> Result<Endpoint> {
384        Endpoint::from_str(self)
385    }
386}
387
388#[cfg(test)]
389mod tests {
390    use super::*;
391    use std::net::Ipv4Addr;
392    use std::path::PathBuf;
393
394    use karyon_core::testing::run_test;
395
396    #[test]
397    fn test_resolve_endpoint() {
398        run_test(10, async {
399            let ep: Endpoint = "tcp://127.0.0.1:3000".parse().unwrap();
400            let addr = resolve_endpoint(&ep).await.unwrap();
401            assert_eq!(addr, "127.0.0.1:3000".parse().unwrap());
402
403            // A domain goes through the async lookup.
404            let ep: Endpoint = "tcp://localhost:3000".parse().unwrap();
405            let addr = resolve_endpoint(&ep).await.unwrap();
406            assert!(addr.ip().is_loopback());
407            assert_eq!(addr.port(), 3000);
408
409            let ep: Endpoint = "unix:///tmp/karyon.sock".parse().unwrap();
410            assert!(resolve_endpoint(&ep).await.is_err());
411        });
412    }
413
414    #[test]
415    fn test_endpoint_from_str() {
416        let endpoint_str: Endpoint = "tcp://127.0.0.1:3000".parse().unwrap();
417        let endpoint = Endpoint::Tcp(Addr::Ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))), 3000);
418        assert_eq!(endpoint_str, endpoint);
419
420        let endpoint_str: Endpoint = "udp://127.0.0.1:4000".parse().unwrap();
421        let endpoint = Endpoint::Udp(Addr::Ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))), 4000);
422        assert_eq!(endpoint_str, endpoint);
423
424        let endpoint_str: Endpoint = "tcp://example.com:3000".parse().unwrap();
425        let endpoint = Endpoint::Tcp(Addr::Domain("example.com".to_string()), 3000);
426        assert_eq!(endpoint_str, endpoint);
427
428        let endpoint_str = "unix:/home/x/s.socket".parse::<Endpoint>().unwrap();
429        let endpoint = Endpoint::Unix(PathBuf::from_str("/home/x/s.socket").unwrap());
430        assert_eq!(endpoint_str, endpoint);
431    }
432
433    #[test]
434    fn test_endpoint_url_preserves_path() {
435        let endpoint: Endpoint = "http://example.com:8080/rpc?x=1".parse().unwrap();
436        match &endpoint {
437            Endpoint::Http(url) => {
438                assert_eq!(url.path(), "/rpc");
439                assert_eq!(url.query(), Some("x=1"));
440                assert_eq!(url.port(), Some(8080));
441            }
442            _ => panic!("expected Http"),
443        }
444        assert_eq!(endpoint.port().unwrap(), 8080);
445    }
446
447    #[test]
448    fn test_endpoint_url_default_port() {
449        let endpoint: Endpoint = "https://example.com/".parse().unwrap();
450        assert_eq!(endpoint.port().unwrap(), 443);
451
452        let endpoint: Endpoint = "ws://example.com/".parse().unwrap();
453        assert_eq!(endpoint.port().unwrap(), 80);
454    }
455}
karyon_net/endpoint.rs

karyon_net/
endpoint.rs
