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

ocpp_client/
connect.rs
