kernel/networking/
subscriptions.rs

1use std::sync::{LazyLock, RwLock};
2use common::networking::messages::{TcpServerMessage, SessionID};
3use common::networking::subscription_objects::{DMXConfigForClientState, SubscribeTopic, TopicPayload, UpdateMode};
4use common::networking::subscription_objects::UpdateMode::*;
5use crate::networking::connection_engine::SERVER_STATE;
6
7/// A globally accessible cache that stores the most recent state of all subscribable topics.
8/// This buffer serves a dual purpose: it provides immediate data to newly connected subscribers
9/// upon request, and it acts as the primary data source for background workers managing
10/// continuous-mode or other subscription streams.
11struct ContinuousBuffer {
12    /// The cached visual representation of the current DMX patch and allocations.
13    dmx_config: DMXConfigForClientState,
14}
15
16impl ContinuousBuffer {
17    /// Initializes a new, empty continuous buffer.
18    fn new() -> Self {
19        Self {
20            dmx_config: vec![vec![]]
21        }
22    }
23}
24
25/// Thread-safe static instance of the [`ContinuousBuffer`].
26static CONTINUOUS_BUFFER: LazyLock<RwLock<ContinuousBuffer>> =
27LazyLock::new(|| RwLock::new(ContinuousBuffer::new()));
28
29
30/// Registers a new topic subscription for a specific client session.
31///
32/// Upon successful registration, this function immediately fetches the latest known state
33/// from the [`CONTINUOUS_BUFFER`] and dispatches it to the client. This guarantees
34/// the client interface is instantly populated with valid data.
35///
36/// # Arguments
37///
38/// * `token`       - The unique session identifier of the requesting client.
39/// * `topic`       - The specific engine state topic the client wishes to monitor.
40/// * `update_mode` - The frequency mode (e.g., continuous or on-change) for this subscription.
41pub fn add_subscription(token: &SessionID, topic: &SubscribeTopic, update_mode: &UpdateMode) {
42    let mut server_state = SERVER_STATE.write().unwrap();
43    let user_data = server_state.get_mut(&token);
44
45    if let Some(user_data) = user_data {
46        user_data.subscriptions.push((topic.clone(), update_mode.clone()));
47
48        if let Some((sender, _)) = &user_data.active_connection {
49
50            let buffer = &CONTINUOUS_BUFFER.read().unwrap();
51            let payload = match topic {
52                SubscribeTopic::DMXConfiguration => {
53                    TopicPayload::DMXConfiguration(buffer.dmx_config.clone())
54                }
55            };
56
57            sender.send(
58                TcpServerMessage::TopicUpdate { data: payload }
59            ).unwrap();
60        }
61    }
62}
63
64/// Entry point for engine-side DMX configuration changes.
65///
66/// When the internal DMX allocation changes, this function updates the global cache
67/// and broadcasts the new state to all clients actively listening for changes.
68///
69/// # Arguments
70///
71/// * `data` - The newly computed 2D structure of the client-facing DMX configuration.
72pub fn on_dmx_config_update(data: DMXConfigForClientState) {
73
74    {
75        CONTINUOUS_BUFFER.write().unwrap().dmx_config = data.clone();
76    }
77
78    let data = TopicPayload::DMXConfiguration(data);
79
80    send_updates(data, OnChange);
81}
82
83//TODO
84/// Internal helper function that broadcasts a state payload to all matching client sessions.
85///
86/// Iterates through the global `SERVER_STATE` and transmits the update to any session
87/// that holds an active TCP connection and matches both the topic and the requested update mode.
88/// Currently, all Update-Modes are handled as OnChange
89///
90/// # Arguments
91///
92/// * `payload`     - The actual data payload to be broadcast.
93/// * `update_mode` - The update condition (e.g., `OnChange`) that triggered this broadcast.
94fn send_updates(payload: TopicPayload, update_mode: UpdateMode) {
95    let server_state = SERVER_STATE.read().unwrap();
96
97    let topic = payload.get_topic();
98
99    let data = TcpServerMessage::TopicUpdate {
100        data: payload
101    };
102
103    for session in server_state.values() {
104        if session.subscriptions.contains(&(topic.clone(), update_mode.clone())) {
105            if let Some((ref sender,_)) = session.active_connection {
106                sender.send(data.clone()).unwrap();
107            }
108        }
109    }
110}
kernel/networking/subscriptions.rs

kernel/networking/
subscriptions.rs
