codetics-ocpp-ws 0.1.0

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)

//! A small, dependency-light WebSocket implementation for OCPP-J.
//!
//! - RFC 6455 framing, fragmentation, control frames and close handshake.
//! - RFC 7692 permessage-deflate, which OCPP 2.0.1/2.1 make mandatory for a
//!   CSMS (Part 4 §3.3), with a memory profile suited to very many idle
//!   connections (see [`deflate`]).
//! - The HTTP upgrade itself is left to the embedding HTTP server (axum/hyper
//!   on the CSMS side); [`client`] contains a minimal client for tests and
//!   simulators.
//!
//! The connection is split into a cancel-safe [`WsReader`] and a [`WsWriter`]
//! so that one task per connection can multiplex reads, timers and commands.

pub mod base64;
pub mod client;
pub mod conn;
pub mod deflate;
pub mod frame;
pub mod handshake;
pub mod sha1;

pub use conn::{Message, Role, WsConfig, WsError, WsReader, WsWriter, close_code};
pub use deflate::{DeflateConfig, DeflateParams};

use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};

/// Wraps an upgraded server-side stream into reader/writer halves.
pub fn server<S>(
    io: S,
    cfg: WsConfig,
    deflate: Option<(DeflateParams, DeflateConfig)>,
) -> (WsReader<ReadHalf<S>>, WsWriter<WriteHalf<S>>)
where
    S: AsyncRead + AsyncWrite,
{
    let (r, w) = tokio::io::split(io);
    (
        WsReader::new(r, Role::Server, cfg, deflate.map(|(p, _)| p), Vec::new()),
        WsWriter::new(w, Role::Server, deflate),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::duplex;

    type End = (
        WsReader<ReadHalf<tokio::io::DuplexStream>>,
        WsWriter<WriteHalf<tokio::io::DuplexStream>>,
    );

    fn pair(deflate: bool) -> (End, End) {
        let (a, b) = duplex(1 << 16);
        let params = DeflateParams {
            client_no_context_takeover: false,
            server_no_context_takeover: true,
        };
        let cfg = DeflateConfig {
            min_compress_size: 0,
            ..Default::default()
        };
        let d = deflate.then_some((params, cfg));
        let server = server(a, WsConfig::default(), d);
        let (r, w) = tokio::io::split(b);
        let client = (
            WsReader::new(
                r,
                Role::Client,
                WsConfig::default(),
                d.map(|x| x.0),
                Vec::new(),
            ),
            WsWriter::new(w, Role::Client, d),
        );
        (server, client)
    }

    #[tokio::test]
    async fn text_both_directions_with_and_without_deflate() {
        for deflate in [false, true] {
            let ((mut sr, mut sw), (mut cr, mut cw)) = pair(deflate);
            for i in 0..50 {
                let msg = format!(r#"[2,"{i}","Heartbeat",{{}}] {}"#, "é".repeat(i * 20));
                cw.send_text(&msg).await.unwrap();
                assert_eq!(sr.recv().await.unwrap(), Some(Message::Text(msg.clone())));
                sw.send_text(&msg).await.unwrap();
                assert_eq!(cr.recv().await.unwrap(), Some(Message::Text(msg)));
            }
            cw.send_ping(b"hi").await.unwrap();
            assert_eq!(
                sr.recv().await.unwrap(),
                Some(Message::Ping(b"hi".to_vec()))
            );
            sw.send_close(close_code::NORMAL, "bye").await.unwrap();
            assert_eq!(
                cr.recv().await.unwrap(),
                Some(Message::Close(Some((1000, "bye".into()))))
            );
            assert!(sw.send_text("x").await.is_err());
        }
    }

    #[tokio::test]
    async fn rejects_unmasked_client_frames_and_reassembles_fragments() {
        use tokio::io::AsyncWriteExt;
        let (a, mut b) = duplex(1 << 16);
        let (mut r, _w) = server(a, WsConfig::default(), None);
        // Fragmented masked text: "Hel" + "lo", with a ping in between.
        let mut raw = Vec::new();
        let m = [1, 2, 3, 4];
        let mut f1 = vec![0x01, 0x83];
        f1.extend_from_slice(&m);
        let mut p = b"Hel".to_vec();
        frame::apply_mask(&mut p, m);
        f1.extend_from_slice(&p);
        raw.extend_from_slice(&f1);
        let mut ping = Vec::new();
        frame::encode(&mut ping, frame::OpCode::Ping, false, b"", Some(m));
        raw.extend_from_slice(&ping);
        let mut f2 = vec![0x80, 0x82];
        f2.extend_from_slice(&m);
        let mut p = b"lo".to_vec();
        frame::apply_mask(&mut p, m);
        f2.extend_from_slice(&p);
        raw.extend_from_slice(&f2);
        // Write byte by byte to exercise partial reads.
        for byte in raw {
            b.write_all(&[byte]).await.unwrap();
        }
        assert_eq!(r.recv().await.unwrap(), Some(Message::Ping(vec![])));
        assert_eq!(r.recv().await.unwrap(), Some(Message::Text("Hello".into())));
        b.write_all(&[0x81, 0x01, b'x']).await.unwrap();
        assert!(matches!(r.recv().await, Err(WsError::Protocol(_))));
    }

    #[tokio::test]
    async fn enforces_max_message_size() {
        let (a, b) = duplex(1 << 20);
        let (mut r, _w) = server(
            a,
            WsConfig {
                max_message_size: 1000,
                ..Default::default()
            },
            None,
        );
        let (_cr, mut cw) = {
            let (r, w) = tokio::io::split(b);
            (
                WsReader::new(r, Role::Client, WsConfig::default(), None, Vec::new()),
                WsWriter::new(w, Role::Client, None),
            )
        };
        cw.send_text(&"x".repeat(5000)).await.unwrap();
        assert!(matches!(r.recv().await, Err(WsError::MessageTooBig)));
    }
}

/// Compiles and runs the examples of the README.
#[cfg(doctest)]
#[doc = include_str!("../README.md")]
pub struct ReadmeDoctests;