TheLanc3/lan-mouse
1
0
Fork 0
mirror of https://github.com/feschber/lan-mouse.git synced 2026-04-15 00:11:26 +03:00
Code Issues Packages Projects Releases Wiki Activity

explicit state synchronisation (#129)

Browse Source 
prevents unnecessary state updates and makes synchronization less error prone
This commit is contained in:
Ferdinand Schober
2024-05-12 00:49:53 +02:00
committed by GitHub
parent 9969f997d3
commit e9738fc024
6 changed files with 91 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/frontend.rs
View File

@@ -107,16 +107,18 @@ pub enum FrontendRequest {
UpdatePosition(ClientHandle, Position), UpdatePosition(ClientHandle, Position),
/// update fix-ips /// update fix-ips
UpdateFixIps(ClientHandle, Vec<IpAddr>), UpdateFixIps(ClientHandle, Vec<IpAddr>),
/// request the state of the given client
GetState(ClientHandle),
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FrontendEvent { pub enum FrontendEvent {
/// a client was created /// a client was created
Created(ClientHandle, ClientConfig, ClientState), Created(ClientHandle, ClientConfig, ClientState),
/// a client was updated /// no such client
Updated(ClientHandle, ClientConfig), NoSuchClient(ClientHandle),
/// state changed /// state changed
StateChange(ClientHandle, ClientState), State(ClientHandle, ClientConfig, ClientState),
/// the client was deleted /// the client was deleted
Deleted(ClientHandle), Deleted(ClientHandle),
/// new port, reason of failure (if failed) /// new port, reason of failure (if failed)
@@ -235,9 +237,7 @@ impl FrontendListener {
let json = serde_json::to_string(&notify).unwrap(); let json = serde_json::to_string(&notify).unwrap();
let payload = json.as_bytes(); let payload = json.as_bytes();
let len = payload.len().to_be_bytes(); let len = payload.len().to_be_bytes();
log::debug!("json: {json}, len: {}", payload.len()); log::debug!("broadcasting event to streams: {json}");
log::debug!("broadcasting event to streams: {:?}", self.tx_streams);
let mut keep = vec![]; let mut keep = vec![];
// TODO do simultaneously // TODO do simultaneously
for tx in self.tx_streams.iter_mut() { for tx in self.tx_streams.iter_mut() {

66
src/frontend/cli.rs
View File

@@ -100,6 +100,18 @@ impl<'a> Cli<'a> {
} }
} }
async fn update_client(&mut self, handle: ClientHandle) -> Result<()> {
self.send_request(FrontendRequest::GetState(handle)).await?;
loop {
let event = self.await_event().await?;
self.handle_event(event.clone());
if let FrontendEvent::State(_, _, _) | FrontendEvent::NoSuchClient(_) = event {
break;
}
}
Ok(())
}
async fn execute(&mut self, cmd: Command) -> Result<()> { async fn execute(&mut self, cmd: Command) -> Result<()> {
match cmd { match cmd {
Command::None => {} Command::None => {}
@@ -125,14 +137,8 @@ impl<'a> Cli<'a> {
FrontendRequest::UpdatePosition(handle, pos), FrontendRequest::UpdatePosition(handle, pos),
] { ] {
self.send_request(request).await?; self.send_request(request).await?;
loop {
let event = self.await_event().await?;
self.handle_event(event.clone());
if let FrontendEvent::Updated(_, _) = event {
break;
}
}
} }
self.update_client(handle).await?;
} }
Command::Disconnect(id) => { Command::Disconnect(id) => {
self.send_request(FrontendRequest::Delete(id)).await?; self.send_request(FrontendRequest::Delete(id)).await?;
@@ -148,26 +154,12 @@ impl<'a> Cli<'a> {
Command::Activate(id) => { Command::Activate(id) => {
self.send_request(FrontendRequest::Activate(id, true)) self.send_request(FrontendRequest::Activate(id, true))
.await?; .await?;
loop { self.update_client(id).await?;
let event = self.await_event().await?;
self.handle_event(event.clone());
if let FrontendEvent::StateChange(_, _) = event {
self.handle_event(event);
break;
}
}
} }
Command::Deactivate(id) => { Command::Deactivate(id) => {
self.send_request(FrontendRequest::Activate(id, false)) self.send_request(FrontendRequest::Activate(id, false))
.await?; .await?;
loop { self.update_client(id).await?;
let event = self.await_event().await?;
self.handle_event(event.clone());
if let FrontendEvent::StateChange(_, _) = event {
self.handle_event(event);
break;
}
}
} }
Command::List => { Command::List => {
self.send_request(FrontendRequest::Enumerate()).await?; self.send_request(FrontendRequest::Enumerate()).await?;
@@ -182,25 +174,12 @@ impl<'a> Cli<'a> {
Command::SetHost(handle, host) => { Command::SetHost(handle, host) => {
let request = FrontendRequest::UpdateHostname(handle, Some(host.clone())); let request = FrontendRequest::UpdateHostname(handle, Some(host.clone()));
self.send_request(request).await?; self.send_request(request).await?;
loop { self.update_client(handle).await?;
let event = self.await_event().await?;
self.handle_event(event.clone());
if let FrontendEvent::Updated(_, _) = event {
self.handle_event(event);
break;
}
}
} }
Command::SetPort(handle, port) => { Command::SetPort(handle, port) => {
let request = FrontendRequest::UpdatePort(handle, port.unwrap_or(DEFAULT_PORT)); let request = FrontendRequest::UpdatePort(handle, port.unwrap_or(DEFAULT_PORT));
self.send_request(request).await?; self.send_request(request).await?;
loop { self.update_client(handle).await?;
let event = self.await_event().await?;
self.handle_event(event.clone());
if let FrontendEvent::Updated(_, _) = event {
break;
}
}
} }
Command::Help => { Command::Help => {
for cmd_type in [ for cmd_type in [
@@ -244,8 +223,11 @@ impl<'a> Cli<'a> {
eprintln!(); eprintln!();
self.clients.push((h, c, s)); self.clients.push((h, c, s));
} }
FrontendEvent::Updated(h, c) => { FrontendEvent::NoSuchClient(h) => {
if let Some((_, config, _)) = self.find_mut(h) { eprintln!("no such client: {h}");
}
FrontendEvent::State(h, c, s) => {
if let Some((_, config, state)) = self.find_mut(h) {
let old_host = config.hostname.clone().unwrap_or("\"\"".into()); let old_host = config.hostname.clone().unwrap_or("\"\"".into());
let new_host = c.hostname.clone().unwrap_or("\"\"".into()); let new_host = c.hostname.clone().unwrap_or("\"\"".into());
if old_host != new_host { if old_host != new_host {
@@ -261,10 +243,6 @@ impl<'a> Cli<'a> {
eprintln!("client {h} ips updated: {:?}", c.fix_ips) eprintln!("client {h} ips updated: {:?}", c.fix_ips)
} }
*config = c; *config = c;
}
}
FrontendEvent::StateChange(h, s) => {
if let Some((_, _, state)) = self.find_mut(h) {
if state.active ^ s.active { if state.active ^ s.active {
eprintln!( eprintln!(
"client {h} {}", "client {h} {}",

7
src/frontend/gtk.rs
View File

@@ -126,12 +126,11 @@ fn build_ui(app: &Application) {
FrontendEvent::Deleted(client) => { FrontendEvent::Deleted(client) => {
window.delete_client(client); window.delete_client(client);
} }
FrontendEvent::Updated(handle, client) => { FrontendEvent::State(handle, config, state) => {
window.update_client_config(handle, client); window.update_client_config(handle, config);
}
FrontendEvent::StateChange(handle, state) => {
window.update_client_state(handle, state); window.update_client_state(handle, state);
} }
FrontendEvent::NoSuchClient(_) => { }
FrontendEvent::Error(e) => { FrontendEvent::Error(e) => {
window.show_toast(e.as_str()); window.show_toast(e.as_str());
}, },

82
src/frontend/gtk/window.rs
View File

@@ -51,6 +51,10 @@ impl Window {
.expect("Could not get clients") .expect("Could not get clients")
} }
fn client_by_idx(&self, idx: u32) -> Option<ClientObject> {
self.clients().item(idx).map(|o| o.downcast().unwrap())
}
fn setup_clients(&self) { fn setup_clients(&self) {
let model = gio::ListStore::new::<ClientObject>(); let model = gio::ListStore::new::<ClientObject>();
self.imp().clients.replace(Some(model)); self.imp().clients.replace(Some(model));
@@ -62,27 +66,24 @@ impl Window {
let client_object = obj.downcast_ref().expect("Expected object of type `ClientObject`."); let client_object = obj.downcast_ref().expect("Expected object of type `ClientObject`.");
let row = window.create_client_row(client_object); let row = window.create_client_row(client_object);
row.connect_closure("request-update", false, closure_local!(@strong window => move |row: ClientRow, active: bool| { row.connect_closure("request-update", false, closure_local!(@strong window => move |row: ClientRow, active: bool| {
let index = row.index() as u32; if let Some(client) = window.client_by_idx(row.index() as u32) {
let Some(client) = window.clients().item(index) else { window.request_client_activate(&client, active);
return; window.request_client_update(&client);
}; window.request_client_state(&client);
let client = client.downcast_ref::<ClientObject>().unwrap(); }
window.request_client_activate(client, active);
window.request_client_update(client);
})); }));
row.connect_closure("request-delete", false, closure_local!(@strong window => move |row: ClientRow| { row.connect_closure("request-delete", false, closure_local!(@strong window => move |row: ClientRow| {
let index = row.index() as u32; if let Some(client) = window.client_by_idx(row.index() as u32) {
window.request_client_delete(index); window.request_client_delete(&client);
}
})); }));
row.connect_closure("request-dns", false, closure_local!(@strong window => move row.connect_closure("request-dns", false, closure_local!(@strong window => move
|row: ClientRow| { |row: ClientRow| {
let index = row.index() as u32; if let Some(client) = window.client_by_idx(row.index() as u32) {
let Some(client) = window.clients().item(index) else { window.request_client_update(&client);
return; window.request_dns(&client);
}; window.request_client_state(&client);
let client = client.downcast_ref::<ClientObject>().unwrap(); }
window.request_client_update(client);
window.request_dns(index);
})); }));
row.upcast() row.upcast()
}) })
@@ -204,20 +205,6 @@ impl Window {
} }
} }
pub fn request_dns(&self, idx: u32) {
let client_object = self.clients().item(idx).unwrap();
let client_object: &ClientObject = client_object.downcast_ref().unwrap();
let data = client_object.get_data();
let event = FrontendRequest::ResolveDns(data.handle);
self.request(event);
}
pub fn request_client_create(&self) {
let event = FrontendRequest::Create;
self.imp().set_port(DEFAULT_PORT);
self.request(event);
}
pub fn request_port_change(&self) { pub fn request_port_change(&self) {
let port = self.imp().port_entry.get().text().to_string(); let port = self.imp().port_entry.get().text().to_string();
if let Ok(port) = port.as_str().parse::<u16>() { if let Ok(port) = port.as_str().parse::<u16>() {
@@ -227,6 +214,23 @@ impl Window {
} }
} }
pub fn request_client_state(&self, client: &ClientObject) {
let handle = client.handle();
let event = FrontendRequest::GetState(handle);
self.request(event);
}
pub fn request_client_create(&self) {
let event = FrontendRequest::Create;
self.request(event);
}
pub fn request_dns(&self, client: &ClientObject) {
let data = client.get_data();
let event = FrontendRequest::ResolveDns(data.handle);
self.request(event);
}
pub fn request_client_update(&self, client: &ClientObject) { pub fn request_client_update(&self, client: &ClientObject) {
let handle = client.handle(); let handle = client.handle();
let data = client.get_data(); let data = client.get_data();
@@ -239,33 +243,25 @@ impl Window {
FrontendRequest::UpdatePosition(handle, position), FrontendRequest::UpdatePosition(handle, position),
FrontendRequest::UpdatePort(handle, port), FrontendRequest::UpdatePort(handle, port),
] { ] {
log::debug!("requesting: {event:?}");
self.request(event); self.request(event);
} }
} }
pub fn request_client_activate(&self, client: &ClientObject, active: bool) { pub fn request_client_activate(&self, client: &ClientObject, active: bool) {
let handle = client.handle(); let handle = client.handle();
let event = FrontendRequest::Activate(handle, active); let event = FrontendRequest::Activate(handle, active);
log::debug!("requesting: {event:?}");
self.request(event); self.request(event);
} }
pub fn request_client_delete(&self, idx: u32) { pub fn request_client_delete(&self, client: &ClientObject) {
if let Some(obj) = self.clients().item(idx) { let handle = client.handle();
let client_object: &ClientObject = obj let event = FrontendRequest::Delete(handle);
.downcast_ref() self.request(event);
.expect("Expected object of type `ClientObject`.");
let handle = client_object.handle();
let event = FrontendRequest::Delete(handle);
self.request(event);
}
} }
pub fn request(&self, event: FrontendRequest) { pub fn request(&self, event: FrontendRequest) {
let json = serde_json::to_string(&event).unwrap(); let json = serde_json::to_string(&event).unwrap();
log::debug!("requesting {json}"); log::debug!("requesting: {json}");
let mut stream = self.imp().stream.borrow_mut(); let mut stream = self.imp().stream.borrow_mut();
let stream = stream.as_mut().unwrap(); let stream = stream.as_mut().unwrap();
let bytes = json.as_bytes(); let bytes = json.as_bytes();

46
src/server/frontend_task.rs
View File

@@ -111,16 +111,17 @@ async fn handle_frontend_event(
} }
FrontendRequest::Activate(handle, active) => { FrontendRequest::Activate(handle, active) => {
if active { if active {
activate_client(server, frontend, capture, emulate, handle).await; activate_client(server, capture, emulate, handle).await;
} else { } else {
deactivate_client(server, frontend, capture, emulate, handle).await; deactivate_client(server, capture, emulate, handle).await;
} }
} }
FrontendRequest::ChangePort(port) => { FrontendRequest::ChangePort(port) => {
let _ = port_tx.send(port).await; let _ = port_tx.send(port).await;
} }
FrontendRequest::Delete(handle) => { FrontendRequest::Delete(handle) => {
remove_client(server, frontend, capture, emulate, handle).await; remove_client(server, capture, emulate, handle).await;
broadcast(frontend, FrontendEvent::Deleted(handle)).await;
} }
FrontendRequest::Enumerate() => { FrontendRequest::Enumerate() => {
let clients = server let clients = server
@@ -131,25 +132,24 @@ async fn handle_frontend_event(
.collect(); .collect();
broadcast(frontend, FrontendEvent::Enumerate(clients)).await; broadcast(frontend, FrontendEvent::Enumerate(clients)).await;
} }
FrontendRequest::GetState(handle) => {
broadcast_client(server, frontend, handle).await;
}
FrontendRequest::Terminate() => { FrontendRequest::Terminate() => {
log::info!("terminating gracefully..."); log::info!("terminating gracefully...");
return true; return true;
} }
FrontendRequest::UpdateFixIps(handle, fix_ips) => { FrontendRequest::UpdateFixIps(handle, fix_ips) => {
update_fix_ips(server, resolve_tx, handle, fix_ips).await; update_fix_ips(server, resolve_tx, handle, fix_ips).await;
broadcast_client_update(server, frontend, handle).await;
} }
FrontendRequest::UpdateHostname(handle, hostname) => { FrontendRequest::UpdateHostname(handle, hostname) => {
update_hostname(server, resolve_tx, handle, hostname).await; update_hostname(server, resolve_tx, handle, hostname).await;
broadcast_client_update(server, frontend, handle).await;
} }
FrontendRequest::UpdatePort(handle, port) => { FrontendRequest::UpdatePort(handle, port) => {
update_port(server, handle, port).await; update_port(server, handle, port).await;
broadcast_client_update(server, frontend, handle).await;
} }
FrontendRequest::UpdatePosition(handle, pos) => { FrontendRequest::UpdatePosition(handle, pos) => {
update_pos(server, handle, capture, emulate, pos).await; update_pos(server, handle, capture, emulate, pos).await;
broadcast_client_update(server, frontend, handle).await;
} }
FrontendRequest::ResolveDns(handle) => { FrontendRequest::ResolveDns(handle) => {
let hostname = server let hostname = server
@@ -181,15 +181,13 @@ pub async fn add_client(server: &Server, frontend: &mut FrontendListener) {
pub async fn deactivate_client( pub async fn deactivate_client(
server: &Server, server: &Server,
frontend: &mut FrontendListener,
capture: &Sender<CaptureEvent>, capture: &Sender<CaptureEvent>,
emulate: &Sender<EmulationEvent>, emulate: &Sender<EmulationEvent>,
handle: ClientHandle, handle: ClientHandle,
) { ) {
let state = match server.client_manager.borrow_mut().get_mut(handle) { match server.client_manager.borrow_mut().get_mut(handle) {
Some((_, s)) => { Some((_, s)) => {
s.active = false; s.active = false;
s.clone()
} }
None => return, None => return,
}; };
@@ -197,13 +195,10 @@ pub async fn deactivate_client(
let event = ClientEvent::Destroy(handle); let event = ClientEvent::Destroy(handle);
let _ = capture.send(CaptureEvent::ClientEvent(event)).await; let _ = capture.send(CaptureEvent::ClientEvent(event)).await;
let _ = emulate.send(EmulationEvent::ClientEvent(event)).await; let _ = emulate.send(EmulationEvent::ClientEvent(event)).await;
let event = FrontendEvent::StateChange(handle, state);
broadcast(frontend, event).await;
} }
pub async fn activate_client( pub async fn activate_client(
server: &Server, server: &Server,
frontend: &mut FrontendListener,
capture: &Sender<CaptureEvent>, capture: &Sender<CaptureEvent>,
emulate: &Sender<EmulationEvent>, emulate: &Sender<EmulationEvent>,
handle: ClientHandle, handle: ClientHandle,
@@ -217,14 +212,13 @@ pub async fn activate_client(
let other = server.client_manager.borrow_mut().find_client(pos); let other = server.client_manager.borrow_mut().find_client(pos);
if let Some(other) = other { if let Some(other) = other {
if other != handle { if other != handle {
deactivate_client(server, frontend, capture, emulate, other).await; deactivate_client(server, capture, emulate, other).await;
} }
} }
/* activate the client */ /* activate the client */
let state = if let Some((_, s)) = server.client_manager.borrow_mut().get_mut(handle) { if let Some((_, s)) = server.client_manager.borrow_mut().get_mut(handle) {
s.active = true; s.active = true;
s.clone()
} else { } else {
return; return;
}; };
@@ -233,13 +227,10 @@ pub async fn activate_client(
let event = ClientEvent::Create(handle, pos); let event = ClientEvent::Create(handle, pos);
let _ = capture.send(CaptureEvent::ClientEvent(event)).await; let _ = capture.send(CaptureEvent::ClientEvent(event)).await;
let _ = emulate.send(EmulationEvent::ClientEvent(event)).await; let _ = emulate.send(EmulationEvent::ClientEvent(event)).await;
let event = FrontendEvent::StateChange(handle, state);
broadcast(frontend, event).await;
} }
pub async fn remove_client( pub async fn remove_client(
server: &Server, server: &Server,
frontend: &mut FrontendListener,
capture: &Sender<CaptureEvent>, capture: &Sender<CaptureEvent>,
emulate: &Sender<EmulationEvent>, emulate: &Sender<EmulationEvent>,
handle: ClientHandle, handle: ClientHandle,
@@ -258,9 +249,6 @@ pub async fn remove_client(
let _ = capture.send(CaptureEvent::ClientEvent(destroy)).await; let _ = capture.send(CaptureEvent::ClientEvent(destroy)).await;
let _ = emulate.send(EmulationEvent::ClientEvent(destroy)).await; let _ = emulate.send(EmulationEvent::ClientEvent(destroy)).await;
} }
let event = FrontendEvent::Deleted(handle);
broadcast(frontend, event).await;
} }
async fn update_fix_ips( async fn update_fix_ips(
@@ -356,11 +344,11 @@ async fn update_pos(
} }
} }
async fn broadcast_client_update( async fn broadcast_client(server: &Server, frontend: &mut FrontendListener, handle: ClientHandle) {
server: &Server, let client = server.client_manager.borrow().get(handle).cloned();
frontend: &mut FrontendListener, if let Some((config, state)) = client {
handle: ClientHandle, broadcast(frontend, FrontendEvent::State(handle, config, state)).await;
) { } else {
let (client, _) = server.client_manager.borrow().get(handle).unwrap().clone(); broadcast(frontend, FrontendEvent::NoSuchClient(handle)).await;
broadcast(frontend, FrontendEvent::Updated(handle, client)).await; }
} }

12
src/server/resolver_task.rs
View File

@@ -35,7 +35,7 @@ pub fn new(
Ok(ips) => ips, Ok(ips) => ips,
Err(e) => { Err(e) => {
log::warn!("could not resolve host '{host}': {e}"); log::warn!("could not resolve host '{host}': {e}");
continue; vec![]
} }
}; };
@@ -59,14 +59,10 @@ async fn notify_state_change(
server: &mut Server, server: &mut Server,
handle: ClientHandle, handle: ClientHandle,
) { ) {
let state = server let state = server.client_manager.borrow_mut().get_mut(handle).cloned();
.client_manager if let Some((config, state)) = state {
.borrow_mut()
.get_mut(handle)
.map(|(_, s)| s.clone());
if let Some(state) = state {
let _ = frontend let _ = frontend
.send(FrontendEvent::StateChange(handle, state)) .send(FrontendEvent::State(handle, config, state))
.await; .await;
} }
} }