ocpp_service/
hub.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Registry of the station connections held by this node.
5//!
6//! Sharded so that tens of thousands of connections and concurrent API calls
7//! do not contend on one lock. Each entry holds the sender side of the
8//! connection's command queue; the session task owns everything else.
9
10use std::collections::HashMap;
11use std::hash::BuildHasher;
12use std::sync::RwLock;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::time::Duration;
15
16use ocpp_core::{DateTime, OcppVersion, RpcError, Violation};
17use serde::Serialize;
18use tokio::sync::{mpsc, oneshot};
19
20const SHARDS: usize = 64;
21
22#[derive(Debug)]
23pub enum Command {
24    /// Send a CALL and report the outcome.
25    Call {
26        action: String,
27        payload: serde_json::Value,
28        reply: oneshot::Sender<CallOutcome>,
29    },
30    /// Close the connection (e.g. replaced by a newer one).
31    Close { code: u16, reason: &'static str },
32}
33
34#[derive(Debug)]
35pub enum CallFailure {
36    /// The station answered with CALLERROR.
37    Error(RpcError),
38    /// The CALLRESULT payload does not match the message definition (raw payload attached).
39    InvalidResponse {
40        violation: Violation,
41        raw: serde_json::Value,
42    },
43    Timeout,
44    Disconnected,
45}
46
47pub type CallOutcome = Result<serde_json::Value, CallFailure>;
48
49#[derive(Debug, Clone)]
50pub struct ConnHandle {
51    pub conn_id: u64,
52    pub station_id: i64,
53    pub version: OcppVersion,
54    pub connected_at: DateTime,
55    pub remote_addr: Option<String>,
56    pub tx: mpsc::Sender<Command>,
57}
58
59#[derive(Debug, Serialize)]
60pub struct ConnInfo {
61    pub ocpp_identity: String,
62    pub station_id: i64,
63    pub ocpp_version: String,
64    pub connected_at: DateTime,
65    pub remote_addr: Option<String>,
66}
67
68pub enum SendError {
69    NotConnected,
70    QueueFull,
71}
72
73pub struct Hub {
74    shards: Vec<RwLock<HashMap<String, ConnHandle>>>,
75    hasher: std::collections::hash_map::RandomState,
76    next_id: AtomicU64,
77}
78
79impl Default for Hub {
80    fn default() -> Self {
81        Self::new()
82    }
83}
84
85impl Hub {
86    pub fn new() -> Self {
87        Hub {
88            shards: (0..SHARDS).map(|_| RwLock::new(HashMap::new())).collect(),
89            hasher: Default::default(),
90            next_id: AtomicU64::new(1),
91        }
92    }
93
94    fn shard(&self, identity: &str) -> &RwLock<HashMap<String, ConnHandle>> {
95        &self.shards[self.hasher.hash_one(identity) as usize % SHARDS]
96    }
97
98    pub fn next_conn_id(&self) -> u64 {
99        self.next_id.fetch_add(1, Ordering::Relaxed)
100    }
101
102    /// Registers a connection, returning the handle it replaced (if any).
103    pub fn register(&self, identity: &str, handle: ConnHandle) -> Option<ConnHandle> {
104        self.shard(identity)
105            .write()
106            .unwrap_or_else(|e| e.into_inner())
107            .insert(identity.to_owned(), handle)
108    }
109
110    /// Removes the entry only if it still belongs to `conn_id`.
111    pub fn unregister(&self, identity: &str, conn_id: u64) {
112        let mut shard = self
113            .shard(identity)
114            .write()
115            .unwrap_or_else(|e| e.into_inner());
116        if shard.get(identity).is_some_and(|h| h.conn_id == conn_id) {
117            shard.remove(identity);
118        }
119    }
120
121    pub fn get(&self, identity: &str) -> Option<ConnHandle> {
122        self.shard(identity)
123            .read()
124            .unwrap_or_else(|e| e.into_inner())
125            .get(identity)
126            .cloned()
127    }
128
129    pub fn is_empty(&self) -> bool {
130        self.len() == 0
131    }
132
133    pub fn len(&self) -> usize {
134        self.shards
135            .iter()
136            .map(|s| s.read().unwrap_or_else(|e| e.into_inner()).len())
137            .sum()
138    }
139
140    pub fn list(&self) -> Vec<ConnInfo> {
141        let mut out = Vec::new();
142        for s in &self.shards {
143            for (id, h) in s.read().unwrap_or_else(|e| e.into_inner()).iter() {
144                out.push(ConnInfo {
145                    ocpp_identity: id.clone(),
146                    station_id: h.station_id,
147                    ocpp_version: h.version.to_string(),
148                    connected_at: h.connected_at,
149                    remote_addr: h.remote_addr.clone(),
150                });
151            }
152        }
153        out.sort_by(|a, b| a.ocpp_identity.cmp(&b.ocpp_identity));
154        out
155    }
156
157    /// Queues a CALL on the station's connection and waits for its outcome.
158    /// The session sends CALLs one at a time (OCPP-J §4.1.1), so `wait`
159    /// covers queueing plus the call timeout.
160    pub async fn call(
161        &self,
162        identity: &str,
163        action: &str,
164        payload: serde_json::Value,
165        wait: Duration,
166    ) -> Result<CallOutcome, SendError> {
167        let handle = self.get(identity).ok_or(SendError::NotConnected)?;
168        let (reply, rx) = oneshot::channel();
169        handle
170            .tx
171            .try_send(Command::Call {
172                action: action.to_owned(),
173                payload,
174                reply,
175            })
176            .map_err(|e| match e {
177                mpsc::error::TrySendError::Full(_) => SendError::QueueFull,
178                mpsc::error::TrySendError::Closed(_) => SendError::NotConnected,
179            })?;
180        Ok(match tokio::time::timeout(wait, rx).await {
181            Ok(Ok(outcome)) => outcome,
182            Ok(Err(_)) => Err(CallFailure::Disconnected),
183            Err(_) => Err(CallFailure::Timeout),
184        })
185    }
186}
ocpp_service/hub.rs

ocpp_service/
hub.rs
