ocpp_service/ocpp/
session.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! One task per station connection.
5//!
6//! The task owns the socket and multiplexes, with a single `select!`:
7//! inbound frames, the outbound command queue, the timeout of the CALL in
8//! flight, keep-alive pings and service shutdown. This keeps the per-connection
9//! footprint to one task, one small read buffer and one bounded channel.
10//!
11//! RPC rules applied here (OCPP-J §4.1.1): at most one CSMS-initiated CALL is
12//! in flight per connection; further commands wait in the queue. Inbound CALLs
13//! are processed in arrival order (the station also sends one at a time).
14
15use std::time::Duration;
16
17use ocpp_core::rpc::{self, Frame, FrameError};
18use ocpp_core::{ErrorCode, OcppVersion, RpcError};
19use ocpp_ws::{DeflateConfig, DeflateParams, Message, WsConfig, WsReader, WsWriter, close_code};
20use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
21use tokio::sync::{mpsc, oneshot};
22use tokio::time::{Instant, MissedTickBehavior};
23
24use super::{Registration, StationCtx, v16, v21, v201};
25use crate::db::{self, stations::Station};
26use crate::hub::{CallFailure, CallOutcome, Command, ConnHandle};
27use crate::state::App;
28
29pub struct SessionParams {
30    pub app: App,
31    pub station: Station,
32    /// `None` when no subprotocol could be agreed: the handshake is completed
33    /// without `Sec-WebSocket-Protocol` and the connection closed (Part 4 §3.2).
34    pub version: Option<OcppVersion>,
35    pub deflate: Option<(DeflateParams, DeflateConfig)>,
36    pub remote_addr: Option<String>,
37}
38
39struct Pending {
40    id: String,
41    action: String,
42    reply: oneshot::Sender<CallOutcome>,
43    deadline: Instant,
44}
45
46struct Session<S> {
47    ctx: StationCtx,
48    writer: WsWriter<WriteHalf<S>>,
49    pending: Option<Pending>,
50    last_rx: Instant,
51}
52
53pub async fn run<S>(p: SessionParams, io: S)
54where
55    S: AsyncRead + AsyncWrite + Send + 'static,
56{
57    let app = p.app;
58    let ws_cfg = WsConfig {
59        max_message_size: app.cfg.max_message_size,
60        ..Default::default()
61    };
62    let (reader, mut writer) = ocpp_ws::server(io, ws_cfg, p.deflate);
63    let Some(version) = p.version else {
64        tracing::info!(station = %p.station.ocpp_identity, "no common OCPP version, closing");
65        let _ = writer
66            .send_close(
67                close_code::PROTOCOL_ERROR,
68                "no supported OCPP version offered",
69            )
70            .await;
71        return;
72    };
73
74    let identity = p.station.ocpp_identity.clone();
75    let station_id = p.station.id;
76    let conn_id = app.hub.next_conn_id();
77    let connection_id = format!("{}:{}", app.cfg.node_id, app.ids.next_id());
78    let (cmd_tx, cmd_rx) = mpsc::channel(app.cfg.command_queue_size.max(1));
79    let handle = ConnHandle {
80        conn_id,
81        station_id,
82        version,
83        connected_at: ocpp_core::DateTime::now(),
84        remote_addr: p.remote_addr.clone(),
85        tx: cmd_tx.clone(),
86    };
87    if let Some(old) = app.hub.register(&identity, handle) {
88        tracing::info!(station = %identity, "new connection replaces the previous one");
89        let _ = old.tx.try_send(Command::Close {
90            code: close_code::POLICY_VIOLATION,
91            reason: "replaced by a newer connection",
92        });
93    }
94    if let Err(e) = db::stations::mark_connected(
95        &app.db,
96        station_id,
97        &app.cfg.node_id,
98        &connection_id,
99        version.as_str(),
100    )
101    .await
102    {
103        tracing::error!(station = %identity, error = %e, "failed to record connection");
104    }
105    tracing::info!(station = %identity, version = %version, remote = ?p.remote_addr, deflate = p.deflate.is_some(), "station connected");
106
107    let registration = Registration::from_db(&p.station.registration_status);
108    if registration == Registration::Unknown {
109        // Never booted (or booted on another CSMS): ask for a BootNotification.
110        // The payload is identical in 1.6, 2.0.1 and 2.1.
111        let (reply, _) = oneshot::channel();
112        let _ = cmd_tx.try_send(Command::Call {
113            action: "TriggerMessage".into(),
114            payload: serde_json::json!({ "requestedMessage": "BootNotification" }),
115            reply,
116        });
117    }
118    drop(cmd_tx);
119
120    let ctx = StationCtx {
121        app: app.clone(),
122        station_id,
123        identity: identity.clone(),
124        station_evse_id: p.station.evse_id.clone(),
125        version,
126        registration,
127    };
128    let mut session = Session {
129        ctx,
130        writer,
131        pending: None,
132        last_rx: Instant::now(),
133    };
134    let reason = session.run_loop(reader, cmd_rx).await;
135    tracing::info!(station = %identity, reason, "station disconnected");
136
137    if let Some(p) = session.pending.take() {
138        let _ = p.reply.send(Err(CallFailure::Disconnected));
139    }
140    app.hub.unregister(&identity, conn_id);
141    if let Err(e) = db::stations::mark_disconnected(&app.db, station_id, &connection_id).await {
142        tracing::error!(station = %identity, error = %e, "failed to record disconnection");
143    }
144}
145
146impl<S: AsyncRead + AsyncWrite + Send + 'static> Session<S> {
147    async fn run_loop(
148        &mut self,
149        mut reader: WsReader<ReadHalf<S>>,
150        mut cmd_rx: mpsc::Receiver<Command>,
151    ) -> &'static str {
152        let app = self.ctx.app.clone();
153        let mut shutdown = app.shutdown.clone();
154        let ping_every = app.cfg.ping_interval;
155        let mut ping = tokio::time::interval(if ping_every.is_zero() {
156            Duration::from_secs(3600)
157        } else {
158            ping_every
159        });
160        ping.set_missed_tick_behavior(MissedTickBehavior::Delay);
161        ping.tick().await;
162        let far = Instant::now() + Duration::from_secs(86_400 * 365);
163
164        loop {
165            let deadline = self.pending.as_ref().map(|p| p.deadline).unwrap_or(far);
166            tokio::select! {
167                _ = shutdown.changed() => {
168                    let _ = self.writer.send_close(close_code::GOING_AWAY, "server shutting down").await;
169                    return "server shutdown";
170                }
171                msg = reader.recv() => {
172                    self.last_rx = Instant::now();
173                    match msg {
174                        Ok(Some(Message::Text(text))) => {
175                            if let Err(e) = self.on_text(&text).await {
176                                tracing::debug!(station = %self.ctx.identity, error = %e, "write failed");
177                                return "write error";
178                            }
179                        }
180                        Ok(Some(Message::Ping(data))) => {
181                            if self.writer.send_pong(&data).await.is_err() {
182                                return "write error";
183                            }
184                        }
185                        Ok(Some(Message::Pong(_))) => {}
186                        Ok(Some(Message::Binary(_))) => {
187                            let _ = self.writer.send_close(close_code::UNSUPPORTED_DATA, "OCPP-J uses text frames").await;
188                            return "binary frame";
189                        }
190                        Ok(Some(Message::Close(frame))) => {
191                            let code = frame.as_ref().map(|f| f.0).unwrap_or(close_code::NORMAL);
192                            let _ = self.writer.send_close(code, "").await;
193                            return "closed by station";
194                        }
195                        Ok(None) => return "connection lost",
196                        Err(e) => {
197                            tracing::info!(station = %self.ctx.identity, error = %e, "WebSocket error");
198                            let _ = self.writer.send_close(e.close_code(), "").await;
199                            return "protocol error";
200                        }
201                    }
202                }
203                cmd = cmd_rx.recv(), if self.pending.is_none() => match cmd {
204                    Some(Command::Call { action, payload, reply }) => {
205                        if reply.is_closed() && action != "TriggerMessage" {
206                            continue; // The API caller gave up while queued.
207                        }
208                        let id = app.ids.next_id();
209                        let text = match rpc::encode_call(&id, &action, &payload) {
210                            Ok(t) => t,
211                            Err(e) => {
212                                let _ = reply.send(Err(CallFailure::Error(RpcError::internal(e.to_string()))));
213                                continue;
214                            }
215                        };
216                        if self.writer.send_text(&text).await.is_err() {
217                            let _ = reply.send(Err(CallFailure::Disconnected));
218                            return "write error";
219                        }
220                        tracing::debug!(station = %self.ctx.identity, %action, %id, "CALL sent");
221                        self.pending = Some(Pending { id, action, reply, deadline: Instant::now() + app.cfg.call_timeout });
222                    }
223                    Some(Command::Close { code, reason }) => {
224                        let _ = self.writer.send_close(code, reason).await;
225                        return reason;
226                    }
227                    None => {
228                        let _ = self.writer.send_close(close_code::POLICY_VIOLATION, "connection replaced").await;
229                        return "unregistered";
230                    }
231                },
232                _ = tokio::time::sleep_until(deadline), if self.pending.is_some() => {
233                    let p = self.pending.take().expect("pending checked");
234                    tracing::warn!(station = %self.ctx.identity, action = %p.action, id = %p.id, "CALL timed out");
235                    let _ = p.reply.send(Err(CallFailure::Timeout));
236                }
237                _ = ping.tick(), if !ping_every.is_zero() => {
238                    let idle = self.last_rx.elapsed();
239                    if idle >= ping_every * 3 {
240                        let _ = self.writer.send_close(close_code::GOING_AWAY, "keep-alive timeout").await;
241                        return "keep-alive timeout";
242                    }
243                    if idle >= ping_every && self.writer.send_ping(b"").await.is_err() {
244                        return "write error";
245                    }
246                }
247            }
248        }
249    }
250
251    async fn send(&mut self, text: String) -> Result<(), ocpp_ws::WsError> {
252        self.writer.send_text(&text).await
253    }
254
255    async fn send_error(
256        &mut self,
257        id: &str,
258        err: &RpcError,
259        result_error: bool,
260    ) -> Result<(), ocpp_ws::WsError> {
261        let text = rpc::encode_error(id, err, self.ctx.version.dialect(), result_error);
262        self.send(text).await
263    }
264
265    async fn on_text(&mut self, text: &str) -> Result<(), ocpp_ws::WsError> {
266        let version = self.ctx.version;
267        let frame = match Frame::parse(text) {
268            Ok(Frame::Send { .. } | Frame::CallResultError { .. })
269                if version != OcppVersion::V21 =>
270            {
271                // Message types 5 and 6 only exist in OCPP 2.1.
272                return self.unknown_message_type(text, None).await;
273            }
274            Ok(f) => f,
275            Err(FrameError::UnknownMessageType { id, type_id }) => {
276                return self
277                    .unknown_message_type(text, id.map(|i| (i, type_id)))
278                    .await;
279            }
280            Err(e @ FrameError::Malformed { .. }) => {
281                tracing::warn!(station = %self.ctx.identity, error = %e, "malformed frame");
282                let code = if e.reply_id() == rpc::UNKNOWN_MESSAGE_ID {
283                    ErrorCode::RpcFrameworkError
284                } else {
285                    ErrorCode::FormatViolation
286                };
287                let id = e.reply_id().to_owned();
288                return self
289                    .send_error(&id, &RpcError::new(code, e.to_string()), false)
290                    .await;
291            }
292        };
293
294        match frame {
295            Frame::Call {
296                id,
297                action,
298                payload,
299            } => {
300                if !self.ctx.registration.allows(&action) {
301                    let err = RpcError::new(
302                        ErrorCode::SecurityError,
303                        "Charging Station is not accepted by the CSMS; send BootNotification first",
304                    );
305                    return self.send_error(&id, &err, false).await;
306                }
307                let result = match version {
308                    OcppVersion::V16 => v16::handle(&mut self.ctx, &action, payload.get()).await,
309                    OcppVersion::V201 => v201::handle(&mut self.ctx, &action, payload.get()).await,
310                    OcppVersion::V21 => v21::handle(&mut self.ctx, &action, payload.get()).await,
311                };
312                match result {
313                    Ok(response) => match rpc::encode_result(&id, &response) {
314                        Ok(t) => self.send(t).await,
315                        Err(e) => {
316                            self.send_error(&id, &RpcError::internal(e.to_string()), false)
317                                .await
318                        }
319                    },
320                    Err(err) => {
321                        if err.code != ErrorCode::InternalError {
322                            tracing::info!(station = %self.ctx.identity, %action, error = %err, "CALL rejected");
323                        }
324                        self.send_error(&id, &err, false).await
325                    }
326                }
327            }
328            Frame::Send {
329                action, payload, ..
330            } => {
331                v21::handle_send(&mut self.ctx, &action, payload.get()).await;
332                Ok(())
333            }
334            Frame::CallResult { id, payload } => {
335                let Some(p) = self.take_pending(&id) else {
336                    return Ok(());
337                };
338                let validated = match version {
339                    OcppVersion::V16 => v16::validate_response(&p.action, payload.get()),
340                    OcppVersion::V201 => v201::validate_response(&p.action, payload.get()),
341                    OcppVersion::V21 => v21::validate_response(&p.action, payload.get()),
342                };
343                match validated {
344                    Ok(v) => {
345                        let _ = p.reply.send(Ok(v));
346                    }
347                    Err(violation) => {
348                        tracing::warn!(station = %self.ctx.identity, action = %p.action, error = %violation, "invalid CALLRESULT");
349                        let raw =
350                            serde_json::from_str(payload.get()).unwrap_or(serde_json::Value::Null);
351                        let err = RpcError::from(violation.clone());
352                        let _ = p
353                            .reply
354                            .send(Err(CallFailure::InvalidResponse { violation, raw }));
355                        if version == OcppVersion::V21 {
356                            // 2.1: tell the station its response could not be processed.
357                            return self.send_error(&id, &err, true).await;
358                        }
359                    }
360                }
361                Ok(())
362            }
363            Frame::CallError {
364                id,
365                code,
366                description,
367                details,
368            } => {
369                if let Some(p) = self.take_pending(&id) {
370                    let _ = p.reply.send(Err(CallFailure::Error(rpc::error_from_frame(
371                        &code,
372                        description,
373                        &details,
374                    ))));
375                } else {
376                    tracing::warn!(station = %self.ctx.identity, %id, %code, %description, "CALLERROR for unknown message id");
377                }
378                Ok(())
379            }
380            Frame::CallResultError {
381                id,
382                code,
383                description,
384                ..
385            } => {
386                tracing::warn!(station = %self.ctx.identity, %id, %code, %description, "station could not process our CALLRESULT");
387                Ok(())
388            }
389        }
390    }
391
392    fn take_pending(&mut self, id: &str) -> Option<Pending> {
393        match &self.pending {
394            Some(p) if p.id == id => self.pending.take(),
395            _ => {
396                tracing::debug!(station = %self.ctx.identity, %id, "response for a call that is not pending (late or duplicate)");
397                None
398            }
399        }
400    }
401
402    /// 1.6 and 2.1: ignore. 2.0.1 Part 4 §4.4: answer MessageTypeNotSupported.
403    async fn unknown_message_type(
404        &mut self,
405        text: &str,
406        id: Option<(String, u64)>,
407    ) -> Result<(), ocpp_ws::WsError> {
408        tracing::warn!(station = %self.ctx.identity, frame = %text.chars().take(120).collect::<String>(), "unsupported message type");
409        if self.ctx.version == OcppVersion::V201 {
410            let id = id
411                .map(|(i, _)| i)
412                .or_else(|| Frame::parse(text).ok().map(|f| f.id().to_owned()))
413                .unwrap_or_else(|| rpc::UNKNOWN_MESSAGE_ID.into());
414            let err = RpcError::new(
415                ErrorCode::MessageTypeNotSupported,
416                "message type not supported",
417            );
418            return self.send_error(&id, &err, false).await;
419        }
420        Ok(())
421    }
422}
ocpp_service/ocpp/session.rs

ocpp_service/ocpp/
session.rs
