codetics_ocpp_ws/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! A small, dependency-light WebSocket implementation for OCPP-J.
5//!
6//! - RFC 6455 framing, fragmentation, control frames and close handshake.
7//! - RFC 7692 permessage-deflate, which OCPP 2.0.1/2.1 make mandatory for a
8//!   CSMS (Part 4 §3.3), with a memory profile suited to very many idle
9//!   connections (see [`deflate`]).
10//! - The HTTP upgrade itself is left to the embedding HTTP server (axum/hyper
11//!   on the CSMS side); [`client`] contains a minimal client for tests and
12//!   simulators.
13//!
14//! The connection is split into a cancel-safe [`WsReader`] and a [`WsWriter`]
15//! so that one task per connection can multiplex reads, timers and commands.
16
17pub mod base64;
18pub mod client;
19pub mod conn;
20pub mod deflate;
21pub mod frame;
22pub mod handshake;
23pub mod sha1;
24
25pub use conn::{Message, Role, WsConfig, WsError, WsReader, WsWriter, close_code};
26pub use deflate::{DeflateConfig, DeflateParams};
27
28use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
29
30/// Wraps an upgraded server-side stream into reader/writer halves.
31pub fn server<S>(
32    io: S,
33    cfg: WsConfig,
34    deflate: Option<(DeflateParams, DeflateConfig)>,
35) -> (WsReader<ReadHalf<S>>, WsWriter<WriteHalf<S>>)
36where
37    S: AsyncRead + AsyncWrite,
38{
39    let (r, w) = tokio::io::split(io);
40    (
41        WsReader::new(r, Role::Server, cfg, deflate.map(|(p, _)| p), Vec::new()),
42        WsWriter::new(w, Role::Server, deflate),
43    )
44}
45
46#[cfg(test)]
47mod tests {
48    use super::*;
49    use tokio::io::duplex;
50
51    type End = (
52        WsReader<ReadHalf<tokio::io::DuplexStream>>,
53        WsWriter<WriteHalf<tokio::io::DuplexStream>>,
54    );
55
56    fn pair(deflate: bool) -> (End, End) {
57        let (a, b) = duplex(1 << 16);
58        let params = DeflateParams {
59            client_no_context_takeover: false,
60            server_no_context_takeover: true,
61        };
62        let cfg = DeflateConfig {
63            min_compress_size: 0,
64            ..Default::default()
65        };
66        let d = deflate.then_some((params, cfg));
67        let server = server(a, WsConfig::default(), d);
68        let (r, w) = tokio::io::split(b);
69        let client = (
70            WsReader::new(
71                r,
72                Role::Client,
73                WsConfig::default(),
74                d.map(|x| x.0),
75                Vec::new(),
76            ),
77            WsWriter::new(w, Role::Client, d),
78        );
79        (server, client)
80    }
81
82    #[tokio::test]
83    async fn text_both_directions_with_and_without_deflate() {
84        for deflate in [false, true] {
85            let ((mut sr, mut sw), (mut cr, mut cw)) = pair(deflate);
86            for i in 0..50 {
87                let msg = format!(r#"[2,"{i}","Heartbeat",{{}}] {}"#, "é".repeat(i * 20));
88                cw.send_text(&msg).await.unwrap();
89                assert_eq!(sr.recv().await.unwrap(), Some(Message::Text(msg.clone())));
90                sw.send_text(&msg).await.unwrap();
91                assert_eq!(cr.recv().await.unwrap(), Some(Message::Text(msg)));
92            }
93            cw.send_ping(b"hi").await.unwrap();
94            assert_eq!(
95                sr.recv().await.unwrap(),
96                Some(Message::Ping(b"hi".to_vec()))
97            );
98            sw.send_close(close_code::NORMAL, "bye").await.unwrap();
99            assert_eq!(
100                cr.recv().await.unwrap(),
101                Some(Message::Close(Some((1000, "bye".into()))))
102            );
103            assert!(sw.send_text("x").await.is_err());
104        }
105    }
106
107    #[tokio::test]
108    async fn rejects_unmasked_client_frames_and_reassembles_fragments() {
109        use tokio::io::AsyncWriteExt;
110        let (a, mut b) = duplex(1 << 16);
111        let (mut r, _w) = server(a, WsConfig::default(), None);
112        // Fragmented masked text: "Hel" + "lo", with a ping in between.
113        let mut raw = Vec::new();
114        let m = [1, 2, 3, 4];
115        let mut f1 = vec![0x01, 0x83];
116        f1.extend_from_slice(&m);
117        let mut p = b"Hel".to_vec();
118        frame::apply_mask(&mut p, m);
119        f1.extend_from_slice(&p);
120        raw.extend_from_slice(&f1);
121        let mut ping = Vec::new();
122        frame::encode(&mut ping, frame::OpCode::Ping, false, b"", Some(m));
123        raw.extend_from_slice(&ping);
124        let mut f2 = vec![0x80, 0x82];
125        f2.extend_from_slice(&m);
126        let mut p = b"lo".to_vec();
127        frame::apply_mask(&mut p, m);
128        f2.extend_from_slice(&p);
129        raw.extend_from_slice(&f2);
130        // Write byte by byte to exercise partial reads.
131        for byte in raw {
132            b.write_all(&[byte]).await.unwrap();
133        }
134        assert_eq!(r.recv().await.unwrap(), Some(Message::Ping(vec![])));
135        assert_eq!(r.recv().await.unwrap(), Some(Message::Text("Hello".into())));
136        b.write_all(&[0x81, 0x01, b'x']).await.unwrap();
137        assert!(matches!(r.recv().await, Err(WsError::Protocol(_))));
138    }
139
140    #[tokio::test]
141    async fn enforces_max_message_size() {
142        let (a, b) = duplex(1 << 20);
143        let (mut r, _w) = server(
144            a,
145            WsConfig {
146                max_message_size: 1000,
147                ..Default::default()
148            },
149            None,
150        );
151        let (_cr, mut cw) = {
152            let (r, w) = tokio::io::split(b);
153            (
154                WsReader::new(r, Role::Client, WsConfig::default(), None, Vec::new()),
155                WsWriter::new(w, Role::Client, None),
156            )
157        };
158        cw.send_text(&"x".repeat(5000)).await.unwrap();
159        assert!(matches!(r.recv().await, Err(WsError::MessageTooBig)));
160    }
161}
162
163/// Compiles and runs the examples of the README.
164#[cfg(doctest)]
165#[doc = include_str!("../README.md")]
166pub struct ReadmeDoctests;
codetics_ocpp_ws/lib.rs

codetics_ocpp_ws/
lib.rs
