ocpp_client/
connect.rs

1use crate::client::ClientConfig;
2use crate::keepalive::KeepaliveBehavior;
3use crate::reconnect::{ReconnectBehavior, Reconnector};
4use crate::transport::{TransportError, TransportSink, TransportStream};
5use base64::Engine;
6use base64::prelude::BASE64_STANDARD;
7use core::future::Future;
8use core::pin::Pin;
9use std::sync::Arc;
10use std::time::Duration;
11use tokio::net::TcpStream;
12use tokio_tungstenite::tungstenite::client::IntoClientRequest;
13use tokio_tungstenite::tungstenite::http::Request;
14use tokio_tungstenite::tungstenite::http::header::{AUTHORIZATION, SEC_WEBSOCKET_PROTOCOL};
15use tokio_tungstenite::{Connector, MaybeTlsStream, WebSocketStream, client_async_tls_with_config};
16use url::Url;
17
18const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
19
20#[derive(Clone)]
21pub struct ConnectOptions<'a> {
22    pub username: Option<&'a str>,
23    /// The HTTP Basic password, as **bytes**. OCPP configures it as the hexadecimal representation
24    /// of an authorization key and the charge point sends the decoded bytes (20 for a 1.6
25    /// `AuthorizationKey`), which are almost never valid UTF-8 - so pass the decoded key, not its
26    /// hex text. A textual password is `Some(password.as_bytes())`.
27    pub password: Option<&'a [u8]>,
28    pub timeout: Option<Duration>,
29    /// Whether the returned client should reconnect automatically when the WebSocket
30    /// connection drops. Defaults to `ReconnectBehavior::Enabled(ReconnectPolicy::default())` -
31    /// set this to `ReconnectBehavior::Disabled` to get the old one-shot-connection behavior.
32    pub reconnect: ReconnectBehavior,
33    /// Custom TLS trust config for `wss://` addresses. `None` (the default) uses
34    /// `tokio-tungstenite`'s built-in `rustls-tls-webpki-roots` default, which only trusts
35    /// public CAs - it cannot validate a CSMS certificate issued by a private/internal CA.
36    /// Build a `rustls::ClientConfig` with a `RootCertStore` containing that CA's certificate
37    /// (and optionally client-cert auth for mTLS) and set it here to connect to such a CSMS.
38    /// `ocpp_client::rustls` re-exports the exact `rustls` version this crate was built
39    /// against, so the `ClientConfig` you build is guaranteed compatible. Reconnect attempts
40    /// (see `reconnect` above) reuse the same config.
41    pub tls_config: Option<Arc<rustls::ClientConfig>>,
42    /// Decides where a dropped connection is redialled. `None` (the default) redials the same
43    /// address, protocol, credentials and TLS config the connection started with, which is what
44    /// almost every caller wants.
45    ///
46    /// Supply one to make the redial target something other than a constant - the case this
47    /// exists for is a charge point that must move to a different CSMS address (an OCPP 2.x
48    /// network connection profile, a failover endpoint) **without tearing down its `Client`**.
49    /// Reconnecting through the same `Client` keeps its identity, so every registered handler,
50    /// every in-flight request and every queued message survives the move; dropping the client
51    /// and calling `connect_*` again does not.
52    ///
53    /// Implementations usually delegate to [`websocket_transport`] rather than building a
54    /// transport by hand - see its docs for a worked reconnector. A custom reconnector is fully
55    /// responsible for the redial: `address` and the credential/TLS fields above apply to the
56    /// *initial* connection only and are not consulted when it runs.
57    ///
58    /// `ReconnectBehavior::Disabled` still wins - it is an explicit "do not redial", and
59    /// supplying a reconnector does not quietly turn reconnect back on.
60    pub reconnector: Option<Arc<dyn Reconnector>>,
61    /// Whether the client pings the CSMS on a schedule, and when it gives up on an unresponsive
62    /// one and redials. Defaults to `KeepaliveBehavior::Enabled` with
63    /// `KeepalivePolicy::default()` - a 60-second interval, tolerating one missed pong.
64    ///
65    /// Keepalive is on by default here for the same reason `reconnect` is: without it, a
66    /// half-open connection (a NAT table entry dropped, a mobile link that went away without a
67    /// FIN) is invisible, and the reconnect machinery never gets a chance to fire because the
68    /// read loop is parked in `recv` until the OS TCP timeout. Set `KeepaliveBehavior::Disabled`
69    /// if the CSMS pings the charge point instead, or if the deployment forbids unsolicited
70    /// traffic.
71    ///
72    /// The interval is also readable and writable at runtime via `Client::ping_interval`/
73    /// `Client::set_ping_interval`, which is how `WebSocketPingInterval` gets reported to, and
74    /// updated by, a CSMS.
75    pub keepalive: KeepaliveBehavior,
76}
77
78/// Hand-written rather than derived because `keepalive` defaults to *enabled*, unlike
79/// `KeepaliveBehavior`'s own `Default` - see the field's docs for why the WebSocket convenience
80/// path opts in where the bare `Client` constructors don't.
81impl Default for ConnectOptions<'_> {
82    fn default() -> Self {
83        Self {
84            username: None,
85            password: None,
86            timeout: None,
87            reconnect: ReconnectBehavior::default(),
88            tls_config: None,
89            reconnector: None,
90            keepalive: KeepaliveBehavior::Enabled(crate::keepalive::KeepalivePolicy::default()),
91        }
92    }
93}
94
95impl core::fmt::Debug for ConnectOptions<'_> {
96    /// Redacts `password`: these options are routinely logged wholesale when a connection fails,
97    /// and a CSMS credential in a log file outlives the debugging session that produced it.
98    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
99        f.debug_struct("ConnectOptions")
100            .field("username", &self.username)
101            .field(
102                "password",
103                &self.password.map(|_| "<redacted>").unwrap_or("None"),
104            )
105            .field("timeout", &self.timeout)
106            .field("reconnect", &self.reconnect)
107            .field("tls_config", &self.tls_config.as_ref().map(|_| "<set>"))
108            .field(
109                "reconnector",
110                &self.reconnector.as_ref().map(|_| "<custom>"),
111            )
112            .field("keepalive", &self.keepalive)
113            .finish()
114    }
115}
116
117/// Opens one WebSocket connection to `address` speaking `version`, and hands back the transport
118/// halves a [`Reconnector`] is required to return - so a custom reconnector can redial without
119/// reimplementing this crate's WebSocket plumbing.
120///
121/// `options` covers credentials and TLS for this dial; its `reconnect`/`reconnector` fields are
122/// ignored, since this opens a bare transport rather than a `Client`. `version` must be the one
123/// already negotiated on the connection being replaced - the client's handlers are bound to that
124/// version, so redialling a different one would leave it speaking the wrong protocol.
125///
126/// ```no_run
127/// # use std::sync::{Arc, Mutex};
128/// # use std::{future::Future, pin::Pin};
129/// # use ocpp_client::{OcppVersion, Reconnector, TransportError, TransportSink, TransportStream, websocket_transport};
130/// /// Redials whatever address it is currently pointed at.
131/// struct SwitchableReconnector(Arc<Mutex<String>>);
132///
133/// impl Reconnector for SwitchableReconnector {
134///     fn connect<'a>(&'a self) -> Pin<Box<dyn Future<Output = Result<(Box<dyn TransportSink>, Box<dyn TransportStream>), TransportError>> + Send + 'a>> {
135///         Box::pin(async move {
136///             let address = self.0.lock().unwrap().clone();
137///             websocket_transport(&address, OcppVersion::V1_6, None).await
138///         })
139///     }
140/// }
141/// ```
142pub async fn websocket_transport(
143    address: &str,
144    version: OcppVersion,
145    options: Option<ConnectOptions<'_>>,
146) -> Result<(Box<dyn TransportSink>, Box<dyn TransportStream>), TransportError> {
147    let (stream, _protocol) = setup_socket(address, version.protocol(), options).await?;
148    Ok(crate::transport::websocket::split(stream))
149}
150
151/// A `Client` for whichever OCPP version the server actually picked when connecting via
152/// [`connect`]. Which variants exist depends on which `ocpp_1_6`/`ocpp_2_0_1`/`ocpp_2_1` features
153/// are enabled, same as the version-specific `connect_*` functions.
154pub enum NegotiatedClient {
155    #[cfg(feature = "ocpp_1_6")]
156    V1_6(crate::ocpp_1_6::OCPP1_6Client),
157    #[cfg(feature = "ocpp_2_0_1")]
158    V2_0_1(crate::ocpp_2_0_1::OCPP2_0_1Client),
159    #[cfg(feature = "ocpp_2_1")]
160    V2_1(crate::ocpp_2_1::OCPP2_1Client),
161}
162
163/// An OCPP version `connect` can offer/accept. Only variants for features enabled in this
164/// build exist, same as `NegotiatedClient`'s variants.
165#[derive(Debug, Clone, Copy, PartialEq, Eq)]
166pub enum OcppVersion {
167    #[cfg(feature = "ocpp_1_6")]
168    V1_6,
169    #[cfg(feature = "ocpp_2_0_1")]
170    V2_0_1,
171    #[cfg(feature = "ocpp_2_1")]
172    V2_1,
173}
174
175impl OcppVersion {
176    fn protocol(self) -> &'static str {
177        match self {
178            #[cfg(feature = "ocpp_1_6")]
179            OcppVersion::V1_6 => "ocpp1.6",
180            #[cfg(feature = "ocpp_2_0_1")]
181            OcppVersion::V2_0_1 => "ocpp2.0.1",
182            #[cfg(feature = "ocpp_2_1")]
183            OcppVersion::V2_1 => "ocpp2.1",
184        }
185    }
186
187    /// Every version compiled into this build, newest first - `connect`'s default set of
188    /// versions to offer when the caller doesn't restrict it via the `versions` argument.
189    #[allow(clippy::vec_init_then_push)]
190    fn all_compiled_in() -> Vec<OcppVersion> {
191        let mut versions = Vec::new();
192        #[cfg(feature = "ocpp_2_1")]
193        versions.push(OcppVersion::V2_1);
194        #[cfg(feature = "ocpp_2_0_1")]
195        versions.push(OcppVersion::V2_0_1);
196        #[cfg(feature = "ocpp_1_6")]
197        versions.push(OcppVersion::V1_6);
198        versions
199    }
200}
201
202/// Connect to an OCPP server over WebSocket, offering the given `versions` (or, if `None`,
203/// every version compiled into this crate via its `ocpp_1_6`/`ocpp_2_0_1`/`ocpp_2_1` features)
204/// in the `Sec-WebSocket-Protocol` header, and using whichever one the server picks - rather
205/// than requiring the caller to already know the server's supported version like
206/// `connect_1_6`/`connect_2_0_1`/`connect_2_1` do. `versions` also controls preference order
207/// (offered in the slice's order); the choice among the offered set is entirely the server's
208/// per RFC 6455.
209pub async fn connect(
210    address: &str,
211    versions: Option<&[OcppVersion]>,
212    options: Option<ConnectOptions<'_>>,
213) -> Result<NegotiatedClient, Box<dyn std::error::Error + Send + Sync>> {
214    let all_compiled_in;
215    let versions = match versions {
216        Some(versions) => versions,
217        None => {
218            all_compiled_in = OcppVersion::all_compiled_in();
219            &all_compiled_in
220        }
221    };
222    let offered = versions
223        .iter()
224        .map(|v| v.protocol())
225        .collect::<Vec<_>>()
226        .join(", ");
227    let (stream, negotiated) = setup_socket(address, &offered, options.clone()).await?;
228    let protocol = versions
229        .iter()
230        .find(|v| v.protocol() == negotiated)
231        .map(|v| v.protocol())
232        .ok_or_else(|| format!("Server negotiated unsupported protocol: {negotiated}"))?;
233    let config = prepare(address, protocol, options);
234    let (sink, source) = crate::transport::websocket::split(stream);
235
236    Ok(match protocol {
237        #[cfg(feature = "ocpp_1_6")]
238        "ocpp1.6" => NegotiatedClient::V1_6(crate::Client::from_transport_with_config(
239            sink,
240            source,
241            Box::new(crate::runtime::tokio::TokioExecutor),
242            Box::new(crate::runtime::tokio::TokioTimer),
243            config,
244        )),
245        #[cfg(feature = "ocpp_2_0_1")]
246        "ocpp2.0.1" => NegotiatedClient::V2_0_1(crate::Client::from_transport_with_config(
247            sink,
248            source,
249            Box::new(crate::runtime::tokio::TokioExecutor),
250            Box::new(crate::runtime::tokio::TokioTimer),
251            config,
252        )),
253        #[cfg(feature = "ocpp_2_1")]
254        "ocpp2.1" => NegotiatedClient::V2_1(crate::Client::from_transport_with_config(
255            sink,
256            source,
257            Box::new(crate::runtime::tokio::TokioExecutor),
258            Box::new(crate::runtime::tokio::TokioTimer),
259            config,
260        )),
261        _ => unreachable!("protocol only ever holds a value returned by OcppVersion::protocol"),
262    })
263}
264
265/// Connect to an OCPP 1.6 server over WebSocket.
266#[cfg(feature = "ocpp_1_6")]
267pub async fn connect_1_6(
268    address: &str,
269    options: Option<ConnectOptions<'_>>,
270) -> Result<crate::ocpp_1_6::OCPP1_6Client, Box<dyn std::error::Error + Send + Sync>> {
271    let config = prepare(address, "ocpp1.6", options.clone());
272    let (stream, _protocol) = setup_socket(address, "ocpp1.6", options).await?;
273    let (sink, source) = crate::transport::websocket::split(stream);
274    Ok(crate::Client::from_transport_with_config(
275        sink,
276        source,
277        Box::new(crate::runtime::tokio::TokioExecutor),
278        Box::new(crate::runtime::tokio::TokioTimer),
279        config,
280    ))
281}
282
283/// Connect to an OCPP 2.0.1 server over WebSocket.
284#[cfg(feature = "ocpp_2_0_1")]
285pub async fn connect_2_0_1(
286    address: &str,
287    options: Option<ConnectOptions<'_>>,
288) -> Result<crate::ocpp_2_0_1::OCPP2_0_1Client, Box<dyn std::error::Error + Send + Sync>> {
289    let config = prepare(address, "ocpp2.0.1", options.clone());
290    let (stream, _protocol) = setup_socket(address, "ocpp2.0.1", options).await?;
291    let (sink, source) = crate::transport::websocket::split(stream);
292    Ok(crate::Client::from_transport_with_config(
293        sink,
294        source,
295        Box::new(crate::runtime::tokio::TokioExecutor),
296        Box::new(crate::runtime::tokio::TokioTimer),
297        config,
298    ))
299}
300
301/// Connect to an OCPP 2.1 server over WebSocket.
302#[cfg(feature = "ocpp_2_1")]
303pub async fn connect_2_1(
304    address: &str,
305    options: Option<ConnectOptions<'_>>,
306) -> Result<crate::ocpp_2_1::OCPP2_1Client, Box<dyn std::error::Error + Send + Sync>> {
307    let config = prepare(address, "ocpp2.1", options.clone());
308    let (stream, _protocol) = setup_socket(address, "ocpp2.1", options).await?;
309    let (sink, source) = crate::transport::websocket::split(stream);
310    Ok(crate::Client::from_transport_with_config(
311        sink,
312        source,
313        Box::new(crate::runtime::tokio::TokioExecutor),
314        Box::new(crate::runtime::tokio::TokioTimer),
315        config,
316    ))
317}
318
319/// Turns `options` into the [`ClientConfig`] the `Client` constructor takes and, if reconnect is
320/// enabled, builds the `Reconnector` that redials this same address/protocol/credentials. Shared
321/// by all three `connect_*` entry points and by `connect`.
322///
323/// Note the two different defaults for an absent `options`: `ConnectOptions::default()` is used
324/// for `reconnect`/`keepalive` (both enabled), so passing `None` behaves the same as passing
325/// `Some(Default::default())` rather than silently disabling them.
326fn prepare(
327    address: &str,
328    protocol: &'static str,
329    options: Option<ConnectOptions<'_>>,
330) -> ClientConfig {
331    let defaults = ConnectOptions::default();
332    let timeout = options
333        .as_ref()
334        .and_then(|o| o.timeout)
335        .unwrap_or(DEFAULT_TIMEOUT);
336    let keepalive = options
337        .as_ref()
338        .map(|o| o.keepalive)
339        .unwrap_or(defaults.keepalive);
340    let reconnect = options.as_ref().map(|o| o.reconnect).unwrap_or_default();
341    let username = options
342        .as_ref()
343        .and_then(|o| o.username)
344        .map(str::to_string);
345    let password = options
346        .as_ref()
347        .and_then(|o| o.password)
348        .map(<[u8]>::to_vec);
349    let tls_config = options.as_ref().and_then(|o| o.tls_config.clone());
350
351    let custom = options.as_ref().and_then(|o| o.reconnector.clone());
352
353    let config = ClientConfig::new(timeout).with_keepalive(keepalive);
354
355    match reconnect {
356        ReconnectBehavior::Disabled => config,
357        ReconnectBehavior::Enabled(policy) if custom.is_some() => {
358            let custom = custom.expect("guarded by the match arm");
359            config.with_reconnect(Box::new(SharedReconnector(custom)), policy)
360        }
361        ReconnectBehavior::Enabled(policy) => config.with_reconnect(
362            Box::new(WebSocketReconnector {
363                address: address.to_string(),
364                protocol,
365                username,
366                password,
367                tls_config,
368            }),
369            policy,
370        ),
371    }
372}
373
374/// Adapts the `Arc<dyn Reconnector>` callers hand us in [`ConnectOptions::reconnector`] to the
375/// `Box<dyn Reconnector>` `Client` takes. `Arc` is what keeps `ConnectOptions` cloneable, which
376/// the `connect_*` functions rely on.
377struct SharedReconnector(Arc<dyn Reconnector>);
378
379impl Reconnector for SharedReconnector {
380    fn connect<'a>(
381        &'a self,
382    ) -> Pin<
383        Box<
384            dyn Future<
385                    Output = Result<
386                        (Box<dyn TransportSink>, Box<dyn TransportStream>),
387                        TransportError,
388                    >,
389                > + Send
390                + 'a,
391        >,
392    > {
393        self.0.connect()
394    }
395}
396
397/// Redials `address` with the original protocol/credentials/TLS config whenever `Client`'s
398/// background read loop needs a fresh transport after a disconnect.
399struct WebSocketReconnector {
400    address: String,
401    protocol: &'static str,
402    username: Option<String>,
403    password: Option<Vec<u8>>,
404    tls_config: Option<Arc<rustls::ClientConfig>>,
405}
406
407impl Reconnector for WebSocketReconnector {
408    fn connect<'a>(
409        &'a self,
410    ) -> Pin<
411        Box<
412            dyn Future<
413                    Output = Result<
414                        (Box<dyn TransportSink>, Box<dyn TransportStream>),
415                        TransportError,
416                    >,
417                > + Send
418                + 'a,
419        >,
420    > {
421        Box::pin(async move {
422            let options = ConnectOptions {
423                username: self.username.as_deref(),
424                password: self.password.as_deref(),
425                timeout: None,
426                reconnect: ReconnectBehavior::Disabled,
427                tls_config: self.tls_config.clone(),
428                reconnector: None,
429                // Both this and `reconnect` are inert here: `setup_socket` only reads the
430                // credential/TLS fields, and this path produces bare transport halves for a
431                // client whose keepalive task is already running.
432                keepalive: KeepaliveBehavior::Disabled,
433            };
434            let (stream, _protocol) =
435                setup_socket(&self.address, self.protocol, Some(options)).await?;
436            Ok(crate::transport::websocket::split(stream))
437        })
438    }
439}
440
441async fn setup_socket(
442    address: &str,
443    protocols: &str,
444    options: Option<ConnectOptions<'_>>,
445) -> Result<
446    (WebSocketStream<MaybeTlsStream<TcpStream>>, String),
447    Box<dyn std::error::Error + Send + Sync>,
448> {
449    let address = Url::parse(address)?;
450
451    let socket_addrs = address.socket_addrs(|| None)?;
452    let stream = TcpStream::connect(&*socket_addrs).await?;
453
454    let mut request: Request<()> = address.to_string().into_client_request()?;
455    request
456        .headers_mut()
457        .insert(SEC_WEBSOCKET_PROTOCOL, protocols.parse()?);
458    let mut tls_config = None;
459    if let Some(options) = options {
460        if let Some(username) = options.username {
461            let mut data = format!("{username}:").into_bytes();
462            data.extend_from_slice(options.password.unwrap_or_default());
463            let encoded = BASE64_STANDARD.encode(data);
464            request
465                .headers_mut()
466                .insert(AUTHORIZATION, format!("Basic {encoded}").parse()?);
467        }
468        tls_config = options.tls_config;
469    }
470
471    let connector = tls_config.map(Connector::Rustls);
472    let (stream, response) = client_async_tls_with_config(request, stream, None, connector).await?;
473
474    let protocol = response
475        .headers()
476        .get(SEC_WEBSOCKET_PROTOCOL)
477        .ok_or("No OCPP protocol negotiated")?;
478
479    Ok((stream, protocol.to_str()?.to_string()))
480}
ocpp_client/connect.rs

ocpp_client/
connect.rs
