moved state into it's own struct

Cargo.toml

@@ -8,10 +8,13 @@ edition = "2021"
 [dependencies]
 async-channel = "2.2.0"
 config = "0.14.0"
+futures-core = "0.3.30"
+futures-util = { version = "0.3.30", features = ["tokio-io"] }
 gilrs = "0.10.6"
 gtk = { version = "0.8.1", package = "gtk4", features = ["v4_12"] }
 log = "0.4.21"
 serde = { version = "1.0.197", features = ["derive"] }
 simplelog = "0.12.2"
+tokio = { version = "1.37.0", features = ["rt-multi-thread", "time"] }
+tokio-tungstenite = "0.21.0"
 toml = "0.8.12"
-websocket = "0.27.0"

settings.toml

@@ -1,2 +1,2 @@
-ip = "localhost"
+ip = "10.0.0.30"
 port = 8765

src/config.rs

@@ -1,6 +1,7 @@
 use config::{Config, FileFormat};
 use std::fs::File;
 use std::io::Write;
+use log::{error, info};
 
 use crate::ui_code::AppState;
 
@@ -21,16 +22,26 @@ pub fn load_config() -> AppState {
 }
 
 pub fn save_config(config: &AppState) {
-    println!("{}", {
-        if let Ok(toml_str) = toml::to_string(&config) {
-            if let Ok(mut file) = File::create("./settings.toml") {
-                file.write_all(toml_str.as_bytes()).unwrap();
-                ""
-            } else {
-                "File could not be opened"
+    match toml::to_string(&config) {
+        Ok(toml_str) => {
+            match File::create("./settings.toml") {
+                Ok(mut file) => {
+                    match file.write_all(toml_str.as_bytes()) {
+                        Ok(_) => {
+                            info!("Config file saved succesfully");
+                        }
+                        Err(e) => {
+                            error!("Couldn't write config file contents to open file: {e}");
+                        }
+                    }
+                }
+                Err(e) => {
+                    error!("Couldn't open settings file: {e}");
+                }
             }
-        } else {
-            "Settings could not be deserialized"
         }
-    });
+        Err(e) => {
+            error!("Could not serialize app state: {e}");
+        }
+    }
 }

src/coordinator.rs

@@ -0,0 +1,170 @@
+use std::sync::{atomic::{AtomicBool, Ordering}, Arc};
+use std::pin::{pin, Pin};
+
+use async_channel::{Receiver, Sender};
+use futures_util::stream::StreamExt;
+use log::{error, info};
+use tokio::runtime::Handle;
+use tokio_tungstenite::tungstenite::Message;
+
+use crate::{joystick_loop::joystick_loop, socket_loop, ui_code::GuiUpdate};
+
+#[derive(Clone)]
+pub struct MoveEvent {
+    pub x: i32,
+    pub y: i32,
+}
+
+pub enum ApplicationEvent {
+    StartSocket(String),
+    SocketMessage(Message),
+    MoveEvent(MoveEvent),
+}
+
+struct CoordState<'a> {
+    pub to_socket: Option<Sender<Message>>,
+
+    pub sck_send_alive: Arc<AtomicBool>,
+    pub sck_recv_alive: Arc<AtomicBool>,
+    pub joystick_loop_alive: Arc<AtomicBool>,
+
+    pub mec: Pin<&'a mut Receiver<ApplicationEvent>>,
+    pub to_mec: Sender<ApplicationEvent>,
+    pub to_gui: Sender<GuiUpdate>,
+    pub rt: Handle,
+}
+
+impl <'a> CoordState<'a> {
+    pub fn new(mec: Pin<&'a mut Receiver<ApplicationEvent>>, to_mec: Sender<ApplicationEvent>, to_gui: Sender<GuiUpdate>, rt: Handle) -> Self {
+        CoordState {
+            to_socket: None,
+
+            sck_send_alive: Arc::new(AtomicBool::new(false)),
+            sck_recv_alive: Arc::new(AtomicBool::new(false)),
+            joystick_loop_alive: Arc::new(AtomicBool::new(false)),
+
+            mec, to_mec, to_gui, rt
+        }
+    }
+
+    pub async fn socket_send(&mut self, message: Message) {
+        if self.sck_send_alive.load(Ordering::SeqCst) {
+            if let Some(tx) = self.to_socket.take() {
+                if let Err(e) = tx.send(message).await {
+                    error!("There was an error sending to the socket send channel: {e}");
+                    self.socket_close().await;
+                }
+                self.to_socket = Some(tx);
+            }
+        }
+    }
+
+    pub async fn socket_start(&mut self, conn: String) {
+        if !(self.sck_recv_alive.load(Ordering::SeqCst) && !self.sck_send_alive.load(Ordering::SeqCst)) {
+
+            info!("Starting socket");
+            let (to_socket, socket_sender_rx) = async_channel::bounded::<Message>(10);
+
+            self.to_socket = Some(to_socket);
+
+            socket_loop::socket_loop(
+                conn,
+                self.to_mec.clone(),
+                socket_sender_rx,
+                self.sck_send_alive.clone(),
+                self.sck_recv_alive.clone(),
+                self.rt.clone()
+            ).await;
+        }
+    }
+
+    pub async fn check_states(&mut self) {
+        if !self.joystick_loop_alive.load(Ordering::SeqCst) {
+            self.rt.spawn(joystick_loop(self.to_mec.clone(), self.joystick_loop_alive.clone()));
+        }
+
+        if !self.sck_recv_alive.load(Ordering::SeqCst) || !self.sck_send_alive.load(Ordering::SeqCst) {
+            self.socket_close().await;
+
+            if let Err(e) = self.to_gui.send(GuiUpdate::SocketState(false)).await {
+                error!("Cannot send message to gui thread: {e}");
+            }
+        }
+    }
+
+    pub async fn close(&mut self) {
+        info!("closing coord state");
+        self.socket_close().await;
+
+        self.joystick_loop_alive.store(false, Ordering::SeqCst);
+        self.to_gui.close();
+        self.mec.close();
+    }
+
+    pub async fn socket_close(&mut self) {
+            self.sck_send_alive.store(false, Ordering::SeqCst);
+            self.sck_recv_alive.store(false, Ordering::SeqCst);
+            if let Some(tx) = self.to_socket.take() {
+                tx.close();
+            }
+    }
+}
+
+//                      Main_Event_Channel
+pub async fn start_coordinator(mec: Receiver<ApplicationEvent>, to_mec: Sender<ApplicationEvent>, to_gui: Sender<GuiUpdate>, runtime: Handle) {
+    info!("Starting coordinator!"); 
+
+    let mec = pin!(mec);
+
+    let mut state = CoordState::new(
+        mec,
+        to_mec,
+        to_gui,
+        runtime
+    );
+
+    while let Some(msg) = state.mec.next().await {
+
+        state.check_states().await;
+
+        match msg {
+            ApplicationEvent::StartSocket(conn) => {
+                state.socket_start(conn).await;
+            }
+            ApplicationEvent::SocketMessage(socket_message) => {
+                if let Err(e) = state.to_gui.send(GuiUpdate::SocketState(true)).await {
+                    error!("Could not send to gui thread! Closing coordinator: {e}");
+                    state.close().await;
+                    break;
+                }
+
+                state.socket_send(socket_message).await;
+            }
+            ApplicationEvent::MoveEvent(coord) => {
+                if let Err(e) = state.to_gui.send(GuiUpdate::MoveEvent(coord.clone())).await {
+                    panic!("Could not set message to gui channel; Unrecoverable: {e}");
+                }
+
+                if let Some(tx) = state.to_socket.take() {
+                    let message = format!(
+                        "{}{}:{}{}",
+                        if coord.y > 0 { "D" } else { "U" },
+                        coord.y.abs(),
+                        if coord.x > 0 { "R" } else { "L" },
+                        coord.x.abs()
+                    );
+
+                    if let Err(e) = tx.send(Message::Text(message)).await {
+                        error!("Couldn't send message to socket, closing: {e}");
+                        tx.close();
+                    } else {
+                        state.to_socket = Some(tx);
+                    }
+                }
+            }
+        }
+    }
+
+    info!("Stopping Coordinator");
+
+}

src/joystick_loop.rs

@@ -1,75 +1,12 @@
-use crate::config::save_config;
-use crate::ui_code::{AppState, SocketConnectionUpdate};
-use crate::JoystickThreadUpdate;
+use crate::coordinator::{ApplicationEvent, MoveEvent};
 
-use async_channel::{Receiver, Sender};
+use async_channel::Sender;
 use gilrs::{ev::filter::FilterFn, Axis, Button, Event, EventType, Filter, Gilrs, GilrsBuilder};
+use log::{warn, info};
 use std::{
-    sync::{atomic::AtomicBool, Arc},
-    time::{Duration, Instant},
-    panic::{self, AssertUnwindSafe},
-    net::TcpStream,
+    panic::{self, AssertUnwindSafe}, sync::{atomic::AtomicBool, Arc}, time::Duration
 };
-use websocket::client::{sync::Client, ClientBuilder};
-use websocket::Message;
-use log::{info, error};
 
-static MAX_RETRY_ATTEMPTS: u32 = 10;
-
-struct SocketState {
-    pub ip: String,
-    pub port: i32,
-    pub socket: Option<Client<TcpStream>>,
-}
-
-impl SocketState {
-    fn is_connected(&self) -> bool {
-        self.socket.is_some()
-    }
-
-    fn close_websocket(&mut self) {
-        if let Some(ref mut x) = self.socket {
-            info!("closing websocket");
-            let _ = x.send_message(&Message::close());
-            let _ = x.shutdown();
-            self.socket = None;
-        }
-    }
-
-    fn reconnect_websocket(&mut self) -> bool {
-        if self.ip.is_empty() {
-            self.socket = None;
-            return false;
-        }
-
-        if let Ok(mut val) = ClientBuilder::new(format!("ws://{}:{}", &self.ip, self.port).as_str())
-        {
-            if let Ok(val2) = val.connect_insecure() {
-                self.socket = Some(val2);
-                true
-            } else {
-                error!("couldn't connect websocket! : Step 1");
-                self.socket = None;
-                false
-            }
-        } else {
-            error!("couldn't connect websocket! : Step 2");
-            self.socket = None;
-            false
-        }
-    }
-}
-
-struct JTState {
-    pub socket: SocketState,
-    pub try_reconnect: bool,
-    pub retry_attempts: u32,
-
-    pub curr_x: i32,
-    pub curr_y: i32,
-
-    pub last_update_time: Instant,
-}
 
 struct UnknownSlayer;
 impl FilterFn for UnknownSlayer {
@@ -95,68 +32,20 @@ impl FilterFn for UnknownSlayer {
     }
 }
 
-pub fn joystick_websocket_loop(
-    tx: Sender<JoystickThreadUpdate>,
-    do_run: Arc<AtomicBool>,
-    rx: Receiver<SocketConnectionUpdate>,
+pub async fn joystick_loop(
+    tx: Sender<ApplicationEvent>,
+    is_alive: Arc<AtomicBool>
 ) {
     let mut gilrs = GilrsBuilder::new().set_update_state(false).build().unwrap();
 
-    let mut state = JTState {
-        socket: SocketState {
-            ip: String::new(),
-            port: 0,
-            socket: None,
-        },
-        try_reconnect: false,
-        retry_attempts: 0,
+    is_alive.store(true, std::sync::atomic::Ordering::SeqCst);
 
-        curr_x: 0,
-        curr_y: 0,
-
-        last_update_time: Instant::now(),
-    };
+    let mut curr_x: i32 = 0;
+    let mut curr_y: i32 = 0;
+    let mut past_x: i32 = 0;
+    let mut past_y: i32 = 0;
 
     loop {
-        match rx.try_recv() {
-            Ok(msg) => {
-                state.socket.ip = msg.ip;
-                state.socket.port = msg.port as i32;
-
-                save_config(&AppState {
-                    ip: state.socket.ip.clone(),
-                    port: state.socket.port as u32,
-                });
-
-                info!("Connecting to: ws://{}:{}", state.socket.ip, state.socket.port);
-
-                if msg.start_websocket {
-                    if !state.socket.is_connected() {
-                        state.socket.reconnect_websocket();
-                    }
-                } else if state.socket.is_connected() {
-                    state.socket.close_websocket();
-                }
-            }
-            Err(async_channel::TryRecvError::Closed) => break,
-            Err(async_channel::TryRecvError::Empty) => {}
-        }
-
-        if state.try_reconnect {
-            if state.retry_attempts > MAX_RETRY_ATTEMPTS {
-                state.try_reconnect = false;
-            }
-
-            if state.socket.is_connected() {
-                state.try_reconnect = false;
-            } else if state.socket.reconnect_websocket() {
-                state.try_reconnect = false;
-                state.retry_attempts = 0;
-            } else {
-                state.retry_attempts += 1;
-            }
-        }
-
         // catch unwind because some buttons on the joystick will panic the gilrs object
         match panic::catch_unwind(AssertUnwindSafe(|| {
             // get the next event, and if it is an axis we are interested in, update the
@@ -164,15 +53,15 @@ pub fn joystick_websocket_loop(
             while let Some(evt) = gilrs.next_event().filter_ev(&UnknownSlayer {}, &mut gilrs) {
                 match evt.event {
                     gilrs::EventType::AxisChanged(gilrs::Axis::LeftStickY, val, _) => {
-                        state.curr_y = (val * 100.0) as i32;
-                        if state.curr_y > -10 && state.curr_y < 10 {
-                            state.curr_y = 0;
+                        curr_y = (val * 100.0) as i32;
+                        if curr_y > -10 && curr_y < 10 {
+                            curr_y = 0;
                         }
                     }
                     gilrs::EventType::AxisChanged(gilrs::Axis::LeftStickX, val, _) => {
-                        state.curr_x = (val * 100.0) as i32;
-                        if state.curr_x > -10 && state.curr_x < 10 {
-                            state.curr_x = 0;
+                        curr_x = (val * 100.0) as i32;
+                        if curr_x > -10 && curr_x < 10 {
+                            curr_x = 0;
                         }
                     }
                     _ => {}
@@ -185,46 +74,24 @@ pub fn joystick_websocket_loop(
             }
         }
 
-        if state.socket.is_connected()
-            && Instant::now().duration_since(state.last_update_time) >= Duration::from_millis(150)
-        {
-            let mut message: String;
-            if state.curr_y > 0 {
-                message = format!("D{}:", state.curr_y);
-            } else {
-                message = format!("U{}:", state.curr_y.abs());
-            }
+        if curr_x != past_x || curr_y != past_y {
+            past_x = curr_x;
+            past_y = curr_y;
 
-            if state.curr_x > 0 {
-                message.push_str(&format!("R{}", state.curr_x));
-            } else {
-                message.push_str(&format!("L{}", state.curr_x.abs()));
-            }
-
-            if let Some(ref mut websocket_tx) = state.socket.socket {
-                if websocket_tx.send_message(&Message::text(message)).is_ok() {
-                } else {
-                    state.socket.close_websocket();
+            match tx.try_send(ApplicationEvent::MoveEvent(MoveEvent { 
+                x: curr_x,
+                y: curr_y,
+            })) {
+                Ok(_) => {}
+                Err(async_channel::TrySendError::Closed(_)) => {
+                    info!("MEC is closed, stopping Joystick loop");
+                    break 
                 }
+                Err(async_channel::TrySendError::Full(_)) => {warn!("[joystick loop] The MEC is full!")}
             }
-            state.last_update_time = Instant::now();
-            continue;
         }
 
-        match tx.try_send(JoystickThreadUpdate {
-            connected: state.socket.is_connected(),
-            x_axis: Some(state.curr_x.to_string()),
-            y_axis: Some(state.curr_y.to_string()),
-        }) {
-            Ok(_) => {}
-            Err(async_channel::TrySendError::Closed(_)) => break,
-            Err(async_channel::TrySendError::Full(_)) => {}
-        }
-
-        if !do_run.load(std::sync::atomic::Ordering::SeqCst) {
-            info!("Exiting joystick thread");
-            break;
-        }
-        std::thread::sleep(Duration::from_millis(25));
+        tokio::time::sleep(Duration::from_millis(50)).await;
     }
+    is_alive.store(false, std::sync::atomic::Ordering::SeqCst);
 }

src/main.rs

@@ -4,17 +4,15 @@ use gtk::{prelude::*, CssProvider};
 use log::info;
 use simplelog::SimpleLogger;
 use std::env;
+use tokio::runtime;
 
 mod config;
+mod coordinator;
 mod joystick_loop;
+mod socket_loop;
 mod ui_code;
 const APP_ID: &str = "net.nickiel.joystick-controller-client";
 
-pub struct JoystickThreadUpdate {
-    pub connected: bool,
-    pub x_axis: Option<String>,
-    pub y_axis: Option<String>,
-}
 
 fn main() -> glib::ExitCode {
     env::set_var("gtk_csd", "0");
@@ -26,18 +24,27 @@ fn main() -> glib::ExitCode {
         }
     }
 
+    let rt = runtime::Runtime::new().expect("Could not start tokio runtime");
+    let handle = rt.handle().clone();
+
     // Create a new application
     let app = Application::builder().application_id(APP_ID).build();
 
     // Connect to "activate" signal of `app`
     app.connect_startup(|_| load_css());
-    app.connect_activate(ui_code::build_ui);
+    app.connect_activate(move |app| {
+        ui_code::build_ui(app, handle.clone());
+    });
 
     // Run the application
     let exit_code = app.run();
 
     info!("Closing down");
 
+    rt.block_on(async {});
+
+    info!("Tokio runtime has shut down");
+
     exit_code
 }
 

src/socket_loop.rs

@@ -0,0 +1,96 @@
+use std::sync::{atomic::{AtomicBool, Ordering}, Arc};
+
+use async_channel::{Receiver, Sender};
+use futures_util::{
+    stream::{SplitSink, SplitStream},
+    SinkExt, StreamExt,
+};
+use log::{error, info};
+use tokio::{net::TcpStream, runtime::Handle};
+use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
+
+use crate::coordinator::ApplicationEvent;
+
+async fn socket_listen(
+    mec: Sender<ApplicationEvent>,
+    socket_recv_is_alive: Arc<AtomicBool>,
+    mut reader: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
+) {
+    if socket_recv_is_alive.load(std::sync::atomic::Ordering::SeqCst) {
+        while let Some(msg) =  reader.next().await {
+            match msg {
+                Ok(val) => {
+                    if let Err(e) = mec.send(ApplicationEvent::SocketMessage(val)).await {
+                        error!("There was an error sending to Main Event Channel, closing socket recv thread: {e}");
+                        break;
+                    }
+                }
+                Err(e) =>  {
+                    error!("Websocket error: {:#?}", e);
+                }
+            }
+        };
+
+        socket_recv_is_alive.store(false, Ordering::SeqCst);
+    }
+    info!("Closed socket reading thread");
+}
+
+async fn socket_write(
+    to_send: Receiver<Message>,
+    socket_send_is_alive: Arc<AtomicBool>,
+    mut sender: SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
+) {
+    while socket_send_is_alive.load(std::sync::atomic::Ordering::SeqCst) {
+        match to_send.recv().await {
+            Ok(msg) => {
+                if let Err(e) = sender.send(msg).await {
+                    error!("There was an error sending to the socket: {:#?}", e);
+                    break; // Exit loop on error
+                }
+            }
+            Err(e) => {
+                // Channel closed, exit loop
+                error!("To Socket channel is closed! Exiting: {}", e);
+                break;
+            }
+        }
+    }
+    socket_send_is_alive.store(false, Ordering::SeqCst);
+    if let Err(e) = sender.close().await {
+        error!("Error closing websocket sender: {}", e);
+    }
+    info!("Closed socket writing thread");
+}
+
+pub async fn socket_loop(
+    connection_string: String,
+    mec: Sender<ApplicationEvent>,
+    send_to_socket: Receiver<Message>,
+    socket_send_is_alive: Arc<AtomicBool>,
+    socket_recv_is_alive: Arc<AtomicBool>,
+    rt: Handle
+) {
+    info!("Starting Socket Loop");
+    
+
+    socket_send_is_alive.store(true, Ordering::SeqCst);
+    socket_recv_is_alive.store(true, Ordering::SeqCst);
+
+    let socket: Option<WebSocketStream<MaybeTlsStream<TcpStream>>> = match connect_async(connection_string).await {
+        Ok((val, _)) => {
+            info!("Socket connection made successfully");
+            Some(val)
+        }
+        Err(_) => {
+            error!("Couldn't connect to URL");
+            None
+        }
+    };
+
+    if let Some(sckt) = socket {
+        let (outbound, inbound) = sckt.split();
+        rt.spawn(socket_listen(mec, socket_recv_is_alive, inbound));
+        rt.spawn(socket_write(send_to_socket, socket_send_is_alive, outbound));
+    }
+}

src/ui_code.rs

@@ -1,17 +1,14 @@
 use gtk::{glib, prelude::*, Box, Entry, Label, ListBox};
 use gtk::{Application, ApplicationWindow, Button};
-use serde::{Deserialize, Serialize};
-use std::sync::{Arc, atomic::AtomicBool};
 use log::error;
+use serde::{Deserialize, Serialize};
+use tokio::runtime::Handle;
+use tokio_tungstenite::tungstenite::Message;
 
-use crate::config::load_config;
-use crate::{joystick_loop, JoystickThreadUpdate};
+use crate::config::{load_config, save_config};
+// use crate::{joystick_loop, JoystickThreadUpdate};
+use crate::coordinator::{start_coordinator, ApplicationEvent, MoveEvent};
 
-pub struct SocketConnectionUpdate {
-    pub ip: String,
-    pub port: u32,
-    pub start_websocket: bool,
-}
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct AppState {
@@ -28,17 +25,21 @@ impl Default for AppState {
     }
 }
 
-pub fn build_ui(app: &Application) {
+pub enum GuiUpdate {
+    SocketState(bool),
+    MoveEvent(MoveEvent),
+}
+
+pub fn build_ui(app: &Application, runtime: Handle) {
     let initial_settings = load_config();
     let main_box = ListBox::new();
 
-    let do_run: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
-    let do_run2 = do_run.clone();
-    let (tx, rx) = async_channel::bounded::<JoystickThreadUpdate>(4);
+    // Main Event Channel
+    let (to_mec, mec) = async_channel::unbounded::<ApplicationEvent>();
+    let (to_gui, gui_recv) = async_channel::bounded::<GuiUpdate>(10);
 
-    let (tx2, rx2) = async_channel::bounded::<SocketConnectionUpdate>(1);
+    runtime.spawn(start_coordinator(mec, to_mec.clone(), to_gui, runtime.clone()));
 
-    let _ = std::thread::spawn(move || joystick_loop::joystick_websocket_loop(tx, do_run2, rx2));
 
     // let conn_status_label = Label::new(Some(&"No Connection".to_string()));
     let conn_status_label = Label::builder()
@@ -66,6 +67,7 @@ pub fn build_ui(app: &Application) {
         .can_focus(true)
         .build();
     let button = Button::builder().margin_top(12).build();
+    let button2 = Button::builder().margin_top(12).build();
 
     content_box.append(&ip_entry);
     content_box.append(&port_entry);
@@ -80,52 +82,69 @@ pub fn build_ui(app: &Application) {
     main_box.append(&conn_status_label);
     main_box.append(&content_box);
     main_box.append(&axis_label);
+    main_box.append(&button2);
 
     // Connect to "clicked" signal of `button`
-    button.connect_clicked(glib::clone!(@weak ip_entry, @weak port_entry, @strong tx2 => move |_button| {
-        // Set the label to "Hello World!" after the button has been clicked on
-
+    button.connect_clicked(glib::clone!(@weak ip_entry, @weak port_entry, @strong to_mec => move |_button| {
         let ip_text = ip_entry.text();
         let port_text = port_entry.text();
 
         if ip_text.len() > 0 {
             if let Ok(val) = port_text.parse::<u32>() {
-                match tx2.try_send(SocketConnectionUpdate {
+                save_config(&AppState {
                     ip: ip_text.to_string(),
-                    port: val,
-                    start_websocket: ip_entry.get_sensitive(),
-                }) {
+                    port: val
+                });
+
+                match to_mec.try_send(ApplicationEvent::StartSocket(
+                    format!("ws://{}:{}", ip_text, val),
+                )) {
                     Ok(_) => { }
-                    Err(async_channel::TrySendError::Closed(_)) => {panic!("Joystick thread was closed. Unrecoverable")}
-                    Err(e) => {error!("There was an error: {e}")}
+                    Err(async_channel::TrySendError::Closed(_)) => {panic!("Coordinator MEC is closed. Unrecoverable error.")}
+                    Err(e) => {error!("There was an error sending to the MEC: {}", e)}
                 }
             }
         }
     }));
 
+    button2.connect_clicked(glib::clone!(@strong to_mec => move |_button| {
+        if let Err(e) = to_mec.try_send(ApplicationEvent::SocketMessage(Message::text("U45:L10"))) {
+            panic!("There was an error in connect clicked: {e}");
+        }
+    }));
+
     glib::spawn_future_local(
-        glib::clone!(@weak axis_label, @weak button, @weak conn_status_label, @weak ip_entry, @weak port_entry, @strong rx => async move {
-            while let Ok(msg) =  rx.recv().await {
-                axis_label.set_text(
-                    format!("X: {:>4}   Y: {:>4}", msg.x_axis.unwrap_or("0".to_string()), msg.y_axis.unwrap_or("0".to_string())).as_str()
-                );
-                button.set_label({
-                    if msg.connected {
-                        ip_entry.set_sensitive(false);
-                        port_entry.set_sensitive(false);
-                        "Currently Connected"
-                    } else {
-                        ip_entry.set_sensitive(true);
-                        port_entry.set_sensitive(true);
-                        "Currently Disconnected"
+        glib::clone!(@weak axis_label, @weak button, @weak conn_status_label, @weak ip_entry, @weak port_entry, @strong gui_recv => async move {
+            while let Ok(d) =  gui_recv.recv().await {
+                match d {
+                    GuiUpdate::MoveEvent(msg) => {
+                        axis_label.set_text(
+                            format!("X: {:>4}   Y: {:>4}", msg.x, msg.y).as_str()
+                        );
+                    }
+                    GuiUpdate::SocketState(v) => {
+                        let label = {
+                            if v {
+                                ip_entry.set_sensitive(false);
+                                port_entry.set_sensitive(false);
+                                "Currently Connected"
+                            } else {
+                                ip_entry.set_sensitive(true);
+                                port_entry.set_sensitive(true);
+                                "Currently Disconnected"
+                            }
+                        };
+
+                        button.set_label(label);
+                        conn_status_label.set_label(label);
+                        if v {
+                            conn_status_label.set_css_classes(&["YesConnection"]);
+                            button.set_css_classes(&["YesConnection"]);
+                        } else {
+                            conn_status_label.set_css_classes(&["NoConnection"]);
+                            button.set_css_classes(&["NoConnection"]);
+                        }
                     }
-                });
-                if msg.connected {
-                    conn_status_label.set_css_classes(&["YesConnection"]);
-                    button.set_css_classes(&["YesConnection"]);
-                } else {
-                    conn_status_label.set_css_classes(&["NoConnection"]);
-                    button.set_css_classes(&["NoConnection"]);
                 }
             }
         }),
@@ -139,7 +158,6 @@ pub fn build_ui(app: &Application) {
         .build();
 
     window.connect_close_request(move |_| {
-        do_run.store(false, std::sync::atomic::Ordering::SeqCst);
         glib::Propagation::Proceed
     });
 

style.css

@@ -26,5 +26,5 @@ label.JoystickCurrent {
 
 button {
     color: black;
-    font-size: 16pt
+    font-size: 16pt;
 }