src/server.rs

   1 pub mod pkts;
   2 mod remote_client;
   3
   4 use crate::quit::Quit;
   5 use connect::{tcp::TcpListener, StreamExt};
   6 use log::*;
   7 use remote_client::{Client, ClientId};
   8 use std::{
   9     collections::HashMap,
  10     sync::{Arc, RwLock},
  11 };
  12 use tokio::{sync::Mutex as AsyncMutex, task};
  13
  14 pub struct ServerData {
  15     clients_by_id: RwLock<HashMap<ClientId, Arc<Client>>>,
  16     clients_by_name: RwLock<HashMap<String, Arc<Client>>>,
  17 }
  18
  19 pub struct Server {
  20     listener: TcpListener,
  21     data: Arc<ServerData>,
  22     quit: Quit,
  23 }
  24
  25 impl Server {
  26     pub async fn new(addr: &str, quit: Quit) -> Self {
  27         Self {
  28             quit,
  29             listener: TcpListener::bind(addr).await.unwrap(),
  30             data: Arc::new(ServerData {
  31                 clients_by_id: RwLock::new(HashMap::new()),
  32                 clients_by_name: RwLock::new(HashMap::new()),
  33             }),
  34         }
  35     }
  36
  37     pub async fn run(mut self) {
  38         let mut next_id: ClientId = 0;
  39         let mut quit = self.quit.subscribe();
  40
  41         loop {
  42             tokio::select! {
  43                 conn = self.listener.next() => {
  44                     let conn = conn.expect("Listener interrupted");
  45
  46                     info!("Client from {} assigned id {next_id}", conn.peer_addr());
  47
  48                     let (reader, writer) = conn.split();
  49                     let client = Arc::new(Client {
  50                         id: next_id,
  51                         conn: AsyncMutex::new(writer),
  52                         server: Arc::downgrade(&self.data),
  53                         quit: self.quit.clone(),
  54                     });
  55
  56                     self.data
  57                         .clients_by_id
  58                         .write()
  59                         .unwrap()
  60                         .insert(next_id, Arc::clone(&client));
  61
  62                     next_id += 1;
  63
  64                     task::spawn(async move { (*client).run(reader).await });
  65                 },
  66                 _ = quit.recv() => {
  67                     trace!("Quit signal received");
  68                     break;
  69                 },
  70                 else => unreachable!("Quit channel broke"),
  71             }
  72         }
  73
  74         info!("Stopped server");
  75     }
  76 }