ocpp_client/
connect.rs

1use crate::reconnect::{ReconnectBehavior, ReconnectPolicy, Reconnector};
2use crate::transport::{TransportError, TransportSink, TransportStream};
3use base64::Engine;
4use base64::prelude::BASE64_STANDARD;
5use core::future::Future;
6use core::pin::Pin;
7use std::sync::Arc;
8use std::time::Duration;
9use tokio::net::TcpStream;
10use tokio_tungstenite::tungstenite::client::IntoClientRequest;
11use tokio_tungstenite::tungstenite::http::Request;
12use tokio_tungstenite::tungstenite::http::header::{AUTHORIZATION, SEC_WEBSOCKET_PROTOCOL};
13use tokio_tungstenite::{Connector, MaybeTlsStream, WebSocketStream, client_async_tls_with_config};
14use url::Url;
15
16const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
17
18#[derive(Debug, Clone, Default)]
19pub struct ConnectOptions<'a> {
20    pub username: Option<&'a str>,
21    pub password: Option<&'a str>,
22    pub timeout: Option<Duration>,
23    /// Whether the returned client should reconnect automatically when the WebSocket
24    /// connection drops. Defaults to `ReconnectBehavior::Enabled(ReconnectPolicy::default())` -
25    /// set this to `ReconnectBehavior::Disabled` to get the old one-shot-connection behavior.
26    pub reconnect: ReconnectBehavior,
27    /// Custom TLS trust config for `wss://` addresses. `None` (the default) uses
28    /// `tokio-tungstenite`'s built-in `rustls-tls-webpki-roots` default, which only trusts
29    /// public CAs - it cannot validate a CSMS certificate issued by a private/internal CA.
30    /// Build a `rustls::ClientConfig` with a `RootCertStore` containing that CA's certificate
31    /// (and optionally client-cert auth for mTLS) and set it here to connect to such a CSMS.
32    /// `ocpp_client::rustls` re-exports the exact `rustls` version this crate was built
33    /// against, so the `ClientConfig` you build is guaranteed compatible. Reconnect attempts
34    /// (see `reconnect` above) reuse the same config.
35    pub tls_config: Option<Arc<rustls::ClientConfig>>,
36}
37
38/// A `Client` for whichever OCPP version the server actually picked when connecting via
39/// [`connect`]. Which variants exist depends on which `ocpp_1_6`/`ocpp_2_0_1`/`ocpp_2_1` features
40/// are enabled, same as the version-specific `connect_*` functions.
41pub enum NegotiatedClient {
42    #[cfg(feature = "ocpp_1_6")]
43    V1_6(crate::ocpp_1_6::OCPP1_6Client),
44    #[cfg(feature = "ocpp_2_0_1")]
45    V2_0_1(crate::ocpp_2_0_1::OCPP2_0_1Client),
46    #[cfg(feature = "ocpp_2_1")]
47    V2_1(crate::ocpp_2_1::OCPP2_1Client),
48}
49
50/// An OCPP version `connect` can offer/accept. Only variants for features enabled in this
51/// build exist, same as `NegotiatedClient`'s variants.
52#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub enum OcppVersion {
54    #[cfg(feature = "ocpp_1_6")]
55    V1_6,
56    #[cfg(feature = "ocpp_2_0_1")]
57    V2_0_1,
58    #[cfg(feature = "ocpp_2_1")]
59    V2_1,
60}
61
62impl OcppVersion {
63    fn protocol(self) -> &'static str {
64        match self {
65            #[cfg(feature = "ocpp_1_6")]
66            OcppVersion::V1_6 => "ocpp1.6",
67            #[cfg(feature = "ocpp_2_0_1")]
68            OcppVersion::V2_0_1 => "ocpp2.0.1",
69            #[cfg(feature = "ocpp_2_1")]
70            OcppVersion::V2_1 => "ocpp2.1",
71        }
72    }
73
74    /// Every version compiled into this build, newest first - `connect`'s default set of
75    /// versions to offer when the caller doesn't restrict it via the `versions` argument.
76    #[allow(clippy::vec_init_then_push)]
77    fn all_compiled_in() -> Vec<OcppVersion> {
78        let mut versions = Vec::new();
79        #[cfg(feature = "ocpp_2_1")]
80        versions.push(OcppVersion::V2_1);
81        #[cfg(feature = "ocpp_2_0_1")]
82        versions.push(OcppVersion::V2_0_1);
83        #[cfg(feature = "ocpp_1_6")]
84        versions.push(OcppVersion::V1_6);
85        versions
86    }
87}
88
89/// Connect to an OCPP server over WebSocket, offering the given `versions` (or, if `None`,
90/// every version compiled into this crate via its `ocpp_1_6`/`ocpp_2_0_1`/`ocpp_2_1` features)
91/// in the `Sec-WebSocket-Protocol` header, and using whichever one the server picks - rather
92/// than requiring the caller to already know the server's supported version like
93/// `connect_1_6`/`connect_2_0_1`/`connect_2_1` do. `versions` also controls preference order
94/// (offered in the slice's order); the choice among the offered set is entirely the server's
95/// per RFC 6455.
96pub async fn connect(
97    address: &str,
98    versions: Option<&[OcppVersion]>,
99    options: Option<ConnectOptions<'_>>,
100) -> Result<NegotiatedClient, Box<dyn std::error::Error + Send + Sync>> {
101    let all_compiled_in;
102    let versions = match versions {
103        Some(versions) => versions,
104        None => {
105            all_compiled_in = OcppVersion::all_compiled_in();
106            &all_compiled_in
107        }
108    };
109    let offered = versions
110        .iter()
111        .map(|v| v.protocol())
112        .collect::<Vec<_>>()
113        .join(", ");
114    let (stream, negotiated) = setup_socket(address, &offered, options.clone()).await?;
115    let protocol = versions
116        .iter()
117        .find(|v| v.protocol() == negotiated)
118        .map(|v| v.protocol())
119        .ok_or_else(|| format!("Server negotiated unsupported protocol: {negotiated}"))?;
120    let (timeout, reconnector, policy) = prepare(address, protocol, options);
121    let (sink, source) = crate::transport::websocket::split(stream);
122
123    Ok(match protocol {
124        #[cfg(feature = "ocpp_1_6")]
125        "ocpp1.6" => NegotiatedClient::V1_6(crate::Client::from_transport_with_reconnect(
126            sink,
127            source,
128            timeout,
129            Box::new(crate::runtime::tokio::TokioExecutor),
130            Box::new(crate::runtime::tokio::TokioTimer),
131            reconnector,
132            policy,
133        )),
134        #[cfg(feature = "ocpp_2_0_1")]
135        "ocpp2.0.1" => NegotiatedClient::V2_0_1(crate::Client::from_transport_with_reconnect(
136            sink,
137            source,
138            timeout,
139            Box::new(crate::runtime::tokio::TokioExecutor),
140            Box::new(crate::runtime::tokio::TokioTimer),
141            reconnector,
142            policy,
143        )),
144        #[cfg(feature = "ocpp_2_1")]
145        "ocpp2.1" => NegotiatedClient::V2_1(crate::Client::from_transport_with_reconnect(
146            sink,
147            source,
148            timeout,
149            Box::new(crate::runtime::tokio::TokioExecutor),
150            Box::new(crate::runtime::tokio::TokioTimer),
151            reconnector,
152            policy,
153        )),
154        _ => unreachable!("protocol only ever holds a value returned by OcppVersion::protocol"),
155    })
156}
157
158/// Connect to an OCPP 1.6 server over WebSocket.
159#[cfg(feature = "ocpp_1_6")]
160pub async fn connect_1_6(
161    address: &str,
162    options: Option<ConnectOptions<'_>>,
163) -> Result<crate::ocpp_1_6::OCPP1_6Client, Box<dyn std::error::Error + Send + Sync>> {
164    let (timeout, reconnector, policy) = prepare(address, "ocpp1.6", options.clone());
165    let (stream, _protocol) = setup_socket(address, "ocpp1.6", options).await?;
166    let (sink, source) = crate::transport::websocket::split(stream);
167    Ok(crate::Client::from_transport_with_reconnect(
168        sink,
169        source,
170        timeout,
171        Box::new(crate::runtime::tokio::TokioExecutor),
172        Box::new(crate::runtime::tokio::TokioTimer),
173        reconnector,
174        policy,
175    ))
176}
177
178/// Connect to an OCPP 2.0.1 server over WebSocket.
179#[cfg(feature = "ocpp_2_0_1")]
180pub async fn connect_2_0_1(
181    address: &str,
182    options: Option<ConnectOptions<'_>>,
183) -> Result<crate::ocpp_2_0_1::OCPP2_0_1Client, Box<dyn std::error::Error + Send + Sync>> {
184    let (timeout, reconnector, policy) = prepare(address, "ocpp2.0.1", options.clone());
185    let (stream, _protocol) = setup_socket(address, "ocpp2.0.1", options).await?;
186    let (sink, source) = crate::transport::websocket::split(stream);
187    Ok(crate::Client::from_transport_with_reconnect(
188        sink,
189        source,
190        timeout,
191        Box::new(crate::runtime::tokio::TokioExecutor),
192        Box::new(crate::runtime::tokio::TokioTimer),
193        reconnector,
194        policy,
195    ))
196}
197
198/// Connect to an OCPP 2.1 server over WebSocket.
199#[cfg(feature = "ocpp_2_1")]
200pub async fn connect_2_1(
201    address: &str,
202    options: Option<ConnectOptions<'_>>,
203) -> Result<crate::ocpp_2_1::OCPP2_1Client, Box<dyn std::error::Error + Send + Sync>> {
204    let (timeout, reconnector, policy) = prepare(address, "ocpp2.1", options.clone());
205    let (stream, _protocol) = setup_socket(address, "ocpp2.1", options).await?;
206    let (sink, source) = crate::transport::websocket::split(stream);
207    Ok(crate::Client::from_transport_with_reconnect(
208        sink,
209        source,
210        timeout,
211        Box::new(crate::runtime::tokio::TokioExecutor),
212        Box::new(crate::runtime::tokio::TokioTimer),
213        reconnector,
214        policy,
215    ))
216}
217
218/// Pulls the timeout/reconnect settings out of `options` and, if reconnect is enabled, builds
219/// the `Reconnector` that redials this same address/protocol/credentials. Shared by all three
220/// `connect_*` entry points.
221fn prepare(
222    address: &str,
223    protocol: &'static str,
224    options: Option<ConnectOptions<'_>>,
225) -> (Duration, Option<Box<dyn Reconnector>>, ReconnectPolicy) {
226    let timeout = options
227        .as_ref()
228        .and_then(|o| o.timeout)
229        .unwrap_or(DEFAULT_TIMEOUT);
230    let reconnect = options.as_ref().map(|o| o.reconnect).unwrap_or_default();
231    let username = options
232        .as_ref()
233        .and_then(|o| o.username)
234        .map(str::to_string);
235    let password = options
236        .as_ref()
237        .and_then(|o| o.password)
238        .map(str::to_string);
239    let tls_config = options.as_ref().and_then(|o| o.tls_config.clone());
240
241    match reconnect {
242        ReconnectBehavior::Disabled => (timeout, None, ReconnectPolicy::default()),
243        ReconnectBehavior::Enabled(policy) => {
244            let reconnector: Box<dyn Reconnector> = Box::new(WebSocketReconnector {
245                address: address.to_string(),
246                protocol,
247                username,
248                password,
249                tls_config,
250            });
251            (timeout, Some(reconnector), policy)
252        }
253    }
254}
255
256/// Redials `address` with the original protocol/credentials/TLS config whenever `Client`'s
257/// background read loop needs a fresh transport after a disconnect.
258struct WebSocketReconnector {
259    address: String,
260    protocol: &'static str,
261    username: Option<String>,
262    password: Option<String>,
263    tls_config: Option<Arc<rustls::ClientConfig>>,
264}
265
266impl Reconnector for WebSocketReconnector {
267    fn connect<'a>(
268        &'a self,
269    ) -> Pin<
270        Box<
271            dyn Future<
272                    Output = Result<
273                        (Box<dyn TransportSink>, Box<dyn TransportStream>),
274                        TransportError,
275                    >,
276                > + Send
277                + 'a,
278        >,
279    > {
280        Box::pin(async move {
281            let options = ConnectOptions {
282                username: self.username.as_deref(),
283                password: self.password.as_deref(),
284                timeout: None,
285                reconnect: ReconnectBehavior::Disabled,
286                tls_config: self.tls_config.clone(),
287            };
288            let (stream, _protocol) =
289                setup_socket(&self.address, self.protocol, Some(options)).await?;
290            Ok(crate::transport::websocket::split(stream))
291        })
292    }
293}
294
295async fn setup_socket(
296    address: &str,
297    protocols: &str,
298    options: Option<ConnectOptions<'_>>,
299) -> Result<
300    (WebSocketStream<MaybeTlsStream<TcpStream>>, String),
301    Box<dyn std::error::Error + Send + Sync>,
302> {
303    let address = Url::parse(address)?;
304
305    let socket_addrs = address.socket_addrs(|| None)?;
306    let stream = TcpStream::connect(&*socket_addrs).await?;
307
308    let mut request: Request<()> = address.to_string().into_client_request()?;
309    request
310        .headers_mut()
311        .insert(SEC_WEBSOCKET_PROTOCOL, protocols.parse()?);
312    let mut tls_config = None;
313    if let Some(options) = options {
314        if let Some(username) = options.username {
315            let data = format!("{}:{}", username, options.password.unwrap_or(""));
316            let encoded = BASE64_STANDARD.encode(data);
317            request
318                .headers_mut()
319                .insert(AUTHORIZATION, format!("Basic {encoded}").parse()?);
320        }
321        tls_config = options.tls_config;
322    }
323
324    let connector = tls_config.map(Connector::Rustls);
325    let (stream, response) = client_async_tls_with_config(request, stream, None, connector).await?;
326
327    let protocol = response
328        .headers()
329        .get(SEC_WEBSOCKET_PROTOCOL)
330        .ok_or("No OCPP protocol negotiated")?;
331
332    Ok((stream, protocol.to_str()?.to_string()))
333}
ocpp_client/connect.rs

ocpp_client/
connect.rs
