qualia_cli/
telemetry_server.rs

1use futures_util::sink::SinkExt;
2use serde::Serialize;
3use sysinfo::System;
4use tokio::net::TcpListener;
5use tokio::sync::broadcast;
6
7#[derive(Serialize, Clone)]
8pub struct TelemetryPayload {
9    pub r#type: String, // "telemetry"
10    pub rss_mb: f64,
11    pub blocks_loaded: usize,
12    pub hot_blocks: Vec<HotBlock>,
13}
14
15#[derive(Serialize, Clone)]
16pub struct HotBlock {
17    pub id: u64,
18    pub source: String, // "local" or "remote"
19}
20
21pub async fn start_telemetry_server(rx: broadcast::Receiver<TelemetryPayload>) {
22    let listener = TcpListener::bind("127.0.0.1:9090")
23        .await
24        .expect("Failed to bind telemetry port");
25    println!("📡 Telemetry WebSocket server running on ws://127.0.0.1:9090");
26
27    while let Ok((stream, _)) = listener.accept().await {
28        let mut rx_clone = rx.resubscribe();
29
30        tokio::spawn(async move {
31            let mut ws_stream = tokio_tungstenite::accept_async(stream)
32                .await
33                .expect("Error during websocket handshake");
34
35            while let Ok(payload) = rx_clone.recv().await {
36                let json = serde_json::to_string(&payload).unwrap();
37                if ws_stream
38                    .send(tokio_tungstenite::tungstenite::Message::Text(json.into()))
39                    .await
40                    .is_err()
41                {
42                    break;
43                }
44            }
45        });
46    }
47}
48
49pub fn get_peak_rss(sys: &mut System) -> f64 {
50    sys.refresh_all();
51    let pid = sysinfo::get_current_pid().unwrap();
52    if let Some(process) = sys.process(pid) {
53        process.memory() as f64 / 1_048_576.0
54    } else {
55        0.0
56    }
57}
qualia_cli/telemetry_server.rs

qualia_cli/
telemetry_server.rs
