gui/network/
tcp_client.rs

1//! # TCP Client Module
2//!
3//! This module manages the TCP network connection from the GUI to the kernel server.
4//! It handles the initial handshake, asynchronous transmission of UI commands, 
5//! and continuous listening for responses (like logs or subscriptions) 
6//! using dedicated background threads.
7
8use crate::network::connection_state::{ConnectionState, SessionState};
9use common::logging::LogLevel::{Error, Info, SuccessEvent};
10use common::networking::messages::{
11    HandshakeRequest, HandshakeResponse, TcpClientMessage, TcpServerMessage,
12};
13use common::{networking, r_log};
14use std::env;
15use std::io::{Read, Write};
16use std::net::{Shutdown, TcpStream};
17use std::sync::atomic::{AtomicBool, Ordering};
18use std::sync::mpsc::{Receiver, Sender};
19use std::sync::Arc;
20use std::thread;
21use crate::controller::{send_ui_event, UiEvent};
22
23/// Manages a persistent TCP connection to the kernel server.
24///
25/// The `TcpClient` encapsulates stream handles and channels. Upon starting, it spawns 
26/// separate background threads for non-blocking reading and writing of network messages.
27pub(crate) struct TcpClient {
28    target: String,
29    tcp_receiver: Receiver<TcpClientMessage>,
30    tcp_sender: Sender<TcpServerMessage>,
31}
32
33impl TcpClient {
34    /// Creates a new [`TcpClient`] instance without starting it immediately.
35    ///
36    /// # Arguments
37    /// * `target` - The IP address or hostname of the target server (e.g., "127.0.0.1:6767").
38    /// * `tcp_receiver` - Channel to receive outgoing messages from the UI.
39    /// * `tcp_sender` - Channel to forward incoming messages from the server to the UI controller.
40    pub(crate) fn new(
41        target: String,
42        tcp_receiver: Receiver<TcpClientMessage>,
43        tcp_sender: Sender<TcpServerMessage>,
44    ) -> Self {
45        Self {
46            target,
47            tcp_receiver,
48            tcp_sender,
49        }
50    }
51
52    /// Helper function to report a connection state change to the GUI controller.
53    ///
54    /// Uses the global `UI_EVENT_SENDER` to trigger a [`UiEvent::SetConnectionState`].
55    ///
56    /// # Arguments
57    /// * `state` - The new connection state to apply in the UI.
58    fn set_connection_state(state: ConnectionState) {
59        if let Some(sender) = crate::UI_EVENT_SENDER.read().unwrap().as_ref() {
60            let _ = sender.send(UiEvent::SetConnectionState { state });
61        }
62    }
63
64    /// Starts the TCP client, performs the version handshake, and spawns the IO threads.
65    ///
66    /// Establishes the TCP connection and sends a `HandshakeRequest`.
67    /// On a version mismatch (`HandshakeResponse::Mismatch`), the function aborts
68    /// and sets the GUI to an `Error` state. Otherwise, the read and write threads 
69    /// are started in the background.
70    pub(crate) fn start_tcp_client(&mut self) {
71        r_log!(
72            Info,
73            "Trying to establish connection to {} ...",
74            self.target
75        );
76
77        let mut write_stream = match TcpStream::connect(&self.target) {
78            Ok(stream) => stream,
79            Err(e) => {
80                eprintln!("[System] Unable to connect to {}: {}", self.target, e);
81                return;
82            }
83        };
84
85        let client_version = env!("CARGO_PKG_VERSION");
86        let protocol_hash = networking::messages::get_protocol_version();
87
88        let req = HandshakeRequest {
89            magic_string: "REKTAL".into(),
90            protocol_hash,
91            client_version: client_version.into(),
92        };
93
94        Self::set_connection_state(ConnectionState::ConnectionPending);
95
96        match bincode::serialize(&req) {
97            Ok(payload) => {
98                let len = payload.len() as u32;
99                if let Err(e) = write_stream.write_all(&len.to_be_bytes()) {
100                    r_log!(Error, "Got error while sending length prefix: {}", e);
101                    return;
102                }
103                if let Err(e) = write_stream.write_all(&payload) {
104                    r_log!(Error, "Got error while sending HandshakeRequest: {}", e);
105                    return;
106                }
107            }
108            Err(e) => {
109                r_log!(Error, "Got error while serializing HandshakeRequest: {}", e);
110                return;
111            }
112        }
113
114        // Read HandshakeResponse with length prefix
115        let mut len_buf = [0u8; 4];
116        if let Err(e) = write_stream.read_exact(&mut len_buf) {
117            r_log!(Error, "Error reading HandshakeResponse length: {}", e);
118            Self::set_connection_state(ConnectionState::Error);
119            return;
120        }
121        let msg_len = u32::from_be_bytes(len_buf) as usize;
122        let mut response_buffer = vec![0u8; msg_len];
123        if let Err(e) = write_stream.read_exact(&mut response_buffer) {
124            r_log!(Error, "Error reading HandshakeResponse body: {}", e);
125            Self::set_connection_state(ConnectionState::Error);
126            return;
127        }
128
129        let res = match bincode::deserialize::<HandshakeResponse>(&response_buffer) {
130            Ok(res) => res,
131            Err(e) => {
132                r_log!(Error, "Error deserializing HandshakeResponse: {}", e);
133                return;
134            }
135        };
136
137        match res {
138            HandshakeResponse::Ok => {
139                r_log!(SuccessEvent, "Version {} verified!", client_version);
140                Self::set_connection_state(ConnectionState::Connected {
141                    session_state: SessionState::LoggedOut,
142                });
143            }
144            HandshakeResponse::Mismatch { server_version } => {
145                r_log!(
146                    Error,
147                    "Version mismatch! this client has the version {}. Kernel has the version {}",
148                    client_version,
149                    server_version
150                );
151                Self::set_connection_state(ConnectionState::Error);
152                return;
153            }
154        }
155
156        let read_stream = match write_stream.try_clone() {
157            Ok(stream) => stream,
158            Err(e) => {
159                r_log!(Error, "Failed to clone TcpStream: {}", e);
160                return;
161            }
162        };
163
164        let tcp_sender = self.tcp_sender.clone();
165        let is_disconnecting = Arc::new(AtomicBool::new(false));
166        let is_disconnecting_clone = Arc::clone(&is_disconnecting);
167
168        thread::spawn(move || {
169            Self::listen_tcp(read_stream, tcp_sender, is_disconnecting_clone);
170        });
171
172        Self::write_thread(self, write_stream, is_disconnecting);
173    }
174
175    /// Background thread: Continuously reads incoming messages from the server.
176    ///
177    /// Blocks while waiting for data, reads the 4-byte length prefix, loads the payload, 
178    /// deserializes the [`TcpServerMessage`], and forwards it to the UI controller 
179    /// via the `tcp_sender`.
180    fn listen_tcp(
181        mut read_stream: TcpStream,
182        tcp_sender: Sender<TcpServerMessage>,
183        is_disconnecting: Arc<AtomicBool>,
184    ) {
185        loop {
186            let mut len_buf = [0u8; 4];
187            let mut response_buffer = match read_stream.read_exact(&mut len_buf) {
188                Ok(_) => {
189                    let msg_len = u32::from_be_bytes(len_buf) as usize;
190                    vec![0u8; msg_len]
191                }
192                Err(e) => {
193                    if is_disconnecting.load(Ordering::SeqCst) {
194                        r_log!(Info, "TCP connection closed (requested by client).");
195                    } else {
196                        r_log!(Error, "TCP connection lost unexpectedly from server: {}", e);
197                        Self::set_connection_state(ConnectionState::Error);
198                    }
199                    break;
200                }
201            };
202
203            match read_stream.read_exact(&mut response_buffer) {
204                Ok(_) => match bincode::deserialize::<TcpServerMessage>(&response_buffer) {
205                    Ok(kernel_msg) => {
206                        if let Err(e) = tcp_sender.send(kernel_msg) {
207                            if is_disconnecting.load(Ordering::SeqCst) {
208                                r_log!(Info, "TCP receiver channel closed, stopping listen thread.");
209                            } else {
210                                r_log!(Error, "Error sending TcpServerMessage: {}", e);
211                            }
212                            break;
213                        }
214                    }
215                    Err(e) => {
216                        r_log!(Error, "Error deserializing TcpServerMessage: {}", e);
217                    }
218                },
219                Err(e) => {
220                    if is_disconnecting.load(Ordering::SeqCst) {
221                        r_log!(Info, "TCP connection closed while reading body.");
222                    } else {
223                        r_log!(Error, "Error reading from TcpStream: {}", e);
224                        Self::set_connection_state(ConnectionState::Error);
225                    }
226                    break;
227                }
228            }
229        }
230    }
231
232    /// Background thread: Waits for outgoing messages from the GUI and sends them.
233    ///
234    /// Receives messages from the `tcp_receiver` channel (blocking), serializes them 
235    /// into bytes, and sends them with a length prefix over the outgoing TCP stream to the kernel.
236    fn write_thread(&self, mut write_stream: TcpStream, is_disconnecting: Arc<AtomicBool>) {
237        while let Ok(message) = self.tcp_receiver.recv() {
238            match bincode::serialize(&message) {
239                Ok(payload) => {
240                    let len = payload.len() as u32;
241                    if let Err(e) = write_stream.write_all(&len.to_be_bytes()) {
242                        r_log!(
243                            Error,
244                            "Got error while sending length prefix: {} Stopped write-Thread",
245                            e
246                        );
247                        break;
248                    }
249                    if let Err(e) = write_stream.write_all(&payload) {
250                        r_log!(
251                            Error,
252                            "Got error while sending to Server: {} Stopped write-Thread",
253                            e
254                        );
255                        break;
256                    }
257                }
258                Err(e) => {
259                    r_log!(Error, "Got error while serializing response: {}", e);
260                    break;
261                }
262            }
263        }
264        is_disconnecting.store(true, Ordering::SeqCst);
265        Self::close_tcp_connection(&write_stream);
266    }
267
268    /// Closes the TCP connection cleanly in both directions.
269    ///
270    /// Using `shutdown(Shutdown::Both)` immediately interrupts traffic and 
271    /// unblocks the reading thread so it can terminate gracefully.
272    fn close_tcp_connection(stream: &TcpStream) {
273        if let Err(e) = stream.shutdown(Shutdown::Both) {
274            r_log!(Error, "Failed to shutdown TcpStream: {}", e);
275        }
276        send_ui_event(UiEvent::SetConnectionState {
277            state: ConnectionState::Disconnected,
278        });
279    }
280}
gui/network/tcp_client.rs

gui/network/
tcp_client.rs
