csi_webclient/core/
mod.rs

1mod http;
2pub mod messages;
3mod ws;
4
5use crate::core::messages::{CoreCommand, CoreEvent};
6use std::collections::HashMap;
7use std::sync::mpsc::{self, Receiver, Sender};
8
9/// App-facing handle for the background core worker.
10///
11/// The handle is intentionally lightweight and only exposes:
12///
13/// - command submission (`submit`)
14/// - non-blocking event polling (`try_recv`)
15pub struct CoreHandle {
16    cmd_tx: Sender<CoreCommand>,
17    event_rx: Receiver<CoreEvent>,
18}
19
20impl CoreHandle {
21    /// Spawn a new core worker thread and return a handle.
22    pub fn new() -> Self {
23        let (cmd_tx, cmd_rx) = mpsc::channel::<CoreCommand>();
24        let (event_tx, event_rx) = mpsc::channel::<CoreEvent>();
25
26        std::thread::Builder::new()
27            .name("csi-core-worker".to_owned())
28            .spawn(move || worker_loop(cmd_rx, event_tx))
29            .expect("failed to spawn core worker thread");
30
31        Self { cmd_tx, event_rx }
32    }
33
34    /// Submit a command to the core worker.
35    pub fn submit(&self, command: CoreCommand) {
36        let _ = self.cmd_tx.send(command);
37    }
38
39    /// Poll the next core event without blocking the UI thread.
40    pub fn try_recv(&self) -> Option<CoreEvent> {
41        self.event_rx.try_recv().ok()
42    }
43}
44
45impl Drop for CoreHandle {
46    fn drop(&mut self) {
47        let _ = self.cmd_tx.send(CoreCommand::Shutdown);
48    }
49}
50
51fn worker_loop(cmd_rx: Receiver<CoreCommand>, event_tx: Sender<CoreEvent>) {
52    let runtime = match tokio::runtime::Builder::new_multi_thread()
53        .enable_all()
54        .build()
55    {
56        Ok(rt) => rt,
57        Err(err) => {
58            let _ = event_tx.send(CoreEvent::Log(format!(
59                "Failed to initialize async runtime: {err}"
60            )));
61            return;
62        }
63    };
64
65    // One live WebSocket task per device id.
66    let mut ws_tasks: HashMap<String, WsTask> = HashMap::new();
67
68    while let Ok(command) = cmd_rx.recv() {
69        match command {
70            CoreCommand::ExecuteApi(request) => {
71                let event = runtime.block_on(http::execute_api_request(request));
72                let _ = event_tx.send(event);
73            }
74            CoreCommand::ConnectWebSocket { device_id, url } => {
75                // Replace any existing connection for this device.
76                if let Some(task) = ws_tasks.remove(&device_id) {
77                    stop_ws_task(&runtime, task);
78                }
79
80                let (stop_tx, stop_rx) = tokio::sync::oneshot::channel();
81                let event_tx_clone = event_tx.clone();
82                let loop_device_id = device_id.clone();
83                let handle = runtime.spawn(async move {
84                    ws::run_ws_loop(loop_device_id, url, stop_rx, event_tx_clone).await;
85                });
86                ws_tasks.insert(device_id, WsTask { stop_tx, handle });
87            }
88            CoreCommand::DisconnectWebSocket { device_id } => {
89                if let Some(task) = ws_tasks.remove(&device_id) {
90                    stop_ws_task(&runtime, task);
91                }
92                let _ = event_tx.send(CoreEvent::WebSocketDisconnected {
93                    device_id,
94                    reason: "Disconnected".to_owned(),
95                });
96            }
97            CoreCommand::Shutdown => {
98                for (_, task) in ws_tasks.drain() {
99                    stop_ws_task(&runtime, task);
100                }
101                break;
102            }
103        }
104    }
105}
106
107/// A running WebSocket task and its stop signal.
108struct WsTask {
109    stop_tx: tokio::sync::oneshot::Sender<()>,
110    handle: tokio::task::JoinHandle<()>,
111}
112
113fn stop_ws_task(runtime: &tokio::runtime::Runtime, task: WsTask) {
114    let _ = task.stop_tx.send(());
115    let _ = runtime.block_on(task.handle);
116}
csi_webclient/core/mod.rs

csi_webclient/core/
mod.rs
