rumqttc_core/
proxy.rs

1use crate::default_socket_connect;
2use crate::{AsyncReadWrite, NetworkOptions, SocketConnector};
3
4use std::io;
5
6#[cfg(any(feature = "use-rustls-no-provider", feature = "use-native-tls"))]
7use crate::{TlsConfiguration, tls};
8
9#[derive(Clone, Debug)]
10pub struct Proxy {
11    pub ty: ProxyType,
12    pub auth: ProxyAuth,
13    pub addr: String,
14    pub port: u16,
15}
16
17#[derive(Clone, Debug)]
18pub enum ProxyType {
19    Http,
20    #[cfg(any(feature = "use-rustls-no-provider", feature = "use-native-tls"))]
21    Https(TlsConfiguration),
22}
23
24#[derive(Clone, Debug)]
25pub enum ProxyAuth {
26    None,
27    Basic { username: String, password: String },
28}
29
30#[derive(Debug, thiserror::Error)]
31pub enum ProxyError {
32    #[error("Socket connect: {0}.")]
33    Io(#[from] io::Error),
34    #[error("Proxy connect: {0}.")]
35    Proxy(#[from] async_http_proxy::HttpError),
36
37    #[cfg(any(feature = "use-rustls-no-provider", feature = "use-native-tls"))]
38    #[error("Tls connect: {0}.")]
39    Tls(#[from] tls::Error),
40}
41
42impl Proxy {
43    pub async fn connect(
44        self,
45        broker_addr: &str,
46        broker_port: u16,
47        network_options: NetworkOptions,
48        socket_connector: Option<SocketConnector>,
49    ) -> Result<Box<dyn AsyncReadWrite>, ProxyError> {
50        let proxy_addr = format!("{}:{}", self.addr, self.port);
51
52        let tcp: Box<dyn AsyncReadWrite> = if let Some(connector) = socket_connector {
53            connector(proxy_addr, network_options).await?
54        } else {
55            Box::new(default_socket_connect(proxy_addr, network_options).await?)
56        };
57        let mut tcp = match self.ty {
58            ProxyType::Http => tcp,
59            #[cfg(any(feature = "use-rustls-no-provider", feature = "use-native-tls"))]
60            ProxyType::Https(tls_config) => {
61                tls::tls_connect(&self.addr, self.port, &tls_config, tcp).await?
62            }
63        };
64        self.auth.auth(broker_addr, broker_port, &mut tcp).await?;
65        Ok(tcp)
66    }
67}
68
69impl ProxyAuth {
70    async fn auth(
71        self,
72        host: &str,
73        port: u16,
74        tcp_stream: &mut Box<dyn AsyncReadWrite>,
75    ) -> Result<(), ProxyError> {
76        match self {
77            Self::None => async_http_proxy::http_connect_tokio(tcp_stream, host, port).await?,
78            Self::Basic { username, password } => {
79                async_http_proxy::http_connect_tokio_with_basic_auth(
80                    tcp_stream, host, port, &username, &password,
81                )
82                .await?
83            }
84        }
85        Ok(())
86    }
87}
rumqttc_core/proxy.rs

rumqttc_core/
proxy.rs
