csi-webclient 0.1.4

Desktop egui client for csi-webserver REST/WebSocket control and CSI stream monitoring
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

use crate::core::messages::CoreEvent;
use futures_util::StreamExt;
use std::sync::mpsc::Sender;
use tokio::sync::oneshot;
use tokio_tungstenite::tungstenite::Message;

/// Run the WebSocket receive loop until remote close, error, or stop signal.
///
/// Every received text/binary payload is forwarded as `CoreEvent::WebSocketFrame`
/// tagged with `device_id`. The caller is responsible for deciding how to
/// interpret payload bytes.
pub async fn run_ws_loop(
    device_id: String,
    url: String,
    mut stop_rx: oneshot::Receiver<()>,
    event_tx: Sender<CoreEvent>,
) {
    let connect_result = tokio_tungstenite::connect_async(url).await;

    let (mut socket, _) = match connect_result {
        Ok(parts) => parts,
        Err(err) => {
            let _ = event_tx.send(CoreEvent::WebSocketDisconnected {
                device_id,
                reason: format!("WebSocket connect failed: {err}"),
            });
            return;
        }
    };

    let _ = event_tx.send(CoreEvent::WebSocketConnected {
        device_id: device_id.clone(),
    });

    loop {
        tokio::select! {
            _ = &mut stop_rx => {
                let _ = socket.close(None).await;
                let _ = event_tx.send(CoreEvent::WebSocketDisconnected {
                    device_id,
                    reason: "Disconnected by user".to_owned(),
                });
                break;
            }
            incoming = socket.next() => {
                match incoming {
                    Some(Ok(Message::Binary(bytes))) => {
                        let _ = event_tx.send(CoreEvent::WebSocketFrame {
                            device_id: device_id.clone(),
                            bytes: bytes.to_vec(),
                        });
                    }
                    Some(Ok(Message::Text(text))) => {
                        let _ = event_tx.send(CoreEvent::WebSocketFrame {
                            device_id: device_id.clone(),
                            bytes: text.as_str().as_bytes().to_vec(),
                        });
                    }
                    Some(Ok(Message::Close(frame))) => {
                        let reason = frame
                            .map(|f| f.reason.to_string())
                            .filter(|r| !r.is_empty())
                            .unwrap_or_else(|| "Remote closed connection".to_owned());
                        let _ = event_tx.send(CoreEvent::WebSocketDisconnected { device_id, reason });
                        break;
                    }
                    Some(Ok(_)) => {}
                    Some(Err(err)) => {
                        let _ = event_tx.send(CoreEvent::WebSocketDisconnected {
                            device_id,
                            reason: format!("WebSocket error: {err}"),
                        });
                        break;
                    }
                    None => {
                        let _ = event_tx.send(CoreEvent::WebSocketDisconnected {
                            device_id,
                            reason: "WebSocket stream ended".to_owned(),
                        });
                        break;
                    }
                }
            }
        }
    }
}