zbus/address/transport/
tcp.rs

1use super::encode_percents;
2use crate::{Address, Error, Result};
3#[cfg(feature = "async-io")]
4use async_io::Async;
5#[cfg(feature = "async-io")]
6use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
7use std::{
8    collections::HashMap,
9    fmt::{Display, Formatter},
10    str::FromStr,
11    sync::Arc,
12};
13#[cfg(all(feature = "tokio", not(feature = "async-io")))]
14use tokio::net::TcpStream;
15
16/// A TCP transport in a D-Bus address.
17#[derive(Clone, Debug, PartialEq, Eq)]
18pub struct Tcp {
19    pub(super) host: String,
20    pub(super) bind: Option<String>,
21    pub(super) port: u16,
22    pub(super) family: Option<TcpTransportFamily>,
23    pub(super) nonce_file: Option<Vec<u8>>,
24}
25
26impl Tcp {
27    /// Create a new TCP transport with the given host and port.
28    pub fn new(host: &str, port: u16) -> Self {
29        Self {
30            host: host.to_owned(),
31            port,
32            bind: None,
33            family: None,
34            nonce_file: None,
35        }
36    }
37
38    /// Set the `tcp:` address `bind` value.
39    pub fn set_bind(mut self, bind: Option<String>) -> Self {
40        self.bind = bind;
41
42        self
43    }
44
45    /// Set the `tcp:` address `family` value.
46    pub fn set_family(mut self, family: Option<TcpTransportFamily>) -> Self {
47        self.family = family;
48
49        self
50    }
51
52    /// Set the `tcp:` address `noncefile` value.
53    pub fn set_nonce_file(mut self, nonce_file: Option<Vec<u8>>) -> Self {
54        self.nonce_file = nonce_file;
55
56        self
57    }
58
59    /// The `tcp:` address `host` value.
60    pub fn host(&self) -> &str {
61        &self.host
62    }
63
64    /// The `tcp:` address `bind` value.
65    pub fn bind(&self) -> Option<&str> {
66        self.bind.as_deref()
67    }
68
69    /// The `tcp:` address `port` value.
70    pub fn port(&self) -> u16 {
71        self.port
72    }
73
74    /// The `tcp:` address `family` value.
75    pub fn family(&self) -> Option<TcpTransportFamily> {
76        self.family
77    }
78
79    /// The nonce file path, if any.
80    pub fn nonce_file(&self) -> Option<&[u8]> {
81        self.nonce_file.as_deref()
82    }
83
84    /// Take ownership of the nonce file path, if any.
85    pub fn take_nonce_file(&mut self) -> Option<Vec<u8>> {
86        self.nonce_file.take()
87    }
88
89    pub(super) fn from_options(
90        opts: HashMap<&str, &str>,
91        nonce_tcp_required: bool,
92    ) -> Result<Self> {
93        let bind = None;
94        if opts.contains_key("bind") {
95            return Err(Error::Address("`bind` isn't yet supported".into()));
96        }
97
98        let host = opts
99            .get("host")
100            .ok_or_else(|| Error::Address("tcp address is missing `host`".into()))?
101            .to_string();
102        let port = opts
103            .get("port")
104            .ok_or_else(|| Error::Address("tcp address is missing `port`".into()))?;
105        let port = port
106            .parse::<u16>()
107            .map_err(|_| Error::Address("invalid tcp `port`".into()))?;
108        let family = opts
109            .get("family")
110            .map(|f| TcpTransportFamily::from_str(f))
111            .transpose()?;
112        let nonce_file = opts
113            .get("noncefile")
114            .map(|f| super::decode_percents(f))
115            .transpose()?;
116        if nonce_tcp_required && nonce_file.is_none() {
117            return Err(Error::Address(
118                "nonce-tcp address is missing `noncefile`".into(),
119            ));
120        }
121
122        Ok(Self {
123            host,
124            bind,
125            port,
126            family,
127            nonce_file,
128        })
129    }
130
131    #[cfg(feature = "async-io")]
132    pub(super) async fn connect(self, address: &Address) -> Result<Async<TcpStream>> {
133        let address_clone = address.clone();
134        let family = self.family();
135        let addrs = crate::Task::spawn_blocking(
136            move || -> Result<Vec<SocketAddr>> {
137                let addrs = (self.host(), self.port())
138                    .to_socket_addrs()
139                    .map_err(|e| Error::Connection(Arc::new(e), address_clone))?
140                    .filter(|a| {
141                        if let Some(family) = self.family() {
142                            if family == TcpTransportFamily::Ipv4 {
143                                a.is_ipv4()
144                            } else {
145                                a.is_ipv6()
146                            }
147                        } else {
148                            true
149                        }
150                    });
151                Ok(addrs.collect())
152            },
153            "connect tcp",
154        )
155        .await
156        .map_err(|e| Error::Address(format!("Failed to receive TCP addresses: {e}")))??;
157
158        let mut last_err = Error::Address(match family {
159            Some(family) => format!("no `{family}` addresses found for `{address}`"),
160            None => format!("no addresses found for `{address}`"),
161        });
162
163        // we could attempt connections in parallel?
164        for addr in addrs {
165            match Async::<TcpStream>::connect(addr).await {
166                Ok(stream) => return Ok(stream),
167                Err(e) => last_err = Error::Connection(Arc::new(e), address.clone()),
168            }
169        }
170
171        Err(last_err)
172    }
173
174    #[cfg(all(feature = "tokio", not(feature = "async-io")))]
175    pub(super) async fn connect(self, address: &Address) -> Result<TcpStream> {
176        TcpStream::connect((self.host(), self.port()))
177            .await
178            .map_err(|e| Error::Connection(Arc::new(e), address.clone()))
179    }
180}
181
182impl Display for Tcp {
183    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
184        match self.nonce_file() {
185            Some(nonce_file) => {
186                f.write_str("nonce-tcp:noncefile=")?;
187                encode_percents(f, nonce_file)?;
188                f.write_str(",")?;
189            }
190            None => f.write_str("tcp:")?,
191        }
192        f.write_str("host=")?;
193
194        encode_percents(f, self.host().as_bytes())?;
195
196        write!(f, ",port={}", self.port())?;
197
198        if let Some(bind) = self.bind() {
199            f.write_str(",bind=")?;
200            encode_percents(f, bind.as_bytes())?;
201        }
202
203        if let Some(family) = self.family() {
204            write!(f, ",family={family}")?;
205        }
206
207        Ok(())
208    }
209}
210
211/// A `tcp:` address family.
212#[derive(Copy, Clone, Debug, PartialEq, Eq)]
213pub enum TcpTransportFamily {
214    Ipv4,
215    Ipv6,
216}
217
218impl FromStr for TcpTransportFamily {
219    type Err = Error;
220
221    fn from_str(family: &str) -> Result<Self> {
222        match family {
223            "ipv4" => Ok(Self::Ipv4),
224            "ipv6" => Ok(Self::Ipv6),
225            _ => Err(Error::Address(format!(
226                "invalid tcp address `family`: {family}"
227            ))),
228        }
229    }
230}
231
232impl Display for TcpTransportFamily {
233    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
234        match self {
235            Self::Ipv4 => write!(f, "ipv4"),
236            Self::Ipv6 => write!(f, "ipv6"),
237        }
238    }
239}
zbus/address/transport/tcp.rs

zbus/address/transport/
tcp.rs
