ocpp_service/
server.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Accept loop for the OCPP listeners (plain and TLS).
5//!
6//! Each accepted connection runs its TLS handshake (if any) in its own task,
7//! then hyper serves HTTP/1.1 with upgrades. Connection metadata (peer
8//! address, TLS outcome) is attached to every request as a [`ConnMeta`]
9//! extension for the WebSocket handshake handler.
10
11use std::net::SocketAddr;
12use std::sync::Arc;
13use std::time::Duration;
14
15use axum::Router;
16use axum::body::Body;
17use hyper::body::Incoming;
18use hyper_util::rt::{TokioIo, TokioTimer};
19use tokio::net::TcpListener;
20use tokio::sync::{Semaphore, watch};
21use tower::ServiceExt;
22
23use crate::tls::{TlsAcceptor, TlsInfo};
24
25const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(15);
26const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(15);
27
28#[derive(Debug, Clone)]
29pub struct ConnMeta {
30    pub peer: SocketAddr,
31    /// `None` for the plain (unencrypted) listener.
32    pub tls: Option<TlsInfo>,
33}
34
35pub async fn serve_ocpp(
36    listener: TcpListener,
37    tls: Option<Arc<TlsAcceptor>>,
38    router: Router,
39    limit: Arc<Semaphore>,
40    mut shutdown: watch::Receiver<bool>,
41) {
42    loop {
43        let (tcp, peer) = tokio::select! {
44            _ = shutdown.changed() => return,
45            accepted = listener.accept() => match accepted {
46                Ok(a) => a,
47                Err(e) => {
48                    // EMFILE and similar: back off instead of spinning.
49                    tracing::warn!(error = %e, "accept failed");
50                    tokio::time::sleep(Duration::from_millis(100)).await;
51                    continue;
52                }
53            },
54        };
55        let Ok(permit) = limit.clone().try_acquire_owned() else {
56            tracing::warn!(%peer, "connection limit reached, refusing connection");
57            continue;
58        };
59        let _ = tcp.set_nodelay(true);
60        let router = router.clone();
61        let tls = tls.clone();
62        tokio::spawn(async move {
63            let _permit = permit;
64            match tls {
65                None => serve_conn(tcp, ConnMeta { peer, tls: None }, router).await,
66                Some(acceptor) => {
67                    match tokio::time::timeout(TLS_HANDSHAKE_TIMEOUT, acceptor.accept(tcp)).await {
68                        Ok(Ok((stream, info))) => {
69                            serve_conn(
70                                stream,
71                                ConnMeta {
72                                    peer,
73                                    tls: Some(info),
74                                },
75                                router,
76                            )
77                            .await
78                        }
79                        Ok(Err(e)) => tracing::info!(%peer, error = %e, "TLS handshake failed"),
80                        Err(_) => tracing::info!(%peer, "TLS handshake timed out"),
81                    }
82                }
83            }
84        });
85    }
86}
87
88async fn serve_conn<S>(stream: S, meta: ConnMeta, router: Router)
89where
90    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
91{
92    let service = hyper::service::service_fn(move |mut req: hyper::Request<Incoming>| {
93        req.extensions_mut().insert(meta.clone());
94        router.clone().oneshot(req.map(Body::new))
95    });
96    let conn = hyper::server::conn::http1::Builder::new()
97        .timer(TokioTimer::new())
98        .header_read_timeout(HEADER_READ_TIMEOUT)
99        .serve_connection(TokioIo::new(stream), service)
100        .with_upgrades();
101    if let Err(e) = conn.await {
102        tracing::debug!(error = %e, "HTTP connection error");
103    }
104}
ocpp_service/server.rs

ocpp_service/
server.rs
