ocpp_service/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! OCPP Charging Station Management System microservice.
5//!
6//! - `/<prefix>/<identity>` (OCPP listeners, plain and/or TLS): OCPP-J 1.6, 2.0.1 and 2.1 over WebSocket.
7//! - `/api/v1/...` (API listener): station registry, search, commands.
8
9pub mod api;
10pub mod auth;
11pub mod calls;
12pub mod config;
13pub mod db;
14pub mod error;
15pub mod hub;
16pub mod ocpp;
17pub mod reservation;
18pub mod server;
19pub mod smart;
20pub mod state;
21pub mod tariff;
22pub mod tls;
23pub mod tz;
24
25use std::sync::Arc;
26
27use axum::Router;
28use axum::routing::any;
29use tokio::net::TcpListener;
30use tokio::sync::{Semaphore, watch};
31
32use crate::config::Config;
33use crate::state::{App, AppState};
34
35pub fn ocpp_router(app: App) -> Router {
36    // Any path: the station identity is the last segment.
37    Router::new()
38        .route("/{*path}", any(ocpp::handshake::upgrade))
39        .with_state(app)
40}
41
42/// Builds the shared state. Run migrations before calling.
43pub fn build_state(cfg: Config, db: db::Db, shutdown: watch::Receiver<bool>) -> App {
44    Arc::new(AppState {
45        auth: auth::Authenticator::new(cfg.auth_concurrency),
46        cfg,
47        db,
48        hub: hub::Hub::new(),
49        ids: ocpp_core::MessageIdGenerator::new(),
50        zones: tz::Zones::default(),
51        shutdown,
52    })
53}
54
55pub struct Listeners {
56    /// Plain OCPP listener (Security Profiles 0/1).
57    pub ocpp: Option<TcpListener>,
58    /// TLS OCPP listener (Security Profiles 2/3).
59    pub ocpp_tls: Option<(TcpListener, Arc<tls::TlsAcceptor>)>,
60    pub api: TcpListener,
61}
62
63/// Serves all listeners until `shutdown` resolves.
64pub async fn serve(
65    app: App,
66    listeners: Listeners,
67    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
68) -> std::io::Result<()> {
69    let (stop_tx, stop_rx) = watch::channel(false);
70    let router = ocpp_router(app.clone());
71    let mut tasks = Vec::new();
72    if let Some(l) = listeners.ocpp {
73        let limit = Arc::new(Semaphore::new(app.cfg.max_connections));
74        tasks.push(tokio::spawn(server::serve_ocpp(
75            l,
76            None,
77            router.clone(),
78            limit,
79            stop_rx.clone(),
80        )));
81    }
82    if let Some((l, acceptor)) = listeners.ocpp_tls {
83        let limit = Arc::new(Semaphore::new(app.cfg.max_connections));
84        tasks.push(tokio::spawn(server::serve_ocpp(
85            l,
86            Some(acceptor),
87            router.clone(),
88            limit,
89            stop_rx.clone(),
90        )));
91    }
92    reservation::spawn_sweeper(app.clone(), stop_rx.clone());
93    let mut api_stop = stop_rx.clone();
94    let api_server =
95        axum::serve(listeners.api, api::router(app.clone())).with_graceful_shutdown(async move {
96            let _ = api_stop.changed().await;
97        });
98    tokio::spawn(async move {
99        shutdown.await;
100        let _ = stop_tx.send(true);
101    });
102    let result = api_server.await;
103    for t in tasks {
104        let _ = t.await;
105    }
106    result
107}
ocpp_service/lib.rs

ocpp_service/
lib.rs
