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(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    /// Decides where a dropped connection is redialled. `None` (the default) redials the same
37    /// address, protocol, credentials and TLS config the connection started with, which is what
38    /// almost every caller wants.
39    ///
40    /// Supply one to make the redial target something other than a constant - the case this
41    /// exists for is a charge point that must move to a different CSMS address (an OCPP 2.x
42    /// network connection profile, a failover endpoint) **without tearing down its `Client`**.
43    /// Reconnecting through the same `Client` keeps its identity, so every registered handler,
44    /// every in-flight request and every queued message survives the move; dropping the client
45    /// and calling `connect_*` again does not.
46    ///
47    /// Implementations usually delegate to [`websocket_transport`] rather than building a
48    /// transport by hand - see its docs for a worked reconnector. A custom reconnector is fully
49    /// responsible for the redial: `address` and the credential/TLS fields above apply to the
50    /// *initial* connection only and are not consulted when it runs.
51    ///
52    /// `ReconnectBehavior::Disabled` still wins - it is an explicit "do not redial", and
53    /// supplying a reconnector does not quietly turn reconnect back on.
54    pub reconnector: Option<Arc<dyn Reconnector>>,
55}
56
57impl core::fmt::Debug for ConnectOptions<'_> {
58    /// Redacts `password`: these options are routinely logged wholesale when a connection fails,
59    /// and a CSMS credential in a log file outlives the debugging session that produced it.
60    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
61        f.debug_struct("ConnectOptions")
62            .field("username", &self.username)
63            .field(
64                "password",
65                &self.password.map(|_| "<redacted>").unwrap_or("None"),
66            )
67            .field("timeout", &self.timeout)
68            .field("reconnect", &self.reconnect)
69            .field("tls_config", &self.tls_config.as_ref().map(|_| "<set>"))
70            .field(
71                "reconnector",
72                &self.reconnector.as_ref().map(|_| "<custom>"),
73            )
74            .finish()
75    }
76}
77
78/// Opens one WebSocket connection to `address` speaking `version`, and hands back the transport
79/// halves a [`Reconnector`] is required to return - so a custom reconnector can redial without
80/// reimplementing this crate's WebSocket plumbing.
81///
82/// `options` covers credentials and TLS for this dial; its `reconnect`/`reconnector` fields are
83/// ignored, since this opens a bare transport rather than a `Client`. `version` must be the one
84/// already negotiated on the connection being replaced - the client's handlers are bound to that
85/// version, so redialling a different one would leave it speaking the wrong protocol.
86///
87/// ```no_run
88/// # use std::sync::{Arc, Mutex};
89/// # use std::{future::Future, pin::Pin};
90/// # use ocpp_client::{OcppVersion, Reconnector, TransportError, TransportSink, TransportStream, websocket_transport};
91/// /// Redials whatever address it is currently pointed at.
92/// struct SwitchableReconnector(Arc<Mutex<String>>);
93///
94/// impl Reconnector for SwitchableReconnector {
95///     fn connect<'a>(&'a self) -> Pin<Box<dyn Future<Output = Result<(Box<dyn TransportSink>, Box<dyn TransportStream>), TransportError>> + Send + 'a>> {
96///         Box::pin(async move {
97///             let address = self.0.lock().unwrap().clone();
98///             websocket_transport(&address, OcppVersion::V1_6, None).await
99///         })
100///     }
101/// }
102/// ```
103pub async fn websocket_transport(
104    address: &str,
105    version: OcppVersion,
106    options: Option<ConnectOptions<'_>>,
107) -> Result<(Box<dyn TransportSink>, Box<dyn TransportStream>), TransportError> {
108    let (stream, _protocol) = setup_socket(address, version.protocol(), options).await?;
109    Ok(crate::transport::websocket::split(stream))
110}
111
112/// A `Client` for whichever OCPP version the server actually picked when connecting via
113/// [`connect`]. Which variants exist depends on which `ocpp_1_6`/`ocpp_2_0_1`/`ocpp_2_1` features
114/// are enabled, same as the version-specific `connect_*` functions.
115pub enum NegotiatedClient {
116    #[cfg(feature = "ocpp_1_6")]
117    V1_6(crate::ocpp_1_6::OCPP1_6Client),
118    #[cfg(feature = "ocpp_2_0_1")]
119    V2_0_1(crate::ocpp_2_0_1::OCPP2_0_1Client),
120    #[cfg(feature = "ocpp_2_1")]
121    V2_1(crate::ocpp_2_1::OCPP2_1Client),
122}
123
124/// An OCPP version `connect` can offer/accept. Only variants for features enabled in this
125/// build exist, same as `NegotiatedClient`'s variants.
126#[derive(Debug, Clone, Copy, PartialEq, Eq)]
127pub enum OcppVersion {
128    #[cfg(feature = "ocpp_1_6")]
129    V1_6,
130    #[cfg(feature = "ocpp_2_0_1")]
131    V2_0_1,
132    #[cfg(feature = "ocpp_2_1")]
133    V2_1,
134}
135
136impl OcppVersion {
137    fn protocol(self) -> &'static str {
138        match self {
139            #[cfg(feature = "ocpp_1_6")]
140            OcppVersion::V1_6 => "ocpp1.6",
141            #[cfg(feature = "ocpp_2_0_1")]
142            OcppVersion::V2_0_1 => "ocpp2.0.1",
143            #[cfg(feature = "ocpp_2_1")]
144            OcppVersion::V2_1 => "ocpp2.1",
145        }
146    }
147
148    /// Every version compiled into this build, newest first - `connect`'s default set of
149    /// versions to offer when the caller doesn't restrict it via the `versions` argument.
150    #[allow(clippy::vec_init_then_push)]
151    fn all_compiled_in() -> Vec<OcppVersion> {
152        let mut versions = Vec::new();
153        #[cfg(feature = "ocpp_2_1")]
154        versions.push(OcppVersion::V2_1);
155        #[cfg(feature = "ocpp_2_0_1")]
156        versions.push(OcppVersion::V2_0_1);
157        #[cfg(feature = "ocpp_1_6")]
158        versions.push(OcppVersion::V1_6);
159        versions
160    }
161}
162
163/// Connect to an OCPP server over WebSocket, offering the given `versions` (or, if `None`,
164/// every version compiled into this crate via its `ocpp_1_6`/`ocpp_2_0_1`/`ocpp_2_1` features)
165/// in the `Sec-WebSocket-Protocol` header, and using whichever one the server picks - rather
166/// than requiring the caller to already know the server's supported version like
167/// `connect_1_6`/`connect_2_0_1`/`connect_2_1` do. `versions` also controls preference order
168/// (offered in the slice's order); the choice among the offered set is entirely the server's
169/// per RFC 6455.
170pub async fn connect(
171    address: &str,
172    versions: Option<&[OcppVersion]>,
173    options: Option<ConnectOptions<'_>>,
174) -> Result<NegotiatedClient, Box<dyn std::error::Error + Send + Sync>> {
175    let all_compiled_in;
176    let versions = match versions {
177        Some(versions) => versions,
178        None => {
179            all_compiled_in = OcppVersion::all_compiled_in();
180            &all_compiled_in
181        }
182    };
183    let offered = versions
184        .iter()
185        .map(|v| v.protocol())
186        .collect::<Vec<_>>()
187        .join(", ");
188    let (stream, negotiated) = setup_socket(address, &offered, options.clone()).await?;
189    let protocol = versions
190        .iter()
191        .find(|v| v.protocol() == negotiated)
192        .map(|v| v.protocol())
193        .ok_or_else(|| format!("Server negotiated unsupported protocol: {negotiated}"))?;
194    let (timeout, reconnector, policy) = prepare(address, protocol, options);
195    let (sink, source) = crate::transport::websocket::split(stream);
196
197    Ok(match protocol {
198        #[cfg(feature = "ocpp_1_6")]
199        "ocpp1.6" => NegotiatedClient::V1_6(crate::Client::from_transport_with_reconnect(
200            sink,
201            source,
202            timeout,
203            Box::new(crate::runtime::tokio::TokioExecutor),
204            Box::new(crate::runtime::tokio::TokioTimer),
205            reconnector,
206            policy,
207        )),
208        #[cfg(feature = "ocpp_2_0_1")]
209        "ocpp2.0.1" => NegotiatedClient::V2_0_1(crate::Client::from_transport_with_reconnect(
210            sink,
211            source,
212            timeout,
213            Box::new(crate::runtime::tokio::TokioExecutor),
214            Box::new(crate::runtime::tokio::TokioTimer),
215            reconnector,
216            policy,
217        )),
218        #[cfg(feature = "ocpp_2_1")]
219        "ocpp2.1" => NegotiatedClient::V2_1(crate::Client::from_transport_with_reconnect(
220            sink,
221            source,
222            timeout,
223            Box::new(crate::runtime::tokio::TokioExecutor),
224            Box::new(crate::runtime::tokio::TokioTimer),
225            reconnector,
226            policy,
227        )),
228        _ => unreachable!("protocol only ever holds a value returned by OcppVersion::protocol"),
229    })
230}
231
232/// Connect to an OCPP 1.6 server over WebSocket.
233#[cfg(feature = "ocpp_1_6")]
234pub async fn connect_1_6(
235    address: &str,
236    options: Option<ConnectOptions<'_>>,
237) -> Result<crate::ocpp_1_6::OCPP1_6Client, Box<dyn std::error::Error + Send + Sync>> {
238    let (timeout, reconnector, policy) = prepare(address, "ocpp1.6", options.clone());
239    let (stream, _protocol) = setup_socket(address, "ocpp1.6", options).await?;
240    let (sink, source) = crate::transport::websocket::split(stream);
241    Ok(crate::Client::from_transport_with_reconnect(
242        sink,
243        source,
244        timeout,
245        Box::new(crate::runtime::tokio::TokioExecutor),
246        Box::new(crate::runtime::tokio::TokioTimer),
247        reconnector,
248        policy,
249    ))
250}
251
252/// Connect to an OCPP 2.0.1 server over WebSocket.
253#[cfg(feature = "ocpp_2_0_1")]
254pub async fn connect_2_0_1(
255    address: &str,
256    options: Option<ConnectOptions<'_>>,
257) -> Result<crate::ocpp_2_0_1::OCPP2_0_1Client, Box<dyn std::error::Error + Send + Sync>> {
258    let (timeout, reconnector, policy) = prepare(address, "ocpp2.0.1", options.clone());
259    let (stream, _protocol) = setup_socket(address, "ocpp2.0.1", options).await?;
260    let (sink, source) = crate::transport::websocket::split(stream);
261    Ok(crate::Client::from_transport_with_reconnect(
262        sink,
263        source,
264        timeout,
265        Box::new(crate::runtime::tokio::TokioExecutor),
266        Box::new(crate::runtime::tokio::TokioTimer),
267        reconnector,
268        policy,
269    ))
270}
271
272/// Connect to an OCPP 2.1 server over WebSocket.
273#[cfg(feature = "ocpp_2_1")]
274pub async fn connect_2_1(
275    address: &str,
276    options: Option<ConnectOptions<'_>>,
277) -> Result<crate::ocpp_2_1::OCPP2_1Client, Box<dyn std::error::Error + Send + Sync>> {
278    let (timeout, reconnector, policy) = prepare(address, "ocpp2.1", options.clone());
279    let (stream, _protocol) = setup_socket(address, "ocpp2.1", options).await?;
280    let (sink, source) = crate::transport::websocket::split(stream);
281    Ok(crate::Client::from_transport_with_reconnect(
282        sink,
283        source,
284        timeout,
285        Box::new(crate::runtime::tokio::TokioExecutor),
286        Box::new(crate::runtime::tokio::TokioTimer),
287        reconnector,
288        policy,
289    ))
290}
291
292/// Pulls the timeout/reconnect settings out of `options` and, if reconnect is enabled, builds
293/// the `Reconnector` that redials this same address/protocol/credentials. Shared by all three
294/// `connect_*` entry points.
295fn prepare(
296    address: &str,
297    protocol: &'static str,
298    options: Option<ConnectOptions<'_>>,
299) -> (Duration, Option<Box<dyn Reconnector>>, ReconnectPolicy) {
300    let timeout = options
301        .as_ref()
302        .and_then(|o| o.timeout)
303        .unwrap_or(DEFAULT_TIMEOUT);
304    let reconnect = options.as_ref().map(|o| o.reconnect).unwrap_or_default();
305    let username = options
306        .as_ref()
307        .and_then(|o| o.username)
308        .map(str::to_string);
309    let password = options
310        .as_ref()
311        .and_then(|o| o.password)
312        .map(str::to_string);
313    let tls_config = options.as_ref().and_then(|o| o.tls_config.clone());
314
315    let custom = options.as_ref().and_then(|o| o.reconnector.clone());
316
317    match reconnect {
318        ReconnectBehavior::Disabled => (timeout, None, ReconnectPolicy::default()),
319        ReconnectBehavior::Enabled(policy) if custom.is_some() => {
320            let custom = custom.expect("guarded by the match arm");
321            (timeout, Some(Box::new(SharedReconnector(custom))), policy)
322        }
323        ReconnectBehavior::Enabled(policy) => {
324            let reconnector: Box<dyn Reconnector> = Box::new(WebSocketReconnector {
325                address: address.to_string(),
326                protocol,
327                username,
328                password,
329                tls_config,
330            });
331            (timeout, Some(reconnector), policy)
332        }
333    }
334}
335
336/// Adapts the `Arc<dyn Reconnector>` callers hand us in [`ConnectOptions::reconnector`] to the
337/// `Box<dyn Reconnector>` `Client` takes. `Arc` is what keeps `ConnectOptions` cloneable, which
338/// the `connect_*` functions rely on.
339struct SharedReconnector(Arc<dyn Reconnector>);
340
341impl Reconnector for SharedReconnector {
342    fn connect<'a>(
343        &'a self,
344    ) -> Pin<
345        Box<
346            dyn Future<
347                    Output = Result<
348                        (Box<dyn TransportSink>, Box<dyn TransportStream>),
349                        TransportError,
350                    >,
351                > + Send
352                + 'a,
353        >,
354    > {
355        self.0.connect()
356    }
357}
358
359/// Redials `address` with the original protocol/credentials/TLS config whenever `Client`'s
360/// background read loop needs a fresh transport after a disconnect.
361struct WebSocketReconnector {
362    address: String,
363    protocol: &'static str,
364    username: Option<String>,
365    password: Option<String>,
366    tls_config: Option<Arc<rustls::ClientConfig>>,
367}
368
369impl Reconnector for WebSocketReconnector {
370    fn connect<'a>(
371        &'a self,
372    ) -> Pin<
373        Box<
374            dyn Future<
375                    Output = Result<
376                        (Box<dyn TransportSink>, Box<dyn TransportStream>),
377                        TransportError,
378                    >,
379                > + Send
380                + 'a,
381        >,
382    > {
383        Box::pin(async move {
384            let options = ConnectOptions {
385                username: self.username.as_deref(),
386                password: self.password.as_deref(),
387                timeout: None,
388                reconnect: ReconnectBehavior::Disabled,
389                tls_config: self.tls_config.clone(),
390                reconnector: None,
391            };
392            let (stream, _protocol) =
393                setup_socket(&self.address, self.protocol, Some(options)).await?;
394            Ok(crate::transport::websocket::split(stream))
395        })
396    }
397}
398
399async fn setup_socket(
400    address: &str,
401    protocols: &str,
402    options: Option<ConnectOptions<'_>>,
403) -> Result<
404    (WebSocketStream<MaybeTlsStream<TcpStream>>, String),
405    Box<dyn std::error::Error + Send + Sync>,
406> {
407    let address = Url::parse(address)?;
408
409    let socket_addrs = address.socket_addrs(|| None)?;
410    let stream = TcpStream::connect(&*socket_addrs).await?;
411
412    let mut request: Request<()> = address.to_string().into_client_request()?;
413    request
414        .headers_mut()
415        .insert(SEC_WEBSOCKET_PROTOCOL, protocols.parse()?);
416    let mut tls_config = None;
417    if let Some(options) = options {
418        if let Some(username) = options.username {
419            let data = format!("{}:{}", username, options.password.unwrap_or(""));
420            let encoded = BASE64_STANDARD.encode(data);
421            request
422                .headers_mut()
423                .insert(AUTHORIZATION, format!("Basic {encoded}").parse()?);
424        }
425        tls_config = options.tls_config;
426    }
427
428    let connector = tls_config.map(Connector::Rustls);
429    let (stream, response) = client_async_tls_with_config(request, stream, None, connector).await?;
430
431    let protocol = response
432        .headers()
433        .get(SEC_WEBSOCKET_PROTOCOL)
434        .ok_or("No OCPP protocol negotiated")?;
435
436    Ok((stream, protocol.to_str()?.to_string()))
437}
ocpp_client/connect.rs

ocpp_client/
connect.rs
