codetics-ocpp-service 0.1.0

Codetics OCPP: Charging Station Management System (CSMS) service for OCPP 1.6J, 2.0.1 and 2.1, with TLS security profiles, smart charging, reservations and tariffs
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

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

//! OCPP Charging Station Management System microservice.
//!
//! - `/<prefix>/<identity>` (OCPP listeners, plain and/or TLS): OCPP-J 1.6, 2.0.1 and 2.1 over WebSocket.
//! - `/api/v1/...` (API listener): station registry, search, commands.

pub mod api;
pub mod auth;
pub mod calls;
pub mod config;
pub mod db;
pub mod error;
pub mod hub;
pub mod ocpp;
pub mod reservation;
pub mod server;
pub mod smart;
pub mod state;
pub mod tariff;
pub mod tls;
pub mod tz;

use std::sync::Arc;

use axum::Router;
use axum::routing::any;
use tokio::net::TcpListener;
use tokio::sync::{Semaphore, watch};

use crate::config::Config;
use crate::state::{App, AppState};

pub fn ocpp_router(app: App) -> Router {
    // Any path: the station identity is the last segment.
    Router::new()
        .route("/{*path}", any(ocpp::handshake::upgrade))
        .with_state(app)
}

/// Builds the shared state. Run migrations before calling.
pub fn build_state(cfg: Config, db: db::Db, shutdown: watch::Receiver<bool>) -> App {
    Arc::new(AppState {
        auth: auth::Authenticator::new(cfg.auth_concurrency),
        cfg,
        db,
        hub: hub::Hub::new(),
        ids: ocpp_core::MessageIdGenerator::new(),
        zones: tz::Zones::default(),
        shutdown,
    })
}

pub struct Listeners {
    /// Plain OCPP listener (Security Profiles 0/1).
    pub ocpp: Option<TcpListener>,
    /// TLS OCPP listener (Security Profiles 2/3).
    pub ocpp_tls: Option<(TcpListener, Arc<tls::TlsAcceptor>)>,
    pub api: TcpListener,
}

/// Serves all listeners until `shutdown` resolves.
pub async fn serve(
    app: App,
    listeners: Listeners,
    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) -> std::io::Result<()> {
    let (stop_tx, stop_rx) = watch::channel(false);
    let router = ocpp_router(app.clone());
    let mut tasks = Vec::new();
    if let Some(l) = listeners.ocpp {
        let limit = Arc::new(Semaphore::new(app.cfg.max_connections));
        tasks.push(tokio::spawn(server::serve_ocpp(
            l,
            None,
            router.clone(),
            limit,
            stop_rx.clone(),
        )));
    }
    if let Some((l, acceptor)) = listeners.ocpp_tls {
        let limit = Arc::new(Semaphore::new(app.cfg.max_connections));
        tasks.push(tokio::spawn(server::serve_ocpp(
            l,
            Some(acceptor),
            router.clone(),
            limit,
            stop_rx.clone(),
        )));
    }
    reservation::spawn_sweeper(app.clone(), stop_rx.clone());
    let mut api_stop = stop_rx.clone();
    let api_server =
        axum::serve(listeners.api, api::router(app.clone())).with_graceful_shutdown(async move {
            let _ = api_stop.changed().await;
        });
    tokio::spawn(async move {
        shutdown.await;
        let _ = stop_tx.send(true);
    });
    let result = api_server.await;
    for t in tasks {
        let _ = t.await;
    }
    result
}