src/server.rs

   1 #![allow(unused)]
   2
   3 pub mod pkts;
   4
   5 use super::pkt;
   6 use super::quit::Quit;
   7 use connect::{tcp::TcpListener, ConnectDatagram, ConnectionReader, ConnectionWriter, StreamExt};
   8 use pkts::*;
   9 use std::{
  10     collections::HashMap,
  11     sync::{Arc, RwLock, Weak},
  12 };
  13 use tokio::{
  14     sync::{broadcast, mpsc, Mutex as AsyncMutex},
  15     task,
  16 };
  17
  18 type ClientId = u64;
  19
  20 pub struct Client {
  21     pub id: ClientId,
  22     pub conn: AsyncMutex<ConnectionWriter>,
  23     pub server: Weak<ServerData>,
  24     quit: Quit,
  25 }
  26
  27 impl Client {
  28     async fn login(&self, pkt: &Login) {
  29         println!("login {} {}", pkt.name, pkt.pwd);
  30     }
  31
  32     async fn run(&self, mut reader: ConnectionReader) {
  33         let mut quit = self.quit.subscribe();
  34
  35         loop {
  36             tokio::select! {
  37                 Some(msg) = reader.next() => match msg.recipient() {
  38                     LOGIN if let Some(pkt) = pkt::get::<Login>(&msg) =>
  39                         self.login(&pkt).await,
  40                     _ => {},
  41                 },
  42                 _ = quit.recv() => break,
  43                 else => break,
  44             }
  45         }
  46
  47         if let Some(server) = self.server.upgrade() {
  48             server.clients_by_id.write().unwrap().remove(&self.id);
  49         }
  50
  51         println!("disconnect {}", self.id);
  52     }
  53 }
  54
  55 pub struct ServerData {
  56     clients_by_id: RwLock<HashMap<ClientId, Arc<Client>>>,
  57     clients_by_name: RwLock<HashMap<String, Arc<Client>>>,
  58 }
  59
  60 pub struct Server {
  61     listener: TcpListener,
  62     data: Arc<ServerData>,
  63     quit: Quit,
  64 }
  65
  66 impl Server {
  67     pub async fn new(addr: &str, quit: Quit) -> Self {
  68         println!("listen {addr}");
  69
  70         Self {
  71             quit,
  72             listener: TcpListener::bind(addr).await.unwrap(),
  73             data: Arc::new(ServerData {
  74                 clients_by_id: RwLock::new(HashMap::new()),
  75                 clients_by_name: RwLock::new(HashMap::new()),
  76             }),
  77         }
  78     }
  79
  80     pub async fn run(mut self) {
  81         let mut next_id: ClientId = 0;
  82         let mut quit = self.quit.subscribe();
  83
  84         loop {
  85             tokio::select! {
  86                 Some(conn) = self.listener.next() => {
  87                     println!("connect {}", conn.peer_addr());
  88
  89                     let (reader, writer) = conn.split();
  90                     let client = Arc::new(Client {
  91                         id: next_id,
  92                         conn: AsyncMutex::new(writer),
  93                         server: Arc::downgrade(&self.data),
  94                         quit: self.quit.clone(),
  95                     });
  96
  97                     self.data
  98                         .clients_by_id
  99                         .write()
 100                         .unwrap()
 101                         .insert(next_id, Arc::clone(&client));
 102
 103                     next_id += 1;
 104
 105                     task::spawn(async move { (*client).run(reader).await });
 106                 },
 107                 _ = quit.recv() => break,
 108                 else => break,
 109             }
 110         }
 111
 112         println!("shutdown");
 113     }
 114 }