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

formatting

Browse Source
This commit is contained in:
Ferdinand Schober
2023-12-09 00:43:54 +01:00
parent 56e5f7a30d
commit 5a7e0cf89c
26 changed files with 881 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
src/server.rs
View File

@@ -1,7 +1,18 @@
use std::{error::Error, io::Result, collections::HashSet, time::{Duration, Instant}, net::IpAddr};
use log;
use tokio::{net::UdpSocket, io::ReadHalf, signal, sync::mpsc::{Sender, Receiver}};
use futures::stream::StreamExt;
use log;
use std::{
collections::HashSet,
error::Error,
io::Result,
net::IpAddr,
time::{Duration, Instant},
};
use tokio::{
io::ReadHalf,
net::UdpSocket,
signal,
sync::mpsc::{Receiver, Sender},
};
#[cfg(unix)]
use tokio::net::UnixStream;
@@ -9,10 +20,17 @@ use tokio::net::UnixStream;
#[cfg(windows)]
use tokio::net::TcpStream;
use std::{net::SocketAddr, io::ErrorKind};
use std::{io::ErrorKind, net::SocketAddr};
use crate::{client::{ClientEvent, ClientManager, Position, ClientHandle}, consumer::EventConsumer, producer::EventProducer, frontend::{FrontendEvent, FrontendListener, FrontendNotify, self}, dns::{self, DnsResolver}, config::Config};
use crate::event::Event;
use crate::{
client::{ClientEvent, ClientHandle, ClientManager, Position},
config::Config,
consumer::EventConsumer,
dns::{self, DnsResolver},
frontend::{self, FrontendEvent, FrontendListener, FrontendNotify},
producer::EventProducer,
};
/// keeps track of state to prevent a feedback loop
/// of continuously sending and receiving the same event.
@@ -41,7 +59,6 @@ impl Server {
consumer: Box<dyn EventConsumer>,
producer: Box<dyn EventProducer>,
) -> anyhow::Result<Self> {
// create dns resolver
let resolver = dns::DnsResolver::new().await?;
@@ -65,7 +82,7 @@ impl Server {
};
// add clients from config
for (c,h,port,p) in config.get_clients().into_iter() {
for (c, h, port, p) in config.get_clients().into_iter() {
server.add_client(h, c, port, p).await;
}
@@ -73,7 +90,6 @@ impl Server {
}
pub async fn run(&mut self) -> anyhow::Result<()> {
loop {
log::trace!("polling ...");
tokio::select! {
@@ -127,14 +143,20 @@ impl Server {
}
}
}
// destroy consumer
self.consumer.destroy().await;
Ok(())
}
pub async fn add_client(&mut self, hostname: Option<String>, mut addr: HashSet<IpAddr>, port: u16, pos: Position) -> ClientHandle {
pub async fn add_client(
&mut self,
hostname: Option<String>,
mut addr: HashSet<IpAddr>,
port: u16,
pos: Position,
) -> ClientHandle {
let ips = if let Some(hostname) = hostname.as_ref() {
match self.resolver.resolve(hostname.as_str()).await {
Ok(ips) => HashSet::from_iter(ips.iter().cloned()),
@@ -147,8 +169,15 @@ impl Server {
HashSet::new()
};
addr.extend(ips.iter());
log::info!("adding client [{}]{} @ {:?}", pos, hostname.as_deref().unwrap_or(""), &ips);
let client = self.client_manager.add_client(hostname.clone(), addr, port, pos);
log::info!(
"adding client [{}]{} @ {:?}",
pos,
hostname.as_deref().unwrap_or(""),
&ips
);
let client = self
.client_manager
.add_client(hostname.clone(), addr, port, pos);
log::debug!("add_client {client}");
let notify = FrontendNotify::NotifyClientCreate(client, hostname, port, pos);
if let Err(e) = self.frontend.notify_all(notify).await {
@@ -161,8 +190,11 @@ impl Server {
if let Some(state) = self.client_manager.get_mut(client) {
state.active = active;
if state.active {
self.producer.notify(ClientEvent::Create(client, state.client.pos));
self.consumer.notify(ClientEvent::Create(client, state.client.pos)).await;
self.producer
.notify(ClientEvent::Create(client, state.client.pos));
self.consumer
.notify(ClientEvent::Create(client, state.client.pos))
.await;
} else {
self.producer.notify(ClientEvent::Destroy(client));
self.consumer.notify(ClientEvent::Destroy(client)).await;
@@ -173,7 +205,11 @@ impl Server {
pub async fn remove_client(&mut self, client: ClientHandle) -> Option<ClientHandle> {
self.producer.notify(ClientEvent::Destroy(client));
self.consumer.notify(ClientEvent::Destroy(client)).await;
if let Some(client) = self.client_manager.remove_client(client).map(|s| s.client.handle) {
if let Some(client) = self
.client_manager
.remove_client(client)
.map(|s| s.client.handle)
{
let notify = FrontendNotify::NotifyClientDelete(client);
log::debug!("{notify:?}");
if let Err(e) = self.frontend.notify_all(notify).await {
@@ -194,7 +230,7 @@ impl Server {
) {
// retrieve state
let Some(state) = self.client_manager.get_mut(client) else {
return
return;
};
// update pos
@@ -209,12 +245,20 @@ impl Server {
// update port
if state.client.port != port {
state.client.port = port;
state.client.addrs = state.client.addrs
state.client.addrs = state
.client
.addrs
.iter()
.cloned()
.map(|mut a| { a.set_port(port); a })
.map(|mut a| {
a.set_port(port);
a
})
.collect();
state.client.active_addr.map(|a| { SocketAddr::new(a.ip(), port) });
state
.client
.active_addr
.map(|a| SocketAddr::new(a.ip(), port));
}
// update hostname
@@ -258,12 +302,12 @@ impl Server {
}
};
// reset ttl for client and
// reset ttl for client and
state.last_seen = Some(Instant::now());
// set addr as new default for this client
state.client.active_addr = Some(addr);
match (event, addr) {
(Event::Pong(), _) => {},
(Event::Pong(), _) => {}
(Event::Ping(), addr) => {
if let Err(e) = send_event(&self.socket, Event::Pong(), addr).await {
log::error!("udp send: {}", e);
@@ -291,16 +335,17 @@ impl Server {
// let the server know we are still alive once every second
let last_replied = state.last_replied;
if last_replied.is_none()
|| last_replied.is_some()
&& last_replied.unwrap().elapsed() > Duration::from_secs(1) {
if last_replied.is_none()
|| last_replied.is_some()
&& last_replied.unwrap().elapsed() > Duration::from_secs(1)
{
state.last_replied = Some(Instant::now());
if let Err(e) = send_event(&self.socket, Event::Pong(), addr).await {
log::error!("udp send: {}", e);
}
}
}
}
},
}
}
@@ -317,7 +362,7 @@ impl Server {
Some(state) => state,
None => {
log::warn!("unknown client!");
return
return;
}
};
// otherwise we should have an address to send to
@@ -331,20 +376,21 @@ impl Server {
// if client last responded > 2 seconds ago
// and we have not sent a ping since 500 milliseconds,
// send a ping
if state.last_seen.is_some()
&& state.last_seen.unwrap().elapsed() < Duration::from_secs(2) {
return
if state.last_seen.is_some() && state.last_seen.unwrap().elapsed() < Duration::from_secs(2)
{
return;
}
// client last seen > 500ms ago
if state.last_ping.is_some()
&& state.last_ping.unwrap().elapsed() < Duration::from_millis(500) {
return
&& state.last_ping.unwrap().elapsed() < Duration::from_millis(500)
{
return;
}
// release mouse if client didnt respond to the first ping
if state.last_ping.is_some()
&& state.last_ping.unwrap().elapsed() < Duration::from_secs(1) {
if state.last_ping.is_some() && state.last_ping.unwrap().elapsed() < Duration::from_secs(1)
{
should_release = true;
}
@@ -414,12 +460,20 @@ impl Server {
async fn handle_frontend_event(&mut self, event: FrontendEvent) -> bool {
log::debug!("frontend: {event:?}");
match event {
FrontendEvent::AddClient(hostname, port, pos) => { self.add_client(hostname, HashSet::new(), port, pos).await; },
FrontendEvent::ActivateClient(client, active) => self.activate_client(client, active).await,
FrontendEvent::AddClient(hostname, port, pos) => {
self.add_client(hostname, HashSet::new(), port, pos).await;
}
FrontendEvent::ActivateClient(client, active) => {
self.activate_client(client, active).await
}
FrontendEvent::ChangePort(port) => {
let current_port = self.socket.local_addr().unwrap().port();
if current_port == port {
if let Err(e) = self.frontend.notify_all(FrontendNotify::NotifyPortChange(port, None)).await {
if let Err(e) = self
.frontend
.notify_all(FrontendNotify::NotifyPortChange(port, None))
.await
{
log::warn!("error notifying frontend: {e}");
}
return false;
@@ -428,39 +482,60 @@ impl Server {
match UdpSocket::bind(listen_addr).await {
Ok(socket) => {
self.socket = socket;
if let Err(e) = self.frontend.notify_all(FrontendNotify::NotifyPortChange(port, None)).await {
if let Err(e) = self
.frontend
.notify_all(FrontendNotify::NotifyPortChange(port, None))
.await
{
log::warn!("error notifying frontend: {e}");
}
},
}
Err(e) => {
log::warn!("could not change port: {e}");
let port = self.socket.local_addr().unwrap().port();
if let Err(e) = self.frontend.notify_all(FrontendNotify::NotifyPortChange(port, Some(format!("could not change port: {e}")))).await {
if let Err(e) = self
.frontend
.notify_all(FrontendNotify::NotifyPortChange(
port,
Some(format!("could not change port: {e}")),
))
.await
{
log::error!("error notifying frontend: {e}");
}
}
}
},
FrontendEvent::DelClient(client) => { self.remove_client(client).await; },
}
FrontendEvent::DelClient(client) => {
self.remove_client(client).await;
}
FrontendEvent::Enumerate() => self.enumerate().await,
FrontendEvent::Shutdown() => {
log::info!("terminating gracefully...");
return true;
},
FrontendEvent::UpdateClient(client, hostname, port, pos) => self.update_client(client, hostname, port, pos).await,
}
FrontendEvent::UpdateClient(client, hostname, port, pos) => {
self.update_client(client, hostname, port, pos).await
}
}
false
}
async fn enumerate(&mut self) {
let clients = self.client_manager.enumerate();
if let Err(e) = self.frontend.notify_all(FrontendNotify::Enumerate(clients)).await {
if let Err(e) = self
.frontend
.notify_all(FrontendNotify::Enumerate(clients))
.await
{
log::error!("error notifying frontend: {e}");
}
}
}
async fn receive_event(socket: &UdpSocket) -> std::result::Result<(Event, SocketAddr), Box<dyn Error>> {
async fn receive_event(
socket: &UdpSocket,
) -> std::result::Result<(Event, SocketAddr), Box<dyn Error>> {
log::trace!("receive_event");
let mut buf = vec![0u8; 22];
match socket.recv_from(&mut buf).await {
@@ -469,7 +544,6 @@ async fn receive_event(socket: &UdpSocket) -> std::result::Result<(Event, Socket
}
}
async fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize> {
log::trace!("{:20} ------>->->-> {addr}", e.to_string());
let data: Vec<u8> = (&e).into();
@@ -477,4 +551,3 @@ async fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usiz
// It may be better to set the socket to non-blocking and only send when ready.
sock.send_to(&data[..], addr).await
}