From 22c98311a8e8f4a1c94b0e4ac0f36b4e016a2bf4 Mon Sep 17 00:00:00 2001 From: littleblack111 Date: Sun, 21 Sep 2025 23:08:41 +0800 Subject: format --- src/server/main.rs | 8 ++- src/server/server/tcpserver.rs | 130 +++++++++++++++++++++++++++++++++-------- src/server/tasks.rs | 19 +++++- 3 files changed, 130 insertions(+), 27 deletions(-) (limited to 'src/server') diff --git a/src/server/main.rs b/src/server/main.rs index 61ab48c..3ecd17b 100644 --- a/src/server/main.rs +++ b/src/server/main.rs @@ -7,5 +7,11 @@ pub(crate) mod tasks; #[tokio::main] async fn main() { - TCPServer::new(TcpListener::bind("0.0.0.0:8080").await.unwrap()).listen().await; + TCPServer::new( + TcpListener::bind("0.0.0.0:8080") + .await + .unwrap(), + ) + .listen() + .await; } diff --git a/src/server/server/tcpserver.rs b/src/server/server/tcpserver.rs index 118560d..d4f1b1a 100644 --- a/src/server/server/tcpserver.rs +++ b/src/server/server/tcpserver.rs @@ -33,13 +33,23 @@ pub struct TCPServer<'a> { impl<'a> TCPServer<'a> { pub fn new(listener: TcpListener) -> Arc { - let s = Arc::new(Self { this: RefCell::new(Weak::new()), listener, task: None, clients: RefCell::new(Vec::new()) }); - *s.this.borrow_mut() = Arc::downgrade(&s); + let s = Arc::new( + Self { + this: RefCell::new(Weak::new()), + listener, + task: None, + clients: RefCell::new(Vec::new()), + }, + ); + *s.this + .borrow_mut() = Arc::downgrade(&s); s } fn remove(&self, client: &Client<'a>) { - self.clients.borrow_mut().retain(|c| c != client); + self.clients + .borrow_mut() + .retain(|c| c != client); } } @@ -50,8 +60,24 @@ impl<'a> Server<'a> for TCPServer<'a> { async fn listen(&self) -> ! { loop { - match self.listener.accept().await { - Ok((socket, addr)) => self.clients.borrow_mut().push(Client::new(self.this.borrow().upgrade().unwrap(), socket, addr)), // the unwrap shouldn't panics, self has to be alive + match self + .listener + .accept() + .await + { + Ok((socket, addr)) => self + .clients + .borrow_mut() + .push( + Client::new( + self.this + .borrow() + .upgrade() + .unwrap(), + socket, + addr, + ), + ), // the unwrap shouldn't panics, self has to be alive Err(e) => eprintln!("Failed to accept connection: {e}"), }; } @@ -67,12 +93,19 @@ struct Client<'a> { impl<'a> Client<'a> { fn new(server: Arc>, stream: TcpStream, addr: SocketAddr) -> Self { - Self { server, stream, addr, data: None } + Self { + server, + stream, + addr, + data: None, + } } fn handle_read(&mut self, buf: io::Result) -> bool { match buf { - Ok(0) => self.server.remove(self), + Ok(0) => self + .server + .remove(self), Ok(_) => {} Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return false, Err(e) => { @@ -84,34 +117,64 @@ impl<'a> Client<'a> { } // TODO: use framed - async fn read(&mut self) -> Option<(Vec, usize)> { + async fn read( + &mut self, + ) -> Option<( + Vec, + usize, + )> { let mut len = vec![0; std::mem::size_of::()]; // vec! only take usize - if !self.handle_read(self.stream.try_read(&mut len)) { + if !self.handle_read( + self.stream + .try_read(&mut len), + ) { return None; } - let len = usize::from_be_bytes(match len.try_into() { - Ok(n) => n, - Err(e) => { - eprintln!("Failed to convert length bytes to length usize: {:?}", e); - return None; - } - }); // TODO: move to a shared fn with client to ensure we use same endian + let len = usize::from_be_bytes( + match len.try_into() { + Ok(n) => n, + Err(e) => { + eprintln!( + "Failed to convert length bytes to length usize: {:?}", + e + ); + return None; + } + }, + ); // TODO: move to a shared fn with client to ensure we use same endian let mut buf = vec![0; len]; - let read = self.stream.read_exact(&mut buf).await; + let read = self + .stream + .read_exact(&mut buf) + .await; if !self.handle_read(read) { return None; } - Some((buf, len)) + Some( + ( + buf, len, + ), + ) } async fn poll(&mut self) -> ! { loop { - if let Err(e) = self.stream.readable().await { + if let Err(e) = self + .stream + .readable() + .await + { eprintln!("Failed waiting for client to be readable: {e}"); continue; } - if let Some((buf, len)) = self.read().await { - let req: Requests = match serde::decode_from_slice(buf.as_slice(), bincode::config::standard()) { + if let Some((buf, len)) = self + .read() + .await + { + let req: Requests = match serde::decode_from_slice( + buf.as_slice(), + bincode::config::standard(), + ) { Ok((decoded, size)) => { if size != len { eprintln!("Decoded size {size} does not match expected size {len}, decode succeeded anyways?"); @@ -119,7 +182,10 @@ impl<'a> Client<'a> { decoded } Err(e) => { - eprintln!("Failed to decode request from client {}: {e}", self.addr); + eprintln!( + "Failed to decode request from client {}: {e}", + self.addr + ); continue; } }; @@ -130,7 +196,25 @@ impl<'a> Client<'a> { // Some(Request) should be send back to client fn handle_request(&self, req: Requests) -> Option { match req { - Requests::Send((prev, (name, args))) => Some(Request::Post((name, Executor::new((name.try_into() as Result<&Task, IntoTaskErr>).unwrap().as_ref(), args.split_whitespace(), ServerCtx { stream: &self.stream }).run().to_owned()).into())), + Requests::Send((prev, (name, args))) => Some( + Request::Post( + ( + name, + Executor::new( + (name.try_into() as Result<&Task, IntoTaskErr>) + .unwrap() + .as_ref(), + args.split_whitespace(), + ServerCtx { + stream: &self.stream, + }, + ) + .run() + .to_owned(), + ) + .into(), + ), + ), _ => { panic!() } diff --git a/src/server/tasks.rs b/src/server/tasks.rs index 88f3767..c30d23a 100644 --- a/src/server/tasks.rs +++ b/src/server/tasks.rs @@ -1,12 +1,25 @@ use remote_control_rs::tasks::{CompletedTask, server::Name}; use serde::Deserialize; -pub type Task = (Name, String); +pub type Task = ( + Name, + String, +); #[derive(Deserialize)] pub enum Requests { - Send((Vec, Task)), - Respond((Vec, CompletedTask)), + Send( + ( + Vec, + Task, + ), + ), + Respond( + ( + Vec, + CompletedTask, + ), + ), } #[derive(Deserialize)] -- cgit v1.2.3