From f8885db3b310c47be94fef30dbaa04ccd1bf6560 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20We=C3=9Fel?= Date: Sat, 17 Jan 2026 12:47:44 +0100 Subject: [PATCH 1/4] docs: document websocket "current_session" event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Florian Weßel --- docs/WebSocket/WebSocket.md | 71 +++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/docs/WebSocket/WebSocket.md b/docs/WebSocket/WebSocket.md index a8e839b..3968975 100644 --- a/docs/WebSocket/WebSocket.md +++ b/docs/WebSocket/WebSocket.md @@ -12,6 +12,7 @@ These are primarily informations that either change often or it doesn't make sen - [Sector finished](#sector-finished-broadcast) - [Lap finished](#lap-finished-broadcast) - [Current Laptime](#current-laptime-broadcast) + - [Current Session](#current-session) - [GNSS Data /v1/gnss_data](#gnss-data-v1gnss_data) - [Success](#success-1) - [Events](#events-1) @@ -103,6 +104,76 @@ Example JSON object: "time": "00:50:456.789" } } + +``` +### Current Session +The current session event provides the complete data of the ongoing session. +It contains information about the track, laps, and log points recorded so far in the session. +This event is sent when a websocket connection is established to provide the current state of the session. +A client can use this data to synchronize its state with the ongoing session. +The client should not display any other data until this event is received to ensure consistency. + +The data object in the event contains the full session data structured in the same as would be downloaded via the [REST API](https://github.com/mFlorianW/rapid-rusty/blob/main/docs/REST/Session.md#get-v1sessionsid). + +Example JSON object: +```json +{ + "event": "current_session", + "data": { + "session": { + "id": 0, + "date": "01.01.1970", + "time": "13:00:00.000", + "track": { + "name": "Oschersleben", + "startline": { + "latitude": 52.025833, + "longitude": 11.279166 + }, + "finishline": { + "latitude": 52.025833, + "longitude": 11.279166 + }, + "sectors": [ + { + "latitude": 52.025833, + "longitude": 11.279166 + }, + { + "latitude": 52.025833, + "longitude": 11.279166 + } + ] + }, + "laps": [ + { + "sectors": [ + "00:00:25.144", + "00:00:25.144", + "00:00:25.144", + "00:00:25.144" + ], + "log_points": [ + { + "velocity": 100, + "longitude": 11, + "latitude": 52, + "time": "00:00:00.000", + "date": "01.01.1970" + }, + { + "velocity": 100, + "longitude": 11, + "latitude": 52, + "time": "00:00:00.000", + "date": "01.01.1970" + } + ] + } + ] + } + } +} ``` ## GNSS Data /v1/gnss_data From 9fd3b91ee4144cc8a4d87242ac764b8f93f8bdca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20We=C3=9Fel?= Date: Sun, 18 Jan 2026 18:02:23 +0100 Subject: [PATCH 2/4] chore: update licensing infos if they don't match MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Florian Weßel --- .pre-commit-config.yaml | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b1f7edd..ccb8f8f 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -17,9 +17,10 @@ repos: - repo: local hooks: - - id: check-license - name: check_license - entry: reuse lint + - id: reuse-annotate + name: REUSE Annotate + entry: reuse + args: [annotate, -c, "All contributors", -l, "GPL-2.0-or-later", --merge-copyrights ] language: python - pass_filenames: false - additional_dependencies: ["reuse"] + pass_filenames: true + types_or: ["rust"] From 92cc89def3cf83a794a47422314a2709c220e865 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20We=C3=9Fel?= Date: Sun, 18 Jan 2026 18:03:32 +0100 Subject: [PATCH 3/4] chore: add editorconfig that removes trailing whitespaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Florian Weßel --- .editorconfig | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 .editorconfig diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..7bf2224 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,3 @@ +# .editorconfig file +[*] +trim_trailing_whitespace = true From d52990b1716103171f8c249486527a9c183468b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20We=C3=9Fel?= Date: Mon, 19 Jan 2026 21:02:47 +0100 Subject: [PATCH 4/4] feat: live_session send current_session as synchronisation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit before sending other events to the clients the current active session is send to the clients can synchronize it's state of the live session. Signed-off-by: Florian Weßel --- Cargo.lock | 1 + module_core/src/lib.rs | 16 +- modules/active_session/src/lib.rs | 12 +- .../tests/tests_active_session.rs | 66 +++++++- modules/rest/Cargo.toml | 1 + modules/rest/src/lib.rs | 51 +++++- modules/rest/src/live_session.rs | 156 +++++++++++++++++- modules/rest/tests/tests_live_session.rs | 134 +++++++++++---- 8 files changed, 396 insertions(+), 41 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 89717e1..d9bc907 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1858,6 +1858,7 @@ dependencies = [ "common", "futures-util", "module_core", + "rand 0.9.2", "reqwest", "rocket", "rocket_ws", diff --git a/module_core/src/lib.rs b/module_core/src/lib.rs index b3b0dbb..b9c7336 100644 --- a/module_core/src/lib.rs +++ b/module_core/src/lib.rs @@ -1,4 +1,4 @@ -// SPDX-FileCopyrightText: 2025 All contributors +// SPDX-FileCopyrightText: 2025, 2026 All contributors // // SPDX-License-Identifier: GPL-2.0-or-later @@ -53,6 +53,7 @@ impl Event { EventKind::SaveSessionRequestEvent(req) => Some(req.id), EventKind::LoadSessionRequestEvent(req) => Some(req.id), EventKind::DeleteSessionRequestEvent(req) => Some(req.id), + EventKind::CurrentSessionRequestEvent(req) => Some(req.id), EventKind::LoadStoredSessionIdsResponseEvent(res) => Some(res.id), EventKind::SaveSessionResponseEvent(res) => Some(res.id), EventKind::LoadSessionResponseEvent(res) => Some(res.id), @@ -60,6 +61,7 @@ impl Event { EventKind::LoadStoredTrackIdsResponseEvent(res) => Some(res.id), EventKind::LoadAllStoredTracksResponseEvent(res) => Some(res.id), EventKind::DetectTrackResponseEvent(res) => Some(res.id), + EventKind::CurrentSessionResponseEvent(res) => Some(res.id), _ => None, } } @@ -78,6 +80,7 @@ impl Event { EventKind::LoadStoredTrackIdsRequest(req) | EventKind::LoadAllStoredTracksRequestEvent(req) | EventKind::DetectTrackRequestEvent(req) => Some(req.sender_addr), + EventKind::CurrentSessionRequestEvent(req) => Some(req.sender_addr), EventKind::LoadStoredSessionIdsResponseEvent(res) => Some(res.receiver_addr), EventKind::SaveSessionResponseEvent(res) => Some(res.receiver_addr), EventKind::LoadSessionResponseEvent(res) => Some(res.receiver_addr), @@ -85,6 +88,7 @@ impl Event { EventKind::LoadStoredTrackIdsResponseEvent(res) => Some(res.receiver_addr), EventKind::LoadAllStoredTracksResponseEvent(res) => Some(res.receiver_addr), EventKind::DetectTrackResponseEvent(res) => Some(res.receiver_addr), + EventKind::CurrentSessionResponseEvent(res) => Some(res.receiver_addr), _ => None, } } @@ -229,6 +233,9 @@ pub type LoadStoredTracksReponsePtr = Arc>>; /// A thread-safe shared pointer to a track detection request. pub type TrackDetectionResponsePtr = Arc>>; +/// A thread-safe shared pointer to a current session response. +pub type CurrentSessionResponsePtr = Arc>>>>; + /// Generic helper macro to extract enum payloads #[macro_export] macro_rules! payload_ref { @@ -337,6 +344,13 @@ pub enum EventKind { /// Event emitted after track detection finishes. /// Contains the `TrackDetectionResponsePtr` with detection results. DetectTrackResponseEvent(TrackDetectionResponsePtr), + + /// Event carrying a request to get the current session. + CurrentSessionRequestEvent(EmptyRequestPtr), + + /// Event emitted in response to a current session request. + /// Contains the `CurrentSessionResponsePtr` with the session data. + CurrentSessionResponseEvent(CurrentSessionResponsePtr), } /// A simple asynchronous event bus for publishing and subscribing to [`Event`]s. diff --git a/modules/active_session/src/lib.rs b/modules/active_session/src/lib.rs index 7b3e60a..9925ced 100644 --- a/modules/active_session/src/lib.rs +++ b/modules/active_session/src/lib.rs @@ -1,4 +1,4 @@ -// SPDX-FileCopyrightText: 2025 All contributors +// SPDX-FileCopyrightText: 2025, 2026 All contributors // // SPDX-License-Identifier: GPL-2.0-or-later @@ -6,7 +6,7 @@ use async_trait::async_trait; use chrono::Utc; use common::{lap::Lap, position::GnssPosition, session::Session}; use module_core::{ - DurationPtr, EventKind, Module, ModuleCtx, Request, SaveSessionRequestPtr, + DurationPtr, EventKind, Module, ModuleCtx, Request, Response, SaveSessionRequestPtr, TrackDetectionResponsePtr, }; use std::sync::{Arc, RwLock}; @@ -132,6 +132,14 @@ impl Module for ActiveSession { EventKind::GnssPositionEvent(gnss_pos) => { self.on_gnss_position(*gnss_pos); } + EventKind::CurrentSessionRequestEvent(request) => { + let resp = Response { + id: request.id, + receiver_addr: request.sender_addr, + data: self.session.as_ref().map(|s| s.clone()), + }; + let _ = self.ctx.publish_event(EventKind::CurrentSessionResponseEvent(resp.into())); + } _ => (), } }, diff --git a/modules/active_session/tests/tests_active_session.rs b/modules/active_session/tests/tests_active_session.rs index 68d8cb4..ed8bf0a 100644 --- a/modules/active_session/tests/tests_active_session.rs +++ b/modules/active_session/tests/tests_active_session.rs @@ -1,11 +1,11 @@ -// SPDX-FileCopyrightText: 2025 All contributors +// SPDX-FileCopyrightText: 2025, 2026 All contributors // // SPDX-License-Identifier: GPL-2.0-or-later use active_session::ActiveSession; use common::{lap::Lap, position::GnssPosition, test_helper::track::get_track}; use module_core::{ - Event, EventBus, EventKind, EventKindType, Module, Response, payload_ref, + Event, EventBus, EventKind, EventKindType, Module, Request, Response, payload_ref, test_helper::{register_response_event, stop_module, wait_for_event}, }; use std::time::Duration; @@ -163,3 +163,65 @@ async fn test_store_log_points() { stop_module(&eb, &mut active_session).await; } + +#[tokio::test] +#[test_log::test] +async fn test_current_session_request_response() { + let eb = EventBus::default(); + let mut active_session = create_module(&eb); + + // Before emitting the lap start wait for the track detected event. + let _track_event = wait_for_event( + &mut eb.subscribe(), + Duration::from_millis(100), + EventKindType::DetectTrackResponseEvent, + ) + .await; + eb.publish(&Event { + kind: EventKind::LapStartedEvent, + }); + eb.publish(&Event { + kind: EventKind::LapFinishedEvent(std::time::Duration::from_secs_f32(30.750).into()), + }); + debug!("Waiting for CurrentSessionRequestEvent..."); + eb.publish(&Event { + kind: EventKind::CurrentSessionRequestEvent( + Request { + id: 20, + sender_addr: 200, + data: {}, + } + .into(), + ), + }); + let current_session_event = wait_for_event( + &mut eb.subscribe(), + Duration::from_millis(100), + EventKindType::CurrentSessionResponseEvent, + ) + .await; + + //scope is needed to clear the rwlock at the end. + { + let session = match payload_ref!( + current_session_event.kind, + EventKind::CurrentSessionResponseEvent + ) { + Some(response) => response.data.clone(), + None => { + panic!("Received session doesn't have a payload"); + } + }; + let session_lock = session.expect("Session data is None"); + let session = session_lock.read().unwrap(); + assert_eq!(session.laps.len(), 1); + let lap = Lap { + sectors: vec![], + log_points: vec![], + }; + assert_eq!(session.laps[0], lap); + assert_eq!(session.track, get_track()); + } + + stop_module(&eb, &mut active_session).await; +} diff --git a/modules/rest/Cargo.toml b/modules/rest/Cargo.toml index 4490d5b..ee8ad79 100644 --- a/modules/rest/Cargo.toml +++ b/modules/rest/Cargo.toml @@ -15,6 +15,7 @@ serde_json.workspace = true rocket = { version = "~0.5", features = ["json"] } rocket_ws = { version = "~0.1" } +rand ={ version = "~0.9" } [dev-dependencies] reqwest = { version = "~0.12", features = ["json"] } diff --git a/modules/rest/src/lib.rs b/modules/rest/src/lib.rs index 7d3c08a..4f4f707 100644 --- a/modules/rest/src/lib.rs +++ b/modules/rest/src/lib.rs @@ -1,4 +1,4 @@ -// SPDX-FileCopyrightText: 2025 All contributors +// SPDX-FileCopyrightText: 2025, 2026 All contributors // // SPDX-License-Identifier: GPL-2.0-or-later @@ -12,6 +12,7 @@ use rocket::{ serde::{Serialize, json::Json}, }; use std::{ + collections::HashMap, env, net::Ipv4Addr, sync::{Arc, RwLock}, @@ -37,6 +38,7 @@ pub(crate) struct RestCtx { ctx: ModuleCtx, module_addr: u64, request_id: u64, + connections: HashMap, } impl RestCtx { @@ -49,6 +51,52 @@ impl RestCtx { self.request_id += 1; id } + + /// Register a new connection in the internal registry. + /// + /// Inserts the given connection ID with an initial state of `false` + /// (e.g., not yet synchronized). If the ID already exists, its value is + /// replaced. + /// + /// - conn_id: Unique identifier for the client/connection. + pub fn register_connection(&mut self, conn_id: &str) { + self.connections.insert(conn_id.to_owned(), false); + } + + /// Unregisters a connection from the internal registry. + /// + /// Removes the entry associated with the given connection ID. + /// + /// - conn_id: Unique identifier for the client/connection. + pub fn unregister_connection(&mut self, conn_id: &str) { + self.connections.remove(conn_id); + } + + /// Checks if a connection is marked as synchronized. + /// + /// Returns `true` if the connection with the given ID is marked as synchronized, + /// otherwise returns `false`. + /// + /// - conn_id: Unique identifier for the client/connection. + pub fn is_connection_synced(&self, conn_id: &str) -> bool { + match self.connections.get(conn_id) { + Some(synced) => *synced, + None => false, + } + } + + /// Sets the synchronization state of a connection. + /// + /// Updates the internal state for the connection with the given ID + /// to reflect whether it is synchronized or not. + /// - conn_id: Unique identifier for the client/connection. + /// + /// - synced: New synchronization state to set for the connection. + pub fn set_connection_synced(&mut self, conn_id: &str, synced: bool) { + if let Some(entry) = self.connections.get_mut(conn_id) { + *entry = synced; + } + } } impl Rest { @@ -67,6 +115,7 @@ impl Rest { ctx, module_addr: 0xff, request_id: 0, + connections: HashMap::new(), })), } } diff --git a/modules/rest/src/live_session.rs b/modules/rest/src/live_session.rs index 9a3022e..7bd8d08 100644 --- a/modules/rest/src/live_session.rs +++ b/modules/rest/src/live_session.rs @@ -6,11 +6,17 @@ use crate::RestCtx; use crate::rocket::futures::StreamExt; use crate::rocket::futures::TryStreamExt; use common::serde::duration; +use common::session::Session; use module_core::EventKind; +use module_core::EventKindType; +use module_core::Request; +use module_core::payload_ref; +use rand::{Rng, distr::Alphanumeric, rng}; use rocket::State; use rocket_ws::Message; use serde::Serialize; use std::sync::Arc; +use std::sync::RwLock; use tokio::sync::Mutex; #[derive(Serialize)] @@ -25,6 +31,17 @@ struct LaptimeData<'a> { time: &'a std::time::Duration, } +#[derive(Serialize)] +struct CurrentSessionEvent<'a> { + event: &'a str, + data: CurrentSessionData<'a>, +} + +#[derive(Serialize)] +struct CurrentSessionData<'a> { + session: &'a Session, +} + #[derive(Serialize)] struct EmptyEvent<'a> { event: &'a str, @@ -79,6 +96,25 @@ fn serialize_empty_event(event: &str) -> String { } } +/// Serializes the current session event into a JSON string. +/// Constructs a `CurrentSessionEvent` with the provided session and +/// +/// returns its JSON representation. +fn serialize_current_session_event(session: &Arc>) -> String { + let session = session.read().unwrap_or_else(|s| s.into_inner()); + let event = CurrentSessionEvent { + event: "current_session", + data: CurrentSessionData { session: &session }, + }; + match serde_json::to_string(&event) { + Ok(json) => json, + Err(e) => { + error!("Failed to serialize current session event: {}", e); + "{}".to_string() + } + } +} + /// WebSocket handler that streams live session updates to clients. /// /// Route: GET /v1/live_session @@ -100,12 +136,26 @@ pub(crate) fn ws_live_session_handler( rocket_ws::Stream! { ws => let ctx = ctx.clone(); let mut stream_ws = ws.into_stream(); + let session_id = generate_connection_id(); let mut event_receiver = { let guard = ctx.lock().await; guard.ctx.receiver.resubscribe() }; + ctx.lock().await.register_connection(&session_id); + info!("WebSocket \"/v1/live_session\" connection established with session_id: {}", session_id); + + match request_current_session(&ctx).await { + Ok(session_ptr) => { + yield Message::Text(serialize_current_session_event(&session_ptr)); + ctx.lock().await.set_connection_synced(&session_id, true); + } + Err(e) => { + error!("Error requesting current session for WebSocket live session initial state: {:?}", e); + } + } + loop { tokio::select!{ event = event_receiver.recv() => { @@ -113,25 +163,46 @@ pub(crate) fn ws_live_session_handler( Ok(event) => { match event.kind { EventKind::QuitEvent => { + ctx.lock().await.unregister_connection(&session_id); info!("Shutting down WebSocket live session handler due to QuitEvent"); break; } EventKind::CurrentLaptimeEvent(laptime) => { - yield Message::Text(serialize_laptime_event(&laptime, "current_laptime")); + if ctx.lock().await.is_connection_synced(&session_id) { + yield Message::Text(serialize_laptime_event(&laptime, "current_laptime")); + } } EventKind::LapStartedEvent => { - yield Message::Text(serialize_empty_event("lap_started")); + if ctx.lock().await.is_connection_synced(&session_id) { + yield Message::Text(serialize_empty_event("lap_started")); + }else{ + match request_current_session(&ctx).await { + Ok(session_ptr) => { + yield Message::Text(serialize_current_session_event(&session_ptr)); + ctx.lock().await.set_connection_synced(&session_id, true); + } + Err(e) => { + error!("Error requesting current session for WebSocket live session sync: {:?}", e); + break; + } + } + } } EventKind::LapFinishedEvent (laptimer)=> { - yield Message::Text(serialize_laptime_event(&laptimer, "lap_finished")); + if ctx.lock().await.is_connection_synced(&session_id) { + yield Message::Text(serialize_laptime_event(&laptimer, "lap_finished")); + } } EventKind::SectorFinishedEvent(sector) => { - yield Message::Text(serialize_laptime_event(§or, "sector_finished")); + if ctx.lock().await.is_connection_synced(&session_id) { + yield Message::Text(serialize_laptime_event(§or, "sector_finished")); + } } _ => {} } } Err(e) => { + ctx.lock().await.unregister_connection(&session_id); error!("Error receiving event in WebSocket live session handler: {}", e); break; } @@ -141,12 +212,14 @@ pub(crate) fn ws_live_session_handler( Some(msg) = stream_ws.next() => { match msg { Ok(Message::Close(_)) => { + ctx.lock().await.unregister_connection(&session_id); info!("WebSocket client disconnected from live session"); break; } Ok(_) => { } Err(e) => { + ctx.lock().await.unregister_connection(&session_id); error!("WebSocket error: {}", e); break; } @@ -156,3 +229,78 @@ pub(crate) fn ws_live_session_handler( } } } + +/// Generates a random connection ID. +/// +/// This function creates a random alphanumeric string of length 16, +/// which can be used as a unique identifier for connections. +/// +/// # Returns +/// A randomly generated connection ID as a `String`. +pub fn generate_connection_id() -> String { + let id: String = rng() + .sample_iter(&Alphanumeric) + .take(16) + .map(char::from) + .collect(); + id +} + +/// Requests the current session from the event bus. +/// +/// Sends a CurrentSessionRequestEvent and waits for the corresponding response. +/// +/// Returns the current session wrapped in an Arc> on success. +/// Returns an std::io::ErrorKind on failure. +/// +/// Arguments: +/// - ctx: Shared RestCtx for publishing and receiving events. +/// +/// Return: +/// - Ok(Arc>) on success. +/// - Err(std::io::ErrorKind) on failure. +async fn request_current_session( + ctx: &Arc>, +) -> Result>, std::io::ErrorKind> { + let mut ctx = ctx.lock().await; + let req_id = ctx.request_id(); + let _ = ctx.ctx.publish_event(EventKind::CurrentSessionRequestEvent( + Request { + id: req_id, + sender_addr: ctx.module_addr, + data: (), + } + .into(), + )); + info!( + "Published CurrentSessionRequestEvent with req_id: {}, addr {:?}", + req_id, ctx.module_addr + ); + let addr = ctx.module_addr; + match ctx + .ctx + .wait_for_event(req_id, addr, &EventKindType::CurrentSessionResponseEvent) + .await + { + Ok(event) => { + let session = match payload_ref!(event.kind, EventKind::CurrentSessionResponseEvent) { + Some(response) => response.data.clone(), + None => { + error!("Received session doesn't have a payload"); + return Err(std::io::ErrorKind::InvalidData); + } + }; + match session { + Some(session) => Ok(session), + None => { + error!("Received session data is None"); + Err(std::io::ErrorKind::InvalidData) + } + } + } + Err(e) => { + error!("Error waiting for CurrentSessionResponseEvent: {:?}", e); + Err(std::io::ErrorKind::TimedOut) + } + } +} diff --git a/modules/rest/tests/tests_live_session.rs b/modules/rest/tests/tests_live_session.rs index 76f48d8..60c9229 100644 --- a/modules/rest/tests/tests_live_session.rs +++ b/modules/rest/tests/tests_live_session.rs @@ -4,12 +4,20 @@ mod test_utils; -use futures_util::StreamExt; -use module_core::{Event, EventBus, EventKind, test_helper::stop_module}; +use common::test_helper::session::get_session; +use futures_util::{StreamExt, stream::SplitStream}; +use module_core::{ + Event, EventBus, EventKind, EventKindType, Response, + test_helper::stop_module, + test_helper::{register_response_event, unregister_response_event}, +}; use serial_test::serial; -use std::time::Duration; +use std::{ + sync::{Arc, RwLock}, + time::Duration, +}; use test_utils::create_module; -use tokio_tungstenite::connect_async; +use tokio_tungstenite::{WebSocketStream, connect_async, tungstenite::Message}; fn get_current_laptime_msg(laptime: Duration, event: &str) -> serde_json::Value { let event = format!( @@ -28,27 +36,60 @@ fn get_lap_started_msg() -> serde_json::Value { serde_json::from_str(event).unwrap() } +fn register_current_session_response_event(eb: &EventBus) { + if register_response_event( + EventKindType::CurrentSessionRequestEvent, + Event { + kind: EventKind::CurrentSessionResponseEvent( + Response { + id: 0, + receiver_addr: 0xff, + data: Some(Arc::new(RwLock::new(get_session()))), + } + .into(), + ), + }, + eb.context(), + ) + .is_err() + { + panic!("Failed to register CurrentSessionResponseEvent"); + } +} + +async fn read_next_websocket_event(read_stream: &mut SplitStream>) -> Message +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + tokio::time::timeout(Duration::from_millis(100), read_stream.next()) + .await + .expect("No message received") + .expect("Error reading message") + .expect("No message in time") +} + +fn unregister_current_session_response_event(eb: &EventBus) { + unregister_response_event(eb.id(), &EventKindType::CurrentSessionRequestEvent) +} + #[tokio::test] #[test_log::test] #[serial] async fn test_current_laptime() { let eb = EventBus::default(); let mut rest = create_module(eb.context()); + register_current_session_response_event(&eb); let (ws_stream, _) = connect_async("ws://localhost:27015/v1/live_session") .await .expect("Failed to connect to WebSocket"); let (_, mut read) = ws_stream.split(); + let _ = read_next_websocket_event(&mut read).await; // Consume the current_session event + eb.publish(&Event { kind: EventKind::CurrentLaptimeEvent(Duration::from_millis(1).into()), }); - - let msg = tokio::time::timeout(Duration::from_millis(100), read.next()) - .await - .expect("No message received") - .expect("Error reading message") - .expect("No message in time"); - + let msg = read_next_websocket_event(&mut read).await; match msg { tokio_tungstenite::tungstenite::Message::Text(text) => { let expected = get_current_laptime_msg(Duration::from_millis(1), "current_laptime"); @@ -57,6 +98,8 @@ async fn test_current_laptime() { } _ => panic!("Unexpected message type received. Msg: {:?}", msg), } + + unregister_current_session_response_event(&eb); stop_module(&eb, &mut rest).await; } @@ -66,21 +109,18 @@ async fn test_current_laptime() { async fn test_lap_finished_event() { let eb = EventBus::default(); let mut rest = create_module(eb.context()); + register_current_session_response_event(&eb); let (ws_stream, _) = connect_async("ws://localhost:27015/v1/live_session") .await .expect("Failed to connect to WebSocket"); let (_, mut read) = ws_stream.split(); + let _ = read_next_websocket_event(&mut read).await; // Consume the current_session event + eb.publish(&Event { kind: EventKind::LapFinishedEvent(Duration::from_millis(1).into()), }); - - let msg = tokio::time::timeout(Duration::from_millis(100), read.next()) - .await - .expect("No message received") - .expect("Error reading message") - .expect("No message in time"); - + let msg = read_next_websocket_event(&mut read).await; match msg { tokio_tungstenite::tungstenite::Message::Text(text) => { let expected = get_current_laptime_msg(Duration::from_millis(1), "lap_finished"); @@ -89,6 +129,8 @@ async fn test_lap_finished_event() { } _ => panic!("Unexpected message type received. Msg: {:?}", msg), } + + unregister_current_session_response_event(&eb); stop_module(&eb, &mut rest).await; } @@ -98,21 +140,18 @@ async fn test_lap_finished_event() { async fn test_sector_finished() { let eb = EventBus::default(); let mut rest = create_module(eb.context()); + register_current_session_response_event(&eb); let (ws_stream, _) = connect_async("ws://localhost:27015/v1/live_session") .await .expect("Failed to connect to WebSocket"); let (_, mut read) = ws_stream.split(); + let _ = read_next_websocket_event(&mut read).await; // Consume the current_session event + eb.publish(&Event { kind: EventKind::SectorFinishedEvent(Duration::from_millis(1).into()), }); - - let msg = tokio::time::timeout(Duration::from_millis(100), read.next()) - .await - .expect("No message received") - .expect("Error reading message") - .expect("No message in time"); - + let msg = read_next_websocket_event(&mut read).await; match msg { tokio_tungstenite::tungstenite::Message::Text(text) => { let expected = get_current_laptime_msg(Duration::from_millis(1), "sector_finished"); @@ -121,6 +160,8 @@ async fn test_sector_finished() { } _ => panic!("Unexpected message type received. Msg: {:?}", msg), } + + unregister_current_session_response_event(&eb); stop_module(&eb, &mut rest).await; } @@ -130,28 +171,59 @@ async fn test_sector_finished() { async fn test_lap_started_event() { let eb = EventBus::default(); let mut rest = create_module(eb.context()); + register_current_session_response_event(&eb); let (ws_stream, _) = connect_async("ws://localhost:27015/v1/live_session") .await .expect("Failed to connect to WebSocket"); let (_, mut read) = ws_stream.split(); + let _ = read_next_websocket_event(&mut read).await; // Consume the current_session event + // eb.publish(&Event { kind: EventKind::LapStartedEvent, }); + let msg = read_next_websocket_event(&mut read).await; + match msg { + tokio_tungstenite::tungstenite::Message::Text(text) => { + let expected = get_lap_started_msg(); + let msg = serde_json::from_slice::(text.as_bytes()).unwrap(); + assert_eq!(msg, expected, "Laptime message does not match expected"); + } + _ => panic!("Unexpected message type received. Msg: {:?}", msg), + } + + unregister_current_session_response_event(&eb); + stop_module(&eb, &mut rest).await; +} - let msg = tokio::time::timeout(Duration::from_millis(100), read.next()) +#[tokio::test] +#[test_log::test] +#[serial] +async fn test_current_session_event_on_connect() { + let eb = EventBus::default(); + let mut rest = create_module(eb.context()); + register_current_session_response_event(&eb); + + let (ws_stream, _) = connect_async("ws://localhost:27015/v1/live_session") .await - .expect("No message received") - .expect("Error reading message") - .expect("No message in time"); + .expect("Failed to connect to WebSocket"); + let (_, mut read) = ws_stream.split(); + let msg = read_next_websocket_event(&mut read).await; match msg { tokio_tungstenite::tungstenite::Message::Text(text) => { - let expected = get_lap_started_msg(); + let expected = serde_json::json!({ + "event": "current_session", + "data": { + "session": get_session() + } + }); let msg = serde_json::from_slice::(text.as_bytes()).unwrap(); - assert_eq!(msg, expected, "Laptime message does not match expected"); + assert_eq!(msg, expected, "Session message does not match expected"); } _ => panic!("Unexpected message type received. Msg: {:?}", msg), } + + unregister_current_session_response_event(&eb); stop_module(&eb, &mut rest).await; }