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

move lan-mouse protocol to separate crate (#178)

Browse Source
This commit is contained in:
Ferdinand Schober
2024-08-11 16:51:47 +02:00
committed by GitHub
parent fe06ca1fae
commit 19c2c4327f
28 changed files with 451 additions and 545 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/server/capture_task.rs
View File

@@ -1,4 +1,5 @@
use futures::StreamExt;
use lan_mouse_proto::ProtoEvent;
use std::net::SocketAddr;
use tokio::{
@@ -7,9 +8,9 @@ use tokio::{
task::JoinHandle,
};
use input_capture::{self, CaptureError, CaptureHandle, InputCapture, InputCaptureError, Position};
use input_event::Event;
use input_capture::{
self, CaptureError, CaptureEvent, CaptureHandle, InputCapture, InputCaptureError, Position,
};
use crate::{client::ClientHandle, frontend::Status, server::State};
@@ -28,7 +29,7 @@ pub(crate) enum CaptureRequest {
pub(crate) fn new(
server: Server,
capture_rx: Receiver<CaptureRequest>,
udp_send: Sender<(Event, SocketAddr)>,
udp_send: Sender<(ProtoEvent, SocketAddr)>,
) -> JoinHandle<()> {
let backend = server.config.capture_backend.map(|b| b.into());
tokio::task::spawn_local(capture_task(server, backend, udp_send, capture_rx))
@@ -37,7 +38,7 @@ pub(crate) fn new(
async fn capture_task(
server: Server,
backend: Option<input_capture::Backend>,
sender_tx: Sender<(Event, SocketAddr)>,
sender_tx: Sender<(ProtoEvent, SocketAddr)>,
mut notify_rx: Receiver<CaptureRequest>,
) {
loop {
@@ -63,7 +64,7 @@ async fn capture_task(
async fn do_capture(
backend: Option<input_capture::Backend>,
server: &Server,
sender_tx: &Sender<(Event, SocketAddr)>,
sender_tx: &Sender<(ProtoEvent, SocketAddr)>,
notify_rx: &mut Receiver<CaptureRequest>,
) -> Result<(), InputCaptureError> {
/* allow cancelling capture request */
@@ -120,17 +121,20 @@ async fn do_capture(
async fn handle_capture_event(
server: &Server,
capture: &mut InputCapture,
sender_tx: &Sender<(Event, SocketAddr)>,
event: (CaptureHandle, Event),
sender_tx: &Sender<(ProtoEvent, SocketAddr)>,
event: (CaptureHandle, CaptureEvent),
) -> Result<(), CaptureError> {
let (handle, event) = event;
log::trace!("({handle}) {event:?}");
// capture started
if event == Event::Enter() {
server.set_state(State::AwaitingLeave);
if event == CaptureEvent::Begin {
// wait for remote to acknowlegde enter
server.set_state(State::AwaitAck);
server.set_active(Some(handle));
// restart ping timer to release capture if unreachable
server.restart_ping_timer();
// spawn enter hook cmd
spawn_hook_command(server, handle);
}
@@ -148,14 +152,18 @@ async fn handle_capture_event(
if let Some(addr) = server.active_addr(handle) {
let event = match server.get_state() {
State::Sending => event,
State::Sending => match event {
CaptureEvent::Begin => ProtoEvent::Enter(0),
CaptureEvent::Input(e) => ProtoEvent::Input(e),
},
/* send additional enter events until acknowleged */
State::AwaitingLeave => Event::Enter(),
State::AwaitAck => ProtoEvent::Enter(0),
/* released capture */
State::Receiving => Event::Disconnect(),
State::Receiving => ProtoEvent::Leave(0),
};
sender_tx.send((event, addr)).await.expect("sender closed");
}
};
Ok(())
}

107
src/server/emulation_task.rs
View File

@@ -1,5 +1,6 @@
use std::net::SocketAddr;
use lan_mouse_proto::ProtoEvent;
use tokio::{
sync::mpsc::{Receiver, Sender},
task::JoinHandle,
@@ -11,9 +12,8 @@ use crate::{
server::State,
};
use input_emulation::{self, EmulationError, EmulationHandle, InputEmulation, InputEmulationError};
use input_event::Event;
use super::{network_task::NetworkError, CaptureRequest, Server};
use super::{network_task::NetworkError, Server};
#[derive(Clone, Debug)]
pub(crate) enum EmulationRequest {
@@ -28,23 +28,21 @@ pub(crate) enum EmulationRequest {
pub(crate) fn new(
server: Server,
emulation_rx: Receiver<EmulationRequest>,
udp_rx: Receiver<Result<(Event, SocketAddr), NetworkError>>,
sender_tx: Sender<(Event, SocketAddr)>,
capture_tx: Sender<CaptureRequest>,
udp_rx: Receiver<Result<(ProtoEvent, SocketAddr), NetworkError>>,
sender_tx: Sender<(ProtoEvent, SocketAddr)>,
) -> JoinHandle<()> {
let emulation_task = emulation_task(server, emulation_rx, udp_rx, sender_tx, capture_tx);
let emulation_task = emulation_task(server, emulation_rx, udp_rx, sender_tx);
tokio::task::spawn_local(emulation_task)
}
async fn emulation_task(
server: Server,
mut rx: Receiver<EmulationRequest>,
mut udp_rx: Receiver<Result<(Event, SocketAddr), NetworkError>>,
sender_tx: Sender<(Event, SocketAddr)>,
capture_tx: Sender<CaptureRequest>,
mut udp_rx: Receiver<Result<(ProtoEvent, SocketAddr), NetworkError>>,
sender_tx: Sender<(ProtoEvent, SocketAddr)>,
) {
loop {
if let Err(e) = do_emulation(&server, &mut rx, &mut udp_rx, &sender_tx, &capture_tx).await {
if let Err(e) = do_emulation(&server, &mut rx, &mut udp_rx, &sender_tx).await {
log::warn!("input emulation exited: {e}");
}
server.set_emulation_status(Status::Disabled);
@@ -66,9 +64,8 @@ async fn emulation_task(
async fn do_emulation(
server: &Server,
rx: &mut Receiver<EmulationRequest>,
udp_rx: &mut Receiver<Result<(Event, SocketAddr), NetworkError>>,
sender_tx: &Sender<(Event, SocketAddr)>,
capture_tx: &Sender<CaptureRequest>,
udp_rx: &mut Receiver<Result<(ProtoEvent, SocketAddr), NetworkError>>,
sender_tx: &Sender<(ProtoEvent, SocketAddr)>,
) -> Result<(), InputEmulationError> {
let backend = server.config.emulation_backend.map(|b| b.into());
log::info!("creating input emulation...");
@@ -84,7 +81,7 @@ async fn do_emulation(
emulation.create(handle).await;
}
let res = do_emulation_session(server, &mut emulation, rx, udp_rx, sender_tx, capture_tx).await;
let res = do_emulation_session(server, &mut emulation, rx, udp_rx, sender_tx).await;
emulation.terminate().await; // manual drop
res
}
@@ -93,9 +90,8 @@ async fn do_emulation_session(
server: &Server,
emulation: &mut InputEmulation,
rx: &mut Receiver<EmulationRequest>,
udp_rx: &mut Receiver<Result<(Event, SocketAddr), NetworkError>>,
sender_tx: &Sender<(Event, SocketAddr)>,
capture_tx: &Sender<CaptureRequest>,
udp_rx: &mut Receiver<Result<(ProtoEvent, SocketAddr), NetworkError>>,
sender_tx: &Sender<(ProtoEvent, SocketAddr)>,
) -> Result<(), InputEmulationError> {
let mut last_ignored = None;
@@ -109,7 +105,7 @@ async fn do_emulation_session(
continue;
}
};
handle_udp_rx(server, capture_tx, emulation, sender_tx, &mut last_ignored, udp_event).await?;
handle_incoming_event(server, emulation, sender_tx, &mut last_ignored, udp_event).await?;
}
emulate_event = rx.recv() => {
match emulate_event.expect("channel closed") {
@@ -123,13 +119,12 @@ async fn do_emulation_session(
}
}
async fn handle_udp_rx(
async fn handle_incoming_event(
server: &Server,
capture_tx: &Sender<CaptureRequest>,
emulate: &mut InputEmulation,
sender_tx: &Sender<(Event, SocketAddr)>,
sender_tx: &Sender<(ProtoEvent, SocketAddr)>,
last_ignored: &mut Option<SocketAddr>,
event: (Event, SocketAddr),
event: (ProtoEvent, SocketAddr),
) -> Result<(), EmulationError> {
let (event, addr) = event;
@@ -143,55 +138,27 @@ async fn handle_udp_rx(
};
match (event, addr) {
(Event::Pong(), _) => { /* ignore pong events */ }
(Event::Ping(), addr) => {
let _ = sender_tx.send((Event::Pong(), addr)).await;
(ProtoEvent::Pong, _) => { /* ignore pong events */ }
(ProtoEvent::Ping, addr) => {
let _ = sender_tx.send((ProtoEvent::Pong, addr)).await;
}
(Event::Disconnect(), _) => emulate.release_keys(handle).await?,
(event, addr) => {
// tell clients that we are ready to receive events
if let Event::Enter() = event {
let _ = sender_tx.send((Event::Leave(), addr)).await;
}
match server.state.get() {
State::Sending => {
if let Event::Leave() = event {
// ignore additional leave events that may
// have been sent for redundancy
} else {
// upon receiving any event, we go back to receiving mode
server.state.replace(State::Receiving);
let _ = capture_tx.send(CaptureRequest::Release).await;
log::trace!("STATE ===> Receiving");
}
}
State::Receiving => {
log::trace!("{event} => emulate");
emulate.consume(event, handle).await?;
let has_pressed_keys = emulate.has_pressed_keys(handle);
server.update_pressed_keys(handle, has_pressed_keys);
if has_pressed_keys {
server.restart_ping_timer();
}
}
State::AwaitingLeave => {
// we just entered the deadzone of a client, so
// we need to ignore events that may still
// be on the way until a leave event occurs
// telling us the client registered the enter
if let Event::Leave() = event {
server.state.replace(State::Sending);
log::trace!("STATE ===> Sending");
}
// entering a client that is waiting for a leave
// event should still be possible
if let Event::Enter() = event {
server.state.replace(State::Receiving);
let _ = capture_tx.send(CaptureRequest::Release).await;
log::trace!("STATE ===> Receiving");
}
(ProtoEvent::Leave(_), _) => emulate.release_keys(handle).await?,
(ProtoEvent::Ack(_), _) => server.set_state(State::Sending),
(ProtoEvent::Enter(_), _) => {
server.set_state(State::Receiving);
sender_tx
.send((ProtoEvent::Ack(0), addr))
.await
.expect("no channel")
}
(ProtoEvent::Input(e), _) => {
if let State::Receiving = server.get_state() {
log::trace!("{event} => emulate");
emulate.consume(e, handle).await?;
let has_pressed_keys = emulate.has_pressed_keys(handle);
server.update_pressed_keys(handle, has_pressed_keys);
if has_pressed_keys {
server.restart_ping_timer();
}
}
}

26
src/server/network_task.rs
View File

@@ -7,14 +7,13 @@ use tokio::{
task::JoinHandle,
};
use input_event::{Event, ProtocolError};
use super::Server;
use lan_mouse_proto::{ProtoEvent, ProtocolError};
pub(crate) async fn new(
server: Server,
udp_recv_tx: Sender<Result<(Event, SocketAddr), NetworkError>>,
udp_send_rx: Receiver<(Event, SocketAddr)>,
udp_recv_tx: Sender<Result<(ProtoEvent, SocketAddr), NetworkError>>,
udp_send_rx: Receiver<(ProtoEvent, SocketAddr)>,
) -> io::Result<JoinHandle<()>> {
// bind the udp socket
let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), server.port.get());
@@ -62,7 +61,7 @@ async fn update_port(server: &Server, socket: &mut UdpSocket) {
async fn udp_receiver(
socket: &UdpSocket,
receiver_tx: &Sender<Result<(Event, SocketAddr), NetworkError>>,
receiver_tx: &Sender<Result<(ProtoEvent, SocketAddr), NetworkError>>,
) {
loop {
let event = receive_event(socket).await;
@@ -70,7 +69,7 @@ async fn udp_receiver(
}
}
async fn udp_sender(socket: &UdpSocket, rx: &mut Receiver<(Event, SocketAddr)>) {
async fn udp_sender(socket: &UdpSocket, rx: &mut Receiver<(ProtoEvent, SocketAddr)>) {
loop {
let (event, addr) = rx.recv().await.expect("channel closed");
if let Err(e) = send_event(socket, event, addr) {
@@ -87,16 +86,17 @@ pub(crate) enum NetworkError {
Io(#[from] io::Error),
}
async fn receive_event(socket: &UdpSocket) -> Result<(Event, SocketAddr), NetworkError> {
let mut buf = vec![0u8; 22];
let (_amt, src) = socket.recv_from(&mut buf).await?;
Ok((Event::try_from(buf)?, src))
async fn receive_event(socket: &UdpSocket) -> Result<(ProtoEvent, SocketAddr), NetworkError> {
let mut buf = [0u8; lan_mouse_proto::MAX_EVENT_SIZE];
let (_len, src) = socket.recv_from(&mut buf).await?;
let event = ProtoEvent::try_from(buf)?;
Ok((event, src))
}
fn send_event(sock: &UdpSocket, e: Event, addr: SocketAddr) -> Result<usize, NetworkError> {
fn send_event(sock: &UdpSocket, e: ProtoEvent, addr: SocketAddr) -> Result<usize, NetworkError> {
log::trace!("{:20} ------>->->-> {addr}", e.to_string());
let data: Vec<u8> = (&e).into();
let (data, len): ([u8; lan_mouse_proto::MAX_EVENT_SIZE], usize) = e.into();
// When udp blocks, we dont want to block the event loop.
// Dropping events is better than potentially crashing the input capture.
Ok(sock.try_send_to(&data, addr)?)
Ok(sock.try_send_to(&data[..len], addr)?)
}

9
src/server/ping_task.rs
View File

@@ -1,9 +1,8 @@
use std::{net::SocketAddr, time::Duration};
use lan_mouse_proto::ProtoEvent;
use tokio::{sync::mpsc::Sender, task::JoinHandle};
use input_event::Event;
use crate::client::ClientHandle;
use super::{capture_task::CaptureRequest, emulation_task::EmulationRequest, Server, State};
@@ -12,7 +11,7 @@ const MAX_RESPONSE_TIME: Duration = Duration::from_millis(500);
pub(crate) fn new(
server: Server,
sender_ch: Sender<(Event, SocketAddr)>,
sender_ch: Sender<(ProtoEvent, SocketAddr)>,
emulate_notify: Sender<EmulationRequest>,
capture_notify: Sender<CaptureRequest>,
) -> JoinHandle<()> {
@@ -27,7 +26,7 @@ pub(crate) fn new(
async fn ping_task(
server: &Server,
sender_ch: Sender<(Event, SocketAddr)>,
sender_ch: Sender<(ProtoEvent, SocketAddr)>,
emulate_notify: Sender<EmulationRequest>,
capture_notify: Sender<CaptureRequest>,
) {
@@ -86,7 +85,7 @@ async fn ping_task(
// ping clients
for addr in ping_addrs {
if sender_ch.send((Event::Ping(), addr)).await.is_err() {
if sender_ch.send((ProtoEvent::Ping, addr)).await.is_err() {
break;
}
}