Compare commits

..

2 commits

Author SHA1 Message Date
Nickiel12
4506d7784b moved state into it's own struct 2024-04-06 10:49:19 -07:00
Nickiel12
6d0b2d652a got a working streams version 2024-04-05 20:58:29 -07:00
10 changed files with 684 additions and 1020 deletions

1072
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -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"

View file

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

View file

@ -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}");
}
}
}
Err(e) => {
error!("Could not serialize app state: {e}");
}
} else {
"Settings could not be deserialized"
}
});
}

170
src/coordinator.rs Normal file
View file

@ -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");
}

View file

@ -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();
}
}
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()),
}) {
match tx.try_send(ApplicationEvent::MoveEvent(MoveEvent {
x: curr_x,
y: curr_y,
})) {
Ok(_) => {}
Err(async_channel::TrySendError::Closed(_)) => break,
Err(async_channel::TrySendError::Full(_)) => {}
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!")}
}
}
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);
}

View file

@ -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
}

96
src/socket_loop.rs Normal file
View file

@ -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));
}
}

View file

@ -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,37 +82,49 @@ 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 {
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_axis.unwrap_or("0".to_string()), msg.y_axis.unwrap_or("0".to_string())).as_str()
format!("X: {:>4} Y: {:>4}", msg.x, msg.y).as_str()
);
button.set_label({
if msg.connected {
}
GuiUpdate::SocketState(v) => {
let label = {
if v {
ip_entry.set_sensitive(false);
port_entry.set_sensitive(false);
"Currently Connected"
@ -119,8 +133,11 @@ pub fn build_ui(app: &Application) {
port_entry.set_sensitive(true);
"Currently Disconnected"
}
});
if msg.connected {
};
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 {
@ -128,6 +145,8 @@ pub fn build_ui(app: &Application) {
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
});

View file

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