async!

Cargo.toml

@@ -19,6 +19,8 @@ path = "src/login_main.rs"
 
 [dependencies]
 libpso = { git = "http://git.sharnoth.com/jake/libpso" }
+async-std = { version = "1.4.0", features = ["unstable"] }
+futures = "0.3.1"
 rand = "0.6.5"
 mio = "0.6"
 mio-extras = "2.0.5"

src/common/client.rs

@@ -1,128 +0,0 @@
-use log::{info, trace, warn};
-
-use libpso::crypto::{PSOCipher, NullCipher};
-
-use crate::common::serverstate::{SendServerPacket, RecvServerPacket, ClientId};
-use crate::common::network::PacketNetworkError;
-
-use std::io::{Read, Write};
-
-pub struct Client<S, R> {
-    pub id: ClientId,
-    running: bool,
-    pub socket: mio::tcp::TcpStream,
-    cipher_in: Box<dyn PSOCipher + Send>,
-    cipher_out: Box<dyn PSOCipher + Send>,
-    recv_buffer: Vec<u8>,
-    incoming_data: Vec<u8>,
-    send_buffer: Vec<u8>,
-    _s: std::marker::PhantomData<S>,
-    _r: std::marker::PhantomData<R>,
-}
-
-impl<S, R> Client<S, R> where
-    S: SendServerPacket + std::fmt::Debug,
-    R: RecvServerPacket + std::fmt::Debug,
-{
-    pub fn new(id: ClientId, socket: mio::tcp::TcpStream) -> Client<S, R> {
-        Client {
-            id: id,
-            running: true,
-            socket: socket,
-            cipher_in: Box::new(NullCipher {}),
-            cipher_out: Box::new(NullCipher {}),
-            recv_buffer: Vec::with_capacity(32),
-            incoming_data: Vec::new(),
-            send_buffer: Vec::new(),
-            _s: std::marker::PhantomData,
-            _r: std::marker::PhantomData,
-        }
-    }
-
-    pub fn set_cipher(&mut self, cin: Box<dyn PSOCipher + Send>, out: Box<dyn PSOCipher + Send>) {
-        self.cipher_in = cin;
-        self.cipher_out = out;
-    }
-
-    pub fn send_data(&mut self) {
-        if self.send_buffer.len() == 0 {
-            return;
-        }
-        match self.socket.write(&self.send_buffer) {
-            Ok(len) => {
-                if len == 0 {
-                    self.running = false;
-                }
-                self.send_buffer.drain(..len);
-            },
-            Err(err) => {
-                warn!("[client] error sending data to {:?}: {:?}", self.socket, err);
-            }
-        }
-    }
-
-    fn read_data_into_buffer(&mut self) -> Result<(), PacketNetworkError> {
-        let mut new_data = [0u8; 0x8000];
-        let len = self.socket.read(&mut new_data)?;
-        if len == 0 {
-            return Err(PacketNetworkError::ClientDisconnected);
-        }
-
-        self.recv_buffer.extend_from_slice(&mut new_data[..len]);
-
-        let block_chunk_len = self.recv_buffer.len() / self.cipher_in.block_size() * self.cipher_in.block_size();
-        let buf = self.recv_buffer.drain(..block_chunk_len).collect();
-        let mut dec_buf = self.cipher_in.decrypt(&buf)?;
-        self.incoming_data.append(&mut dec_buf);
-        Ok(())
-    }
-
-    pub fn read_pkts(&mut self) -> Result<Vec<R>, PacketNetworkError> {
-        self.read_data_into_buffer()?;
-        let mut result = Vec::new();
-
-        loop {
-            if self.incoming_data.len() < 2 {
-                break;
-            }
-            let pkt_size = u16::from_le_bytes([self.incoming_data[0], self.incoming_data[1]]) as usize;
-            let mut pkt_len = pkt_size;
-            while pkt_len % self.cipher_in.block_size() != 0 {
-                pkt_len += 1;
-            }
-
-            if pkt_len > self.incoming_data.len() {
-                break;
-            }
-
-            let pkt_data = self.incoming_data.drain(..pkt_len).collect::<Vec<_>>();
-
-            trace!("[recv buf from {:?}] {:?}", self.id, pkt_data);
-            let pkt = match R::from_bytes(&pkt_data[..pkt_size]) {
-                Ok(p) => p,
-                Err(err) => {
-                    warn!("error RecvServerPacket::from_bytes: {:?}", err);
-                    continue
-                },
-            };
-
-            trace!("[recv from {:?}] {:?}", self.id, pkt);
-            result.push(pkt);
-        }
-        Ok(result)
-    }
-
-    pub fn send_pkt(&mut self, pkt: S) {
-        trace!("[send to {:?}] {:?}", self.id, pkt);
-        let buf = pkt.as_bytes();
-        if buf.len() < 1024*2 {
-            trace!("[send: buf] {:?}", buf);
-        }
-        else {
-            trace!("[send: buf] [...large buffer...]");
-        }
-        let mut cbuf = self.cipher_out.encrypt(&buf).unwrap();
-        self.send_buffer.append(&mut cbuf);
-        self.send_data();
-    }
-}

src/common/clientpool.rs

@@ -1,184 +0,0 @@
-//use std::thread;
-use std::collections::HashMap;
-
-use std::net::{SocketAddr, Ipv4Addr};
-
-use std::sync::mpsc::TryRecvError;
-use mio::tcp::TcpListener;
-use mio::{Events, Poll, Token, Ready, PollOpt};
-use log::{info, warn};
-
-use crate::common::client::Client;
-use crate::common::serverstate::{SendServerPacket, RecvServerPacket, ClientId};
-use libpso::crypto::PSOCipher;
-
-use mio_extras::channel::{Sender, Receiver};
-
-use crate::common::network::PacketNetworkError;
-
-//use threadpool::ThreadPool;
-
-//const THREAD_COUNT: usize = 4;
-
-fn client_read<S, R>(sender: &Sender<ClientPoolAction<R>>, client: &mut Client<S, R>) -> Result<(), PacketNetworkError> where
-    S: SendServerPacket + std::fmt::Debug,
-    R: RecvServerPacket + std::fmt::Debug,
-{
-    let pkts = client.read_pkts();
-
-    for pkt in pkts? {
-        sender.send(ClientPoolAction::Packet(client.id, pkt)).unwrap();
-    }
-    Ok(())
-}
-
-fn client_write<S, R>(client: &mut Client<S, R>) where
-    S: SendServerPacket + std::fmt::Debug,
-    R: RecvServerPacket + std::fmt::Debug,
-{
-    client.send_data();
-}
-
-pub enum ClientAction<S> {
-    EncryptionKeys(ClientId, Box<dyn PSOCipher + Send>, Box<dyn PSOCipher + Send>),
-    Packet(ClientId, S)
-}
-
-#[derive(Debug)]
-pub enum ClientPoolAction<R> {
-    NewClient(ClientId),
-    Packet(ClientId, R),
-    Disconnect(ClientId),
-}
-
-
-pub struct ClientPool<S, R>{
-    poll: Poll,
-    receiver: Receiver<ClientAction<S>>,
-    sender: Sender<ClientPoolAction<R>>,
-    client_ids: HashMap<Token, ClientId>,
-    clients: HashMap<ClientId, Client<S, R>>,
-    listener: TcpListener,
-    client_id_incr: usize,
-}
-
-
-
-
-impl<S, R> ClientPool<S, R> where
-    S: SendServerPacket + std::fmt::Debug,
-    R: RecvServerPacket + std::fmt::Debug,
-{
-    pub fn new(
-        receiver: Receiver<ClientAction<S>>,
-        sender: Sender<ClientPoolAction<R>>,
-        port: u16
-    ) -> ClientPool<S, R> {
-        ClientPool {
-            poll: Poll::new().unwrap(),
-            receiver: receiver,
-            sender: sender,
-            client_ids: HashMap::new(),
-            clients: HashMap::new(),
-            listener: TcpListener::bind(&SocketAddr::from((Ipv4Addr::new(0,0,0,0), port))).unwrap(),
-            client_id_incr: 3,
-        }
-    }
-
-    fn new_client(&mut self) {
-        let (socket, _addr) = self.listener.accept().unwrap();
-
-        let client_id = ClientId(self.client_id_incr);
-        self.client_id_incr += 1;
-
-        self.poll.register(&socket, Token(client_id.0), Ready::readable() | Ready::writable(), PollOpt::edge()).unwrap();
-        let client = Client::new(client_id, socket);
-
-        self.client_ids.insert(Token(client_id.0), client_id);
-        self.clients.insert(client_id, client);
-        self.sender.send(ClientPoolAction::NewClient(client_id)).unwrap();
-    }
-
-    fn packet_to_send(&mut self) {
-        loop {
-            match self.receiver.try_recv() {
-                Ok(action) => {
-                    match action {
-                        ClientAction::EncryptionKeys(client_id, cipher_in, cipher_out) => {
-                            self.clients.get_mut(&client_id)
-                                .map(|client| {
-                                    client.set_cipher(cipher_in, cipher_out);
-                                });
-                        }
-                        ClientAction::Packet(client_id, pkt) => {
-                            self.clients.get_mut(&client_id)
-                                .map(|client| {
-                                    client.send_pkt(pkt);
-                                });
-                        }
-                    }
-                },
-                Err(err) => {
-                    match err {
-                        TryRecvError::Empty => break,
-                        TryRecvError::Disconnected => {
-                            break;
-                            // TODO!
-                        }
-                    }
-                }
-            }
-        }
-    }
-
-
-    pub fn io_loop(mut self) {
-        self.poll.register(&self.listener, Token(0), Ready::readable(), PollOpt::edge()).unwrap();
-        self.poll.register(&self.receiver, Token(1), Ready::readable(), PollOpt::edge()).unwrap();
-
-        let mut events = Events::with_capacity(1024);
-
-        loop {
-            self.poll.poll(&mut events, None).unwrap();
-
-            for event in &events {
-                match event.token() {
-                    Token(0) => self.new_client(),
-                    Token(1) => self.packet_to_send(),
-                    _ => {
-                        let client_id = match self.client_ids.get(&event.token()) {
-                            Some(client_id) => client_id,
-                            None => continue,
-                        };
-
-                        let client = match self.clients.get_mut(&client_id) {
-                            Some(client) => client,
-                            None => continue,
-                        };
-
-                        if event.readiness().is_writable() {
-                            client_write(client);
-                        }
-                        if event.readiness().is_readable() {
-                            match client_read(&self.sender, client) {
-                                Ok(()) =>{},
-                                Err(err) => {
-                                    match err {
-                                        PacketNetworkError::ClientDisconnected => {
-                                            info!("client {:?} disconnected", client_id);
-                                            self.poll.deregister(&client.socket).unwrap();
-                                            self.sender.send(ClientPoolAction::Disconnect(*client_id)).unwrap();
-                                        },
-                                        _ => {
-                                            warn!("pkt err: {:?}", err);
-                                        },
-                                    }
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        }
-    }
-}

src/common/mainloop.rs

@@ -1,99 +1,260 @@
-use std::thread;
-use log::warn;
-use mio::{Events, Poll, Token, Ready, PollOpt};
-use mio_extras::channel::{channel, Sender, Receiver};
+use log::{trace, info, warn};
+use async_std::sync::{Arc, Mutex};
+use async_std::io::{Read, Write};
+use async_std::io::prelude::{ReadExt, WriteExt};
+use async_std::prelude::{StreamExt};
+use std::collections::HashMap;
 
-use crate::common::clientpool::{ClientPool, ClientAction, ClientPoolAction};
+use libpso::crypto::{PSOCipher, NullCipher};
+use crate::common::serverstate::ClientId;
 use crate::common::serverstate::{RecvServerPacket, SendServerPacket, ServerState, OnConnect};
 
-fn recv_from_clientpool<STATE, S, R, E>(state: &mut STATE,
-                                        pool_recv: &Receiver<ClientPoolAction<R>>,
-                                        pool_send: &Sender<ClientAction<S>>) where
-    STATE: ServerState<SendPacket=S, RecvPacket=R, PacketError=E>,
-    S: SendServerPacket,
-    R: RecvServerPacket,
-    E: std::fmt::Debug,
+
+enum PacketReceiverError {
+    ClientDisconnect,
+}
+
+struct PacketReceiver {
+    socket: Arc<async_std::net::TcpStream>,
+    cipher: Arc<Mutex<Box<dyn PSOCipher + Send>>>,
+    recv_buffer: Vec<u8>,
+    incoming_data: Vec<u8>,
+}
+
+impl PacketReceiver {
+    fn new(socket: Arc<async_std::net::TcpStream>, cipher: Arc<Mutex<Box<dyn PSOCipher + Send>>>) -> PacketReceiver {
+        PacketReceiver {
+            socket: socket,
+            cipher: cipher,
+            recv_buffer: Vec::new(),
+            incoming_data: Vec::new(),
+        }
+    }
+
+    async fn fill_recv_buffer(&mut self) -> Result<(), PacketReceiverError>{
+        let mut data = [0u8; 0x8000];
+
+        let mut socket = &*self.socket;
+        let len = socket.read(&mut data).await.unwrap();
+        if len == 0 {
+            return Err(PacketReceiverError::ClientDisconnect);
+        }
+
+        self.recv_buffer.extend_from_slice(&mut data[..len]);
+
+        let mut dec_buf = {
+            let mut cipher = self.cipher.lock().await;
+            let block_chunk_len = self.recv_buffer.len() / cipher.block_size() * cipher.block_size();
+            let buf = self.recv_buffer.drain(..block_chunk_len).collect();
+            cipher.decrypt(&buf).unwrap()
+        };
+        self.incoming_data.append(&mut dec_buf);
+
+        Ok(())
+    }
+
+    async fn recv_pkts<R: RecvServerPacket + Send + std::fmt::Debug>(&mut self) -> Result<Vec<R>, PacketReceiverError> {
+        self.fill_recv_buffer().await?;
+
+        let mut result = Vec::new();
+        loop {
+            if self.incoming_data.len() < 2 {
+                break;
+            }
+            let pkt_size = u16::from_le_bytes([self.incoming_data[0], self.incoming_data[1]]) as usize;
+            let mut pkt_len = pkt_size;
+            while pkt_len % self.cipher.lock().await.block_size() != 0 {
+                pkt_len += 1;
+            }
+
+            if pkt_len > self.incoming_data.len() {
+                break;
+            }
+
+            let pkt_data = self.incoming_data.drain(..pkt_len).collect::<Vec<_>>();
+
+            trace!("[recv buf] {:?}", pkt_data);
+            let pkt = match R::from_bytes(&pkt_data[..pkt_size]) {
+                Ok(p) => p,
+                Err(err) => {
+                    warn!("error RecvServerPacket::from_bytes: {:?}", err);
+                    continue
+                },
+            };
+
+            result.push(pkt);
+        }
+
+        Ok(result)
+    }
+}
+
+async fn send_pkt<S: SendServerPacket + Send + std::fmt::Debug>(socket: Arc<async_std::net::TcpStream>, cipher: Arc<Mutex<Box<dyn PSOCipher + Send>>>, pkt: S) {
+    let buf = pkt.as_bytes();
+    let cbuf = cipher.lock().await.encrypt(&buf).unwrap();
+    let mut ssock = &*socket;
+    ssock.write_all(&cbuf).await;
+
+}
+
+
+enum ClientAction<S, R> {
+    NewClient(ClientId, async_std::sync::Sender<S>),
+    Packet(ClientId, R),
+    Disconnect(ClientId),
+}
+
+enum ServerStateAction<S> {
+    Cipher(Box<dyn PSOCipher + Send + Sync>, Box<dyn PSOCipher + Send + Sync>),
+    Packet(S),
+    Disconnect,
+}
+
+async fn server_state_loop<STATE, S, R, E>(mut state: STATE,
+                                           server_state_receiver: async_std::sync::Receiver<ClientAction<ServerStateAction<S>, R>>) where
+    STATE: ServerState<SendPacket=S, RecvPacket=R, PacketError=E> + Send + 'static,
+    S: SendServerPacket + std::fmt::Debug + Send + 'static,
+    R: RecvServerPacket + std::fmt::Debug + Send + 'static,
+    E: std::fmt::Debug + Send,
 {
-    loop {
-        match pool_recv.try_recv() {
-            Ok(incoming) => {
-                match incoming {
-                    ClientPoolAction::NewClient(client_id) => {
-                        for action in state.on_connect(client_id).into_iter() {
-                            match action {
-                                OnConnect::Cipher((in_cipher, out_cipher)) => {
-                                    pool_send.send(ClientAction::EncryptionKeys(client_id, in_cipher, out_cipher)).unwrap();
-                                }
-                                OnConnect::Packet(pkt) => {
-                                    pool_send.send(ClientAction::Packet(client_id, pkt)).unwrap();
-                                }
-                            }
-                        }
-                    },
-                    ClientPoolAction::Packet(client_id, pkt) => {
-                        let to_send = state.handle(client_id, &pkt);
-                        match to_send {
-                            Ok(pkts) =>  {
-                                for pkt in pkts {
-                                    pool_send.send(ClientAction::Packet(pkt.0, pkt.1)).unwrap();
-                                }
+    async_std::task::spawn(async move {
+       let mut clients = HashMap::new();
+
+        {
+            let action = server_state_receiver.recv().await.unwrap();
+
+            match action {
+                ClientAction::NewClient(client_id, sender) => {
+                    clients.insert(client_id, sender.clone());
+                    for action in state.on_connect(client_id) {
+                        match action {
+                            OnConnect::Cipher((inc, outc)) => {
+                                sender.send(ServerStateAction::Cipher(inc, outc)).await;
                             },
-                            Err(err) => {
-                                // TODO: break?
-                                warn!("[handler error]: {:?} {:?}", client_id, err);
+                            OnConnect::Packet(pkt) => {
+                                sender.send(ServerStateAction::Packet(pkt)).await;
                             }
                         }
-                    },
-                    ClientPoolAction::Disconnect(client_id) => {
-                        for pkt in state.on_disconnect(client_id) {
-                            pool_send.send(ClientAction::Packet(pkt.0, pkt.1)).unwrap();
+                    }
+                },
+                ClientAction::Packet(client_id, pkt) => {
+                    let k = state.handle(client_id, &pkt);
+                    let pkts = k.unwrap().collect::<Vec<_>>();
+                    for (client_id, pkt) in pkts {
+                        let client = clients.get_mut(&client_id).unwrap();
+                        client.send(ServerStateAction::Packet(pkt)).await;
+                    }
+                },
+                ClientAction::Disconnect(client_id) => {
+                    let pkts = state.on_disconnect(client_id);
+                    for (client_id, pkt) in pkts {
+                        let client = clients.get_mut(&client_id).unwrap();
+                        client.send(ServerStateAction::Packet(pkt)).await;
+                    }
+
+                    let client = clients.remove(&client_id).unwrap();
+                    client.send(ServerStateAction::Disconnect).await;
+                }
+            }
+        }
+    });
+}
+
+async fn client_recv_loop<S, R>(client_id: ClientId,
+                                socket: Arc<async_std::net::TcpStream>,
+                                cipher: Arc<Mutex<Box<dyn PSOCipher + Send>>>,
+                                server_sender: async_std::sync::Sender<ClientAction<ServerStateAction<S>, R>>,
+                                client_sender: async_std::sync::Sender<ServerStateAction<S>>) where
+    S: SendServerPacket + std::fmt::Debug + Send + 'static,
+    R: RecvServerPacket + std::fmt::Debug + Send + 'static,
+{
+    async_std::task::spawn(async move {
+        server_sender.send(ClientAction::NewClient(client_id, client_sender)).await;
+        let mut pkt_receiver = PacketReceiver::new(socket, cipher);
+
+        loop {
+            match pkt_receiver.recv_pkts().await {
+                Ok(pkts) => {
+                    for pkt in pkts {
+                        trace!("[recv from {:?}] {:?}", client_id, pkt);
+                        server_sender.send(ClientAction::Packet(client_id, pkt)).await;
+                    }
+                },
+                Err(err) => {
+                    match err {
+                        PacketReceiverError::ClientDisconnect => {
+                            trace!("[client disconnected] {:?}", client_id);
+                            server_sender.send(ClientAction::Disconnect(client_id));
+                            break;
                         }
                     }
                 }
-            },
-            Err(_err) => {
-                break;
             }
         }
-    }  
+    });
 }
 
+async fn client_send_loop<S>(client_id: ClientId,
+                                 socket: Arc<async_std::net::TcpStream>,
+                                 cipher_in: Arc<Mutex<Box<dyn PSOCipher + Send>>>,
+                                 cipher_out: Arc<Mutex<Box<dyn PSOCipher + Send>>>,
+                                 client_receiver: async_std::sync::Receiver<ServerStateAction<S>>,
 
-pub fn mainloop<STATE, S, R, E>(mut state: STATE, port: u16) where
-    STATE: ServerState<SendPacket=S, RecvPacket=R, PacketError=E> + Send + 'static,
+) where
     S: SendServerPacket + std::fmt::Debug + Send + 'static,
-    R: RecvServerPacket + std::fmt::Debug + Send + 'static,
-    E: std::fmt::Debug,
 {
-    let (pool_send, pool_recv) = channel();
-    //let (patch_handler_send, patch_handler_recv) = channel::<ClientPoolAction<RecvPatchPacket>>();
-    let (handler_send, handler_recv) = channel();
-
-    //let sender_clone = patch_handler_send.clone();
-    let client_thread = thread::spawn(move || {
-        let clientpool = ClientPool::new(pool_recv, handler_send, port);
-        clientpool.io_loop();
+    async_std::task::spawn(async move {
+        loop {
+            let action = client_receiver.recv().await.unwrap();
+            match action {
+                ServerStateAction::Cipher(inc, outc) => {
+                    *cipher_in.lock().await = inc;
+                    *cipher_out.lock().await = outc;
+                }
+                ServerStateAction::Packet(pkt) => {
+                    trace!("[send to {:?}] {:?}", client_id, pkt);
+                    send_pkt(socket.clone(), cipher_out.clone(), pkt).await
+                },
+                ServerStateAction::Disconnect => {
+                    break;
+                }
+            };
+        }
     });
+}
 
-    //let handler_threadpool = threadpool::ThreadPool::new(4);
-    let handler_thread = thread::spawn(move || {
-        let poll = Poll::new().unwrap();
-        poll.register(&handler_recv, Token(0), Ready::readable(), PollOpt::edge()).unwrap();
 
-        let mut events = Events::with_capacity(1024);
+pub async fn mainloop_async<STATE, S, R, E>(mut state: STATE, port: u16) where
+    STATE: ServerState<SendPacket=S, RecvPacket=R, PacketError=E> + Send + 'static,
+    S: SendServerPacket + std::fmt::Debug + Send + Sync + 'static,
+    R: RecvServerPacket + std::fmt::Debug + Send + Sync + 'static,
+    E: std::fmt::Debug + Send,
+{
+
+    let listener = async_std::task::spawn(async move {
+        let listener = async_std::net::TcpListener::bind(&std::net::SocketAddr::from((std::net::Ipv4Addr::new(0,0,0,0), port))).await.unwrap();
+        let mut id = 1;
+
+        let (server_state_sender, server_state_receiver) = async_std::sync::channel(1024);
+
+        server_state_loop(state, server_state_receiver).await;
 
         loop {
-            poll.poll(&mut events, None).unwrap();
-            
-            for event in &events {
-                match event.token() {
-                    Token(0) => recv_from_clientpool(&mut state, &handler_recv, &pool_send),
-                    _ => panic!()
-                }
-            }
+            let (sock, addr) = listener.accept().await.unwrap();
+            let client_id = crate::common::serverstate::ClientId(id);
+            id += 1;
+
+            info!("new client {:?} {:?} {:?}", client_id, sock, addr);
+
+            let (client_sender, client_receiver) = async_std::sync::channel(64);
+            let socket = Arc::new(sock);
+            let cipher_in: Arc<Mutex<Box<dyn PSOCipher + Send>>> = Arc::new(Mutex::new(Box::new(NullCipher {})));
+            let cipher_out: Arc<Mutex<Box<dyn PSOCipher + Send>>> = Arc::new(Mutex::new(Box::new(NullCipher {})));
+
+            client_recv_loop(client_id, socket.clone(), cipher_in.clone(), server_state_sender.clone(), client_sender).await;
+            client_send_loop(client_id, socket.clone(), cipher_in.clone(), cipher_out.clone(), client_receiver).await;
         }
     });
 
-    client_thread.join().unwrap();
-    handler_thread.join().unwrap();
+    listener.await
 }

src/common/mod.rs

@@ -1,8 +1,6 @@
 pub mod cipherkeys;
 pub mod network;
 pub mod serverstate;
-pub mod client;
-pub mod clientpool;
 pub mod mainloop;
 pub mod leveltable;
 

src/common/network.rs

@@ -31,44 +31,3 @@ impl From<PacketParseError> for PacketNetworkError {
         PacketNetworkError::PacketParseError(err)
     }
 }
-
-pub fn recv_packet<T: Read>(socket: &mut T, cipher: &mut dyn PSOCipher) -> Result<Vec<u8>, PacketNetworkError> {
-    let mut size_buf = vec![0u8; cipher.header_size()];
-    let mut offset = 0;
-    while offset < cipher.header_size() {
-        let diff = socket.read(&mut size_buf[offset..])?;
-        if diff == 0 {
-            return Err(PacketNetworkError::ClientDisconnected);
-        }
-        offset += diff;
-    }
-
-    let mut dec_size_buf = cipher.decrypt(&size_buf)?;
-    let pkt_size = u16::from_le_bytes([dec_size_buf[0], dec_size_buf[1]]) as usize;
-
-    let mut buf_size = pkt_size;
-    while buf_size % cipher.block_size() != 0 {
-        buf_size += 1;
-    }
-    buf_size -= cipher.header_size();
-
-    let mut data_buf = vec![0u8; buf_size];
-    let mut offset = 0;
-    while offset < buf_size {
-        let diff = socket.read(&mut data_buf[offset..])?;
-        if diff == 0 {
-            return Err(PacketNetworkError::ClientDisconnected);
-        }
-        offset += diff;
-    }
-
-    let mut dec_data_buf = cipher.decrypt(&data_buf.to_vec())?;
-
-    let mut full_buf = Vec::new();
-    full_buf.append(&mut dec_size_buf);
-    full_buf.append(&mut dec_data_buf);
-    full_buf = full_buf[..pkt_size].to_vec();
-
-    trace!("[recv: buf]: {:X?}", full_buf);
-    Ok(full_buf)
-}

src/common/serverstate.rs

@@ -6,7 +6,7 @@ pub struct ClientId(pub usize);
 
 pub enum OnConnect<S: SendServerPacket> {
     Packet(S),
-    Cipher((Box<dyn PSOCipher + Send>, Box<dyn PSOCipher + Send>)),
+    Cipher((Box<dyn PSOCipher + Send + Sync>, Box<dyn PSOCipher + Send + Sync>)),
 }
 
 pub trait RecvServerPacket: Sized {
@@ -24,7 +24,7 @@ pub trait ServerState {
     
     fn on_connect(&mut self, id: ClientId) -> Vec<OnConnect<Self::SendPacket>>;
     fn handle(&mut self, id: ClientId, pkt: &Self::RecvPacket)
-              -> Result<Box<dyn Iterator<Item = (ClientId, Self::SendPacket)>>, Self::PacketError>;
+              -> Result<Box<dyn Iterator<Item = (ClientId, Self::SendPacket)> + Send>, Self::PacketError>;
     fn on_disconnect(&mut self, id: ClientId) -> Vec<(ClientId, Self::SendPacket)>;
 }
 

src/login/character.rs

@@ -477,7 +477,7 @@ impl<EG: EntityGateway> ServerState for CharacterServerState<EG> {
     }
 
     fn handle(&mut self, id: ClientId, pkt: &RecvCharacterPacket)
-              -> Result<Box<dyn Iterator<Item = (ClientId, SendCharacterPacket)>>, CharacterError> {
+              -> Result<Box<dyn Iterator<Item = (ClientId, SendCharacterPacket)> + Send>, CharacterError> {
         Ok(match pkt {
             RecvCharacterPacket::Login(login) => {
                 if login.session.action == SessionAction::SelectCharacter {

src/login/login.rs

@@ -120,7 +120,7 @@ impl<EG: EntityGateway> ServerState for LoginServerState<EG> {
     }
 
     fn handle(&mut self, id: ClientId, pkt: &Self::RecvPacket)
-              -> Result<Box<dyn Iterator<Item = (ClientId, Self::SendPacket)>>, LoginError> {
+              -> Result<Box<dyn Iterator<Item = (ClientId, Self::SendPacket)> + Send>, LoginError> {
         Ok(match pkt {
             RecvLoginPacket::Login(login) => {
                 Box::new(self.validate_login(login)

src/main.rs

@@ -145,35 +145,41 @@ fn main() {
     character.character.name = utf8_to_utf16_array!("Test Char 8", 0x10);
     entity_gateway.set_character(&character);
 
-    let patch_thread = thread::spawn(|| {
-        info!("[patch] starting server");
-        let patch_config = load_config();
-        let patch_motd = load_motd();
-        let (patch_file_tree, patch_file_lookup) = generate_patch_tree(patch_config.path.as_str());
-        let patch_state = PatchServerState::new(patch_file_tree, patch_file_lookup, patch_motd);
-        common::mainloop::mainloop(patch_state, patch_config.port);
-    });
-    let thread_entity_gateway = entity_gateway.clone();
-    let auth_thread = thread::spawn(|| {
-        info!("[auth] starting server");
-        let auth_state = LoginServerState::new(thread_entity_gateway);
-        common::mainloop::mainloop(auth_state, login::login::LOGIN_PORT);
-    });
-    let thread_entity_gateway = entity_gateway.clone();
-    let char_thread = thread::spawn(|| {
-        info!("[character] starting server");
-        let char_state = CharacterServerState::new(thread_entity_gateway);
-        common::mainloop::mainloop(char_state, login::character::CHARACTER_PORT);
-    });
-    let thread_entity_gateway = entity_gateway.clone();
-    let ship_thread = thread::spawn(|| {
-        info!("[ship] starting server");
-        let ship_state = ShipServerState::new(thread_entity_gateway);
-        common::mainloop::mainloop(ship_state, ship::ship::SHIP_PORT);
-    });
+    async_std::task::block_on(async move {
+        let thread_entity_gateway = entity_gateway.clone();
+        let patch = async_std::task::spawn(async {
+            info!("[patch] starting server");
+            let patch_config = load_config();
+            let patch_motd = load_motd();
+            let (patch_file_tree, patch_file_lookup) = generate_patch_tree(patch_config.path.as_str());
+            let patch_state = PatchServerState::new(patch_file_tree, patch_file_lookup, patch_motd);
+
+            crate::common::mainloop::mainloop_async(patch_state, patch_config.port).await;
+        });
+
+        let thread_entity_gateway = entity_gateway.clone();
+        let auth = async_std::task::spawn(async {
+            info!("[auth] starting server");
+            let auth_state = LoginServerState::new(thread_entity_gateway);
 
-    patch_thread.join().unwrap();
-    auth_thread.join().unwrap();
-    char_thread.join().unwrap();
-    ship_thread.join().unwrap();
+            common::mainloop::mainloop_async(auth_state, login::login::LOGIN_PORT).await;
+        });
+
+        let thread_entity_gateway = entity_gateway.clone();
+        let character = async_std::task::spawn(async {
+            info!("[character] starting server");
+            let char_state = CharacterServerState::new(thread_entity_gateway);
+
+            common::mainloop::mainloop_async(char_state, login::character::CHARACTER_PORT).await;
+        });
+
+        let thread_entity_gateway = entity_gateway.clone();
+        let ship = async_std::task::spawn(async {
+            info!("[ship] starting server");
+            let ship_state = ShipServerState::new(thread_entity_gateway);
+            common::mainloop::mainloop_async(ship_state, ship::ship::SHIP_PORT).await;
+        });
+
+        futures::join!(patch, auth, character, ship);
+    });
 }

src/patch/patch.rs

@@ -170,7 +170,7 @@ impl ServerState for PatchServerState {
     }
 
     fn handle(&mut self, id: ClientId, pkt: &RecvPatchPacket)
-              -> Result<Box<dyn Iterator<Item = (ClientId, SendPatchPacket)>>, PatchError> {
+              -> Result<Box<dyn Iterator<Item = (ClientId, SendPatchPacket)> + Send>, PatchError> {
         Ok(match pkt {
             RecvPatchPacket::PatchWelcomeReply(_pkt) => {
                 Box::new(vec![SendPatchPacket::RequestLogin(RequestLogin {})].into_iter().map(move |pkt| (id, pkt)))
@@ -287,7 +287,7 @@ fn does_file_need_updating(file_info: &FileInfoReply, patch_file_lookup: &HashMa
 
 struct SendFileIterator {
     done: bool,
-    file_iter: Box<dyn Iterator<Item = PatchTreeIterItem>>,
+    file_iter: Box<dyn Iterator<Item = PatchTreeIterItem> + Send>,
     patch_file_lookup: HashMap<u32, PatchFile>,
     current_file: Option<io::BufReader<fs::File>>,
     chunk_num: u32,

src/ship/ship.rs

@@ -314,7 +314,7 @@ impl<EG: EntityGateway> ShipServerState<EG> {
         Ok(v)
     }
 
-    fn message(&mut self, id: ClientId, msg: &Message) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)>> {
+    fn message(&mut self, id: ClientId, msg: &Message) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)> + Send> {
         let cmsg = msg.clone();
         Box::new(self.client_location.get_area_by_user(id).clients().iter()
                  .filter(|client| client.client_id != id)
@@ -323,7 +323,7 @@ impl<EG: EntityGateway> ShipServerState<EG> {
                  }).collect::<Vec<_>>().into_iter())
     }
 
-    fn direct_message(&mut self, id: ClientId, msg: &DirectMessage) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)>> {
+    fn direct_message(&mut self, id: ClientId, msg: &DirectMessage) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)> + Send> {
         let cmsg = msg.clone();
         Box::new(self.client_location.get_area_by_user(id).clients().iter()
                  .filter(|client| client.index == cmsg.flag as usize)
@@ -332,7 +332,7 @@ impl<EG: EntityGateway> ShipServerState<EG> {
                  }).collect::<Vec<_>>().into_iter())
     }
 
-    fn player_chat(&mut self, id: ClientId, msg: &PlayerChat) -> Result<Box<dyn Iterator<Item = (ClientId, SendShipPacket)>>, ShipError> {
+    fn player_chat(&mut self, id: ClientId, msg: &PlayerChat) -> Result<Box<dyn Iterator<Item = (ClientId, SendShipPacket)> + Send>, ShipError> {
         let client = self.clients.get_mut(&id).ok_or(ShipError::ClientNotFound(id))?;
         let cmsg = PlayerChat::new(client.user.guildcard.unwrap(), msg.message.clone());
 
@@ -342,7 +342,7 @@ impl<EG: EntityGateway> ShipServerState<EG> {
                 }).collect::<Vec<_>>().into_iter()))
     }
 
-    fn create_room(&mut self, id: ClientId, create_room: &CreateRoom) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)>> {
+    fn create_room(&mut self, id: ClientId, create_room: &CreateRoom) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)> + Send> {
         let area = self.client_location.get_area_by_user(id);
         let area_client = area.clients().into_iter().filter(|client| {
             client.client_id == id
@@ -394,7 +394,7 @@ impl<EG: EntityGateway> ShipServerState<EG> {
         })))
     }
 
-    fn room_name_request(&mut self, id: ClientId) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)>> {
+    fn room_name_request(&mut self, id: ClientId) -> Box<dyn Iterator<Item = (ClientId, SendShipPacket)> + Send> {
         let area = self.client_location.get_area_by_user(id);
         let room_state = self.rooms[area.id()].as_ref().unwrap();
         Box::new(vec![(id, SendShipPacket::RoomNameResponse(RoomNameResponse {name: room_state.name.clone()}))].into_iter())
@@ -422,7 +422,7 @@ impl<EG: EntityGateway> ServerState for ShipServerState<EG> {
     }
 
     fn handle(&mut self, id: ClientId, pkt: &RecvShipPacket)
-              -> Result<Box<dyn Iterator<Item = (ClientId, SendShipPacket)>>, ShipError> {
+              -> Result<Box<dyn Iterator<Item = (ClientId, SendShipPacket)> + Send>, ShipError> {
         Ok(match pkt {
             RecvShipPacket::Login(login) => {
                 Box::new(self.validate_login(id, login)?.into_iter().map(move |pkt| (id, pkt)))