that one huge refactor

Cargo.toml

@@ -16,5 +16,7 @@ path = "src/login/main.rs"
 libpso = { path = "../libpso" }
 rand = "0.6.5"
 mio = "0.6"
+mio-extras = "2.0.5"
 crc = "^1.0.0"
-bcrypt = "0.4"
+bcrypt = "0.4"
+threadpool = "1.0"

src/common/client.rs

@@ -1,55 +1,54 @@
 use libpso::crypto::{PSOCipher, NullCipher};
 use libpso::{PSOPacket, PacketParseError};
 
-use crate::common::serverstate::{ServerState, SendServerPacket, RecvServerPacket, OnConnect};
+use crate::common::serverstate::{ServerState, SendServerPacket, RecvServerPacket, OnConnect, ClientId};
 use crate::common::network::{recv_packet, PacketNetworkError};
 
 use std::net;
-use std::io::Write;
+use std::io::{Read, Write};
 use mio::tcp::TcpStream;
 use mio::{Poll, Events, Token, Ready, PollOpt};
+use mio_extras::channel::Sender;
 
-pub struct Client<S, R, E> {
+pub struct Client<S, R> {
+    pub id: ClientId,
     running: bool,
-    socket: mio::tcp::TcpStream,
-    cipher_in: Box<dyn PSOCipher>,
-    cipher_out: Box<dyn PSOCipher>,
-    state: Box<dyn ServerState<SendPacket = S, RecvPacket = R, PacketError = E>>,
+    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, E> Client<S, R, E> where
+impl<S, R> Client<S, R> where
     S: SendServerPacket + std::fmt::Debug,
     R: RecvServerPacket + std::fmt::Debug,
-    E: std::fmt::Debug,
 {
-    pub fn new(socket: mio::tcp::TcpStream,
-               state: Box<dyn ServerState<SendPacket = S, RecvPacket = R, PacketError = E>>)
-               -> Client<S, R, E>
+    pub fn new(id: ClientId, socket: mio::tcp::TcpStream) -> Client<S, R>
     {
-        let mut client = Client {
+        Client {
+            id: id,
             running: true,
             socket: socket,
             cipher_in: Box::new(NullCipher {}),
             cipher_out: Box::new(NullCipher {}),
-            state: state,
+            recv_buffer: Vec::with_capacity(32),
+            incoming_data: Vec::new(),
             send_buffer: Vec::new(),
-        };
-
-        for task in client.state.on_connect() {
-            match task {
-                OnConnect::Packet(pkt) => client.send(pkt),
-                OnConnect::Cipher((cipher_in, cipher_out)) => {
-                    client.cipher_in = cipher_in;
-                    client.cipher_out = cipher_out;
-                },
-            }
+            _s: std::marker::PhantomData,
+            _r: std::marker::PhantomData,
         }
-        
-        client
     }
 
-    fn send_data(&mut self) {
+    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;
         }
@@ -66,8 +65,51 @@ impl<S, R, E> Client<S, R, E> where
         }
     }
 
-    // TODO: this may need to pad to 8 bytes for bb cipher
-    fn send(&mut self, pkt: S) {
+    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<_>>();
+
+            let pkt = R::from_bytes(&pkt_data[..pkt_size])
+                .map_err(|err| -> PacketNetworkError { err.into() })?;
+
+            result.push(pkt);
+        }
+        Ok(result)
+    }
+
+    pub fn send_pkt(&mut self, pkt: S) {
         println!("[send] {:?}", pkt);
         let buf = pkt.as_bytes();
         println!("[send: buf] {:?}", buf);
@@ -75,49 +117,4 @@ impl<S, R, E> Client<S, R, E> where
         self.send_buffer.append(&mut cbuf);
         self.send_data();
     }
-
-    pub fn io_loop(mut self) {
-        let poll = Poll::new().unwrap();
-        poll.register(&self.socket, Token(0), Ready::readable() | Ready::writable(), PollOpt::edge()).unwrap();
-
-        let mut events = Events::with_capacity(1024);
-        while self.running {
-            poll.poll(&mut events, None).unwrap();
-
-            for event in &events{
-                if event.token() == Token(0) {
-                    if event.readiness().is_writable() {
-                        self.send_data();
-                    }
-                    if event.readiness().is_readable() {
-                        loop {
-                            let pkt = recv_packet(&mut self.socket, &mut *self.cipher_in)
-                                .and_then(|pkt| {
-                                    R::from_bytes(&pkt)
-                                        .map_err(|err| err.into())
-                                });
-
-                            match pkt {
-                                Ok(pkt) => {
-                                    println!("[recv] {:?}", pkt);
-                                    let response = self.state.handle(&pkt);
-                                    for r in response {
-                                        self.send(r);
-                                    }
-                                },
-                                Err(err) => {
-                                    match err {
-                                        PacketNetworkError::ClientDisconnected => self.running = false,
-                                        PacketNetworkError::PacketParseError(err) => println!("undef pkt {:?}: {:?}", self.socket, err),
-                                        _ => {} //println!("error recv-ing packet with {:?}: {:?}", self.socket, err),
-                                    }
-                                    break;
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        }
-    }
 }

src/common/clientpool.rs

@@ -0,0 +1,190 @@
+use std::thread;
+use std::collections::HashMap;
+
+use std::net::{SocketAddr, Ipv4Addr};
+
+use std::sync::mpsc::TryRecvError;
+use mio::tcp::{TcpStream, TcpListener};
+use mio::{Events, Poll, Token, Ready, PollOpt};
+
+use crate::common::client::Client;
+use crate::common::serverstate::{ServerState, SendServerPacket, RecvServerPacket, OnConnect, ClientId};
+//use std::sync::mpsc;
+//use mio::channel as mpsc;
+use libpso::crypto::PSOCipher;
+
+use mio_extras::channel::{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,
+{
+    println!("client read");
+    let pkts = client.read_pkts();
+    println!("pkts: {:?}", 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),
+}
+
+
+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: ClientId,
+}
+
+
+
+
+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 = self.client_id_incr;
+        self.client_id_incr += 1;
+
+        self.poll.register(&socket, Token(client_id), Ready::readable() | Ready::writable(), PollOpt::edge()).unwrap();
+        let client = Client::new(client_id, socket);
+
+        self.client_ids.insert(Token(client_id), 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) => {
+                            println!("enc {:?}", client_id);
+                            self.clients.get_mut(&client_id)
+                                .map(|client| {
+                                    client.set_cipher(cipher_in, cipher_out);
+                                });
+                        }
+                        ClientAction::Packet(client_id, pkt) => {
+                            println!("action pkt {:?}", pkt);
+                            println!("clients! {:?} {:?}", client_id, self.clients.get(&client_id).is_some());
+                            self.clients.get_mut(&client_id)
+                                .map(|client| {
+                                    client.send_pkt(pkt);
+                                });
+                        }
+                    }
+                },
+                Err(err) => {
+                    println!("err! {:?}", err);
+                    match err {
+                        TryRecvError::Empty => break,
+                        TryRecvError::Disconnected => {
+                            // 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);
+
+        println!("looping");
+        loop {
+            self.poll.poll(&mut events, None).unwrap();
+            println!("events! {:?}", events);
+
+            for event in &events {
+                println!("event! {:?}", event);
+                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 => {
+                                            self.poll.deregister(&client.socket).unwrap();
+                                        },
+                                        _ => {},
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

src/common/mainloop.rs

@@ -0,0 +1,90 @@
+use std::thread;
+use mio::{Events, Poll, Token, Ready, PollOpt};
+use mio::tcp::TcpStream;
+use mio_extras::channel::{channel, Sender, Receiver};
+
+use crate::common::clientpool::{ClientPool, ClientAction, ClientPoolAction};
+use crate::common::serverstate::{RecvServerPacket, SendServerPacket, ServerState, OnConnect, ClientId};
+
+
+
+
+
+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,
+{
+    loop {
+        match pool_recv.try_recv() {
+            Ok(incoming) => {
+                match incoming {
+                    ClientPoolAction::NewClient(client_id) => {
+                        for s in state.on_connect().into_iter() {
+                            match s {
+                                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);
+                        for s in to_send {
+                            pool_send.send(ClientAction::Packet(s.0, s.1)).unwrap();
+                        }
+                    }
+                }
+            },
+            Err(_err) => {
+                break;
+            }
+        }
+    }  
+}
+
+
+pub fn mainloop<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 + '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();
+    });
+
+    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);
+
+        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!()
+                }
+            }
+        }
+    });
+
+    client_thread.join().unwrap();
+    handler_thread.join().unwrap();
+}

src/common/mod.rs

@@ -5,4 +5,6 @@ pub mod cipherkeys;
 pub mod network;
 pub mod serverstate;
 pub mod client;
+pub mod clientpool;
 pub mod util;
+pub mod mainloop;

src/common/serverstate.rs

@@ -1,14 +1,15 @@
 use libpso::PacketParseError;
 use libpso::crypto::PSOCipher;
 
+pub type ClientId = usize;
 
 pub enum OnConnect<S: SendServerPacket> {
     Packet(S),
-    Cipher((Box<dyn PSOCipher>, Box<dyn PSOCipher>)),
+    Cipher((Box<dyn PSOCipher + Send>, Box<dyn PSOCipher + Send>)),
 }
 
 pub trait RecvServerPacket: Sized {
-    fn from_bytes(data: &Vec<u8>) -> Result<Self, PacketParseError>;
+    fn from_bytes(data: &[u8]) -> Result<Self, PacketParseError>;
 }
 
 pub trait SendServerPacket: Sized {
@@ -21,6 +22,6 @@ pub trait ServerState {
     type PacketError;
     
     fn on_connect(&mut self) -> Vec<OnConnect<Self::SendPacket>>;
-    fn handle(&mut self, pkt: &Self::RecvPacket) -> Box<dyn Iterator<Item = Self::SendPacket>>;
+    fn handle(&mut self, id: ClientId, pkt: &Self::RecvPacket) -> Box<dyn Iterator<Item = (ClientId, Self::SendPacket)>>;
 }
 

src/login/character.rs

@@ -13,13 +13,13 @@ use elseware::common::pktvec::PktVec;
 use elseware::common::cipherkeys::{ELSEWHERE_PRIVATE_KEY, ELSEWHERE_PARRAY};
 use elseware::common::network::{PacketNetworkError};
 use elseware::common::client::Client;
-use elseware::common::serverstate::{SendServerPacket, RecvServerPacket, ServerState, OnConnect};
+use elseware::common::serverstate::{SendServerPacket, RecvServerPacket, ServerState, OnConnect, ClientId};
 use elseware::common::util::array_to_utf8;
 
 use crate::dataaccess::DataAccess;
 use crate::login::{SharedLoginState, get_login_status};
 //use crate::models::UserAccount;
-use crate::entities::UserAccount;
+use crate::entities::{UserAccount, Character};
 
 pub const CHARACTER_PORT: u16 = 12001;
 
@@ -36,7 +36,7 @@ pub enum RecvCharacterPacket {
 }
 
 impl RecvServerPacket for RecvCharacterPacket {
-    fn from_bytes(data: &Vec<u8>) -> Result<RecvCharacterPacket, PacketParseError> {
+    fn from_bytes(data: &[u8]) -> Result<RecvCharacterPacket, PacketParseError> {
         match u16::from_le_bytes([data[2], data[3]]) {
             0x93 => Ok(RecvCharacterPacket::Login(Login::from_bytes(data)?)),
             0xE0 => Ok(RecvCharacterPacket::RequestSettings(RequestSettings::from_bytes(data)?)),
@@ -71,20 +71,24 @@ impl SendServerPacket for SendCharacterPacket {
 }
 
 pub struct CharacterServerState<DA: DataAccess> {
-    shared_state: SharedLoginState<DA>,
+    //shared_state: SharedLoginState<DA>,
+    data_access: DA,
     user: Option<UserAccount>,
+    characters: Option<[Option<Character>; 4]>
 }
 
 impl<DA: DataAccess> CharacterServerState<DA> {
-    fn new(shared_state: SharedLoginState<DA>) -> CharacterServerState<DA> {
+    fn new(data_access: DA) -> CharacterServerState<DA> {
         CharacterServerState {
-            shared_state: shared_state,
+            //shared_state: shared_state,
+            data_access: data_access,
             user: None,
+            characters: None,
         }
     }
 
     fn validate_login(&mut self, pkt: &Login) -> Vec<SendCharacterPacket> {
-        match get_login_status(&self.shared_state.data_access, pkt) {
+        match get_login_status(&self.data_access, pkt) {
             Ok(user) => {
                 let mut response = LoginResponse::by_status(AccountStatus::Ok, pkt.security_data);
                 response.guildcard = user.guildcard.map_or(0, |gc| gc) as u32;
@@ -100,9 +104,9 @@ impl<DA: DataAccess> CharacterServerState<DA> {
 
     fn get_settings(&mut self) -> Vec<SendCharacterPacket> {
         let user = self.user.as_ref().unwrap();
-        let settings = match self.shared_state.data_access.get_user_settings_by_user(&user) {
+        let settings = match self.data_access.get_user_settings_by_user(&user) {
             Some(settings) => settings,
-            None => self.shared_state.data_access.create_user_settings_by_user(&user),
+            None => self.data_access.create_user_settings_by_user(&user),
         };
 
         let pkt = SendKeyAndTeamSettings::new(settings.settings.key_config,
@@ -113,11 +117,22 @@ impl<DA: DataAccess> CharacterServerState<DA> {
     }
 
     fn char_select(&mut self, select: &CharSelect) -> Vec<SendCharacterPacket> {
-        vec![SendCharacterPacket::CharAck(CharAck {
-            flag: 0,
-            slot: select.slot,
-            code: 2,
-        })]
+        /*if self.characters.is_none() {
+            self.characters = Some(self.data_access.get_characters_by_user(self.user.as_ref().unwrap()));
+        }
+
+        let chars = self.characters.as_ref().unwrap();
+        if let Some(ref char) = chars[select.slot as usize] {
+            Vec::new()
+        }
+        else {
+            vec![SendCharacterPacket::CharAck(CharAck {
+                flag: 0,
+                slot: select.slot,
+                code: 2,
+            })]
+    }*/
+        Vec::new()
     }
 
     fn validate_checksum(&mut self) -> Vec<SendCharacterPacket> {
@@ -147,16 +162,16 @@ impl<DA: DataAccess> ServerState for CharacterServerState<DA> {
         ]
     }
     
-    fn handle(&mut self, pkt: &RecvCharacterPacket) -> Box<dyn Iterator<Item = SendCharacterPacket>> {
+    fn handle(&mut self, id: ClientId, pkt: &RecvCharacterPacket) -> Box<dyn Iterator<Item = (ClientId, SendCharacterPacket)>> {
         match pkt {
             RecvCharacterPacket::Login(login) => {
-                Box::new(self.validate_login(login).into_iter())
+                Box::new(self.validate_login(login).into_iter().map(move |pkt| (id, pkt)))
             },
             RecvCharacterPacket::RequestSettings(_req) => {
-                Box::new(self.get_settings().into_iter())
+                Box::new(self.get_settings().into_iter().map(move |pkt| (id, pkt)))
             },
             RecvCharacterPacket::CharSelect(sel) => {
-                Box::new(self.char_select(sel).into_iter())
+                Box::new(self.char_select(sel).into_iter().map(move |pkt| (id, pkt)))
             }
             RecvCharacterPacket::Checksum(checksum) => {
                 Box::new(self.validate_checksum().into_iter())
@@ -165,13 +180,6 @@ impl<DA: DataAccess> ServerState for CharacterServerState<DA> {
     }
 }
 
-pub fn new_client<DA: DataAccess + 'static>(socket: mio::tcp::TcpStream, shared_state: SharedLoginState<DA>) {
-    let state = CharacterServerState::new(shared_state);
-    let client = Client::new(socket, Box::new(state));
-    client.io_loop();
-}
-
-
 #[cfg(test)]
 mod test {
     use super::*;
@@ -194,8 +202,7 @@ mod test {
             }
         }
 
-        let shared = SharedLoginState::new(TestData {});
-        let mut server = CharacterServerState::new(shared);
+        let mut server = CharacterServerState::new(TestData {});
         server.user = Some(UserAccount {
             id: 1,
             username: "testuser".to_owned(),
@@ -207,10 +214,11 @@ mod test {
             created_at: SystemTime::now(),
         });
 
-        let send = server.handle(&RecvCharacterPacket::RequestSettings(RequestSettings {flag: 0})).collect::<Vec<_>>();
+        let send = server.handle(5, &RecvCharacterPacket::RequestSettings(RequestSettings {flag: 0})).collect::<Vec<_>>();
         assert!(send.len() == 1);
 
-        let bytes = send[0].as_bytes();
+        assert!(send[0].0 == 5);
+        let bytes = send[0].1.as_bytes();
         assert!(bytes[2] == 0xE2);
         assert!(bytes.len() == 0xAFC);
     }

src/login/login.rs

@@ -15,7 +15,7 @@ use elseware::common::pktvec::PktVec;
 use elseware::common::cipherkeys::{ELSEWHERE_PRIVATE_KEY, ELSEWHERE_PARRAY};
 //use elseware::common::network::{PacketNetworkError};
 use elseware::common::client::Client;
-use elseware::common::serverstate::{SendServerPacket,  RecvServerPacket, ServerState, OnConnect};
+use elseware::common::serverstate::{SendServerPacket,  RecvServerPacket, ServerState, OnConnect, ClientId};
 use elseware::common::util::array_to_utf8;
 
 use crate::dataaccess::DataAccess;
@@ -34,7 +34,7 @@ pub enum RecvLoginPacket {
 }
 
 impl RecvServerPacket for RecvLoginPacket {
-    fn from_bytes(data: &Vec<u8>) -> Result<RecvLoginPacket, PacketParseError> {
+    fn from_bytes(data: &[u8]) -> Result<RecvLoginPacket, PacketParseError> {
         match data[2] {
             0x93 => Ok(RecvLoginPacket::Login(Login::from_bytes(data)?)),
             _ => Err(PacketParseError::WrongPacketForServerType)
@@ -75,7 +75,7 @@ impl<DA: DataAccess> SharedLoginState<DA> {
 }
 
 pub struct LoginServerState<DA: DataAccess> {
-    pub shared_state: SharedLoginState<DA>,
+    data_access: DA,
 }
 
 
@@ -92,14 +92,14 @@ pub fn get_login_status(data_access: &dyn DataAccess, pkt: &Login) -> Result<Use
 
 
 impl<DA: DataAccess> LoginServerState<DA> {
-    fn new(shared_state: SharedLoginState<DA>) -> LoginServerState<DA> {
+    pub fn new(data_access: DA) -> LoginServerState<DA> {
         LoginServerState {
-            shared_state: shared_state,
+            data_access: data_access,
         }
     }
 
     fn validate_login(&mut self, pkt: &Login) -> Vec<SendLoginPacket> {
-        match get_login_status(&self.shared_state.data_access, pkt) {
+        match get_login_status(&self.data_access, pkt) {
             Ok(_user) => {
                 let response = SendLoginPacket::LoginResponse(LoginResponse::by_status(AccountStatus::Ok, pkt.security_data));
                 let ip = net::Ipv4Addr::new(127,0,0,1);
@@ -133,22 +133,19 @@ impl<DA: DataAccess> ServerState for LoginServerState<DA> {
         ]
     }
 
-    fn handle(&mut self, pkt: &Self::RecvPacket) -> Box<dyn Iterator<Item = Self::SendPacket>> {
+    fn handle(&mut self, id: ClientId, pkt: &Self::RecvPacket) -> Box<dyn Iterator<Item = (ClientId, Self::SendPacket)>> {
         match pkt {
             RecvLoginPacket::Login(login) => {
-                Box::new(self.validate_login(login).into_iter())
+                Box::new(self.validate_login(login)
+                         .into_iter()
+                         .map(move |pkt| {
+                             (id, pkt)
+                         }))
             }
         }
     }
 }
 
-
-pub fn new_client<DA: DataAccess + 'static>(socket: mio::tcp::TcpStream, shared_state: SharedLoginState<DA>) {
-    let state = LoginServerState::new(shared_state);
-    let client = Client::new(socket, Box::new(state));
-    client.io_loop();
-}
-
 #[cfg(test)]
 mod test {
     use std::time::SystemTime;
@@ -175,10 +172,9 @@ mod test {
             }
         }
 
-        let shared = SharedLoginState::new(TestData {});
-        let mut server = LoginServerState::new(shared);
+        let mut server = LoginServerState::new(TestData {});
 
-        let send = server.handle(&RecvLoginPacket::Login(Login {
+        let send = server.handle(1, &RecvLoginPacket::Login(Login {
             flag: 0,
             tag: 65536,
             guildcard: 0,
@@ -198,7 +194,7 @@ mod test {
 
         })).collect::<Vec<_>>();
         assert!(send == vec![
-            SendLoginPacket::LoginResponse(LoginResponse {
+            (1, SendLoginPacket::LoginResponse(LoginResponse {
                 flag: 0,
                 status: AccountStatus::Ok,
                 tag: 65536,
@@ -207,12 +203,12 @@ mod test {
                 security_data: [74, 97, 107, 101, 115, 101, 114, 118, 50, 48, 50, 48, 0, 0, 0, 0,
                                 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                 caps: 258
-            }),
-            SendLoginPacket::RedirectClient(RedirectClient {
+            })),
+            (1, SendLoginPacket::RedirectClient(RedirectClient {
                 flag: 0,
                 ip: 16777343,
                 port: 12001,
                 padding: 0,
-            })])
+            }))])
     }
 }

src/login/main.rs

@@ -13,19 +13,30 @@ use std::env;
 
 use mio::tcp::TcpListener;
 use mio::{Events, Poll, Token, Ready, PollOpt};
+use mio_extras::channel::{channel, Sender, Receiver};
+//use std::sync::mpsc::{channel, Sender, Receiver};
 //use dotenv::dotenv;
 
+use elseware::common::serverstate::ServerState;
+
 use bcrypt;
 //use diesel::r2d2;
 //use diesel::prelude::*;
 //use diesel::pg::PgConnection;
 
 use libpso::character::settings;
+use libpso::character::character as pso_character;
+use libpso::character::guildcard;
 
 //use models::{NewUser, UserAccount, UserSettings, NewUserSettings};
 //use dataaccess::DBAccess;
-use entities::{UserAccount, UserSettings};
+use entities::{UserAccount, UserSettings, Character, GuildCardData};
 use dataaccess::DataAccess;
+use elseware::utf8_to_utf16_array;
+
+use elseware::common::clientpool::{ClientPool, ClientPoolAction};
+use login::LoginServerState;
+use character::CharacterServerState;
 
 use std::time::SystemTime;
 
@@ -59,87 +70,31 @@ impl DataAccess for LoginStubData {
             settings: settings::UserSettings::default()
         })
     }
-}
 
-fn main() {
-    //dotenv().ok();
-    //let database_url = env::var("DATABASE_URL").unwrap();
+    fn get_characters_by_user(&self, user: &UserAccount) -> [Option<Character>; 4] {
+        let mut c = pso_character::Character::default();
+        c.name = utf8_to_utf16_array!("Test Char", 16);
+        [Some(Character {
+            id: 1,
+            user_id: user.id,
+            character: c,
+        }),
+        None, None, None]
+    }
 
-    /*if let Some(arg) = env::args().nth(1) {
-        if arg == "dbstuff" {
-            let conn = PgConnection::establish(&database_url).unwrap();
-
-            use elseware::schema::user_accounts::dsl::*;
-            use elseware::schema::user_settings::dsl::*;
-
-            let u = NewUser::new("hi".to_string(), "qwer".to_string());
-            diesel::insert_into(user_accounts).values(&u).execute(&conn).unwrap();
-            let u = NewUser::new("hi2".to_string(), "qwer".to_string());
-            let user: UserAccount = diesel::insert_into(user_accounts).values(&u).get_result(&conn).unwrap();
-
-            let mut s = models::EUserSettings(libpso::character::settings::UserSettings::default());
-            s.0.blocked_users[5] = 99;
-            s.0.blocked_users[6] = 123;
-
-            diesel::insert_into(user_settings).values(& NewUserSettings {
-                user_id: user.id,
-                settings: s,
-            }).execute(&conn).unwrap();
-
-            let us = user_settings.load::<UserSettings>(&conn).unwrap();
-            for u in us {
-                println!("{:?}", u.settings.0.blocked_users[4]);
-                println!("{:?}", u.settings.0.blocked_users[5]);
-                println!("{:?}", u.settings.0.blocked_users[6]);
-            }
-        }
-    }*/
-
-    //let database_url = env::var("DATABASE_URL").unwrap();
-
-    println!("[login+character] starting server");
-
-    /*let connection_manager = r2d2::ConnectionManager::<PgConnection>::new(database_url);
-    let connection_pool = r2d2::Pool::builder()
-        .build(connection_manager).unwrap();*/
-
-    let login_listener = TcpListener::bind(&SocketAddr::from((Ipv4Addr::new(0,0,0,0), login::LOGIN_PORT))).unwrap();
-    let character_listener = TcpListener::bind(&SocketAddr::from((Ipv4Addr::new(0,0,0,0), character::CHARACTER_PORT))).unwrap();
-
-    //let login_shared_state = login::SharedLoginState::new(DBAccess::new(connection_pool));
-    let login_shared_state = login::SharedLoginState::new(LoginStubData {});
-
-    let poll = Poll::new().unwrap();
-    poll.register(&login_listener, Token(0), Ready::readable(), PollOpt::edge()).unwrap();
-    poll.register(&character_listener, Token(1), Ready::readable(), PollOpt::edge()).unwrap();
-
-    let mut events = Events::with_capacity(1024);
-
-    loop {
-        poll.poll(&mut events, None).unwrap();
-
-        for event in &events {
-            match event.token() {
-                Token(0) => {
-                    login_listener.accept().map(|(socket, addr)| {
-                        let shared_state_clone = login_shared_state.clone();
-                        thread::spawn(move || {
-                            println!("[login] accepted connection: {}", addr);
-                            login::new_client(socket, shared_state_clone);
-                        });
-                    });
-                },
-                Token(1) => {
-                    character_listener.accept().map(|(socket, addr)| {
-                        let shared_state_clone = login_shared_state.clone();
-                        thread::spawn(move || {
-                            println!("[character] accepted connection: {}", addr);
-                            character::new_client(socket, shared_state_clone);
-                        });
-                    });
-                },
-                _ => {}
-            }
+    fn get_guild_card_data_by_user(&self, user: &UserAccount) -> GuildCardData {
+        GuildCardData {
+            id: 1,
+            user_id: user.id,
+            guildcard: guildcard::GuildCardData::default(),
         }
     }
 }
+
+fn main() {
+    println!("[login+character] starting server");
+
+    // TODO: character mainloop
+    let auth_state = LoginServerState::new(LoginStubData {});
+    elseware::common::mainloop::mainloop(auth_state, login::LOGIN_PORT);
+}

src/patch/main.rs

@@ -6,15 +6,18 @@ use std::fs;
 use std::io;
 use std::io::{Read};
 use std::path::PathBuf;
+use mio_extras::channel::{channel, Sender, Receiver};
 use rand::{Rng, RngCore};
 use crc::{crc32, Hasher32};
+use mio::{Events, Poll, Token, Ready, PollOpt};
 use mio::tcp::TcpStream;
 use libpso::{PacketParseError, PSOPacket};
 use libpso::packet::patch::*;
 use libpso::crypto::pc::PSOPCCipher;
+use elseware::common::clientpool::{ClientPool, ClientAction, ClientPoolAction};
 use elseware::common::network::{PacketNetworkError};
 use elseware::common::client::Client;
-use elseware::common::serverstate::{RecvServerPacket, SendServerPacket, ServerState, OnConnect};
+use elseware::common::serverstate::{RecvServerPacket, SendServerPacket, ServerState, OnConnect, ClientId};
 
 const PATCH_PORT: u16 = 11000;
 
@@ -30,7 +33,6 @@ enum PatchError {
 // convert_error!(PacketNetworkError into PatchError)
 // or
 // convert_error!(std::io::Error into PatchError as IOError)
-
 impl From<PacketNetworkError> for PatchError {
     fn from(err: PacketNetworkError) -> PatchError {
         PatchError::PacketNetworkError(err)
@@ -98,7 +100,7 @@ pub enum RecvPatchPacket {
 }
 
 impl RecvServerPacket for RecvPatchPacket {
-    fn from_bytes(data: &Vec<u8>) -> Result<RecvPatchPacket, PacketParseError> {
+    fn from_bytes(data: &[u8]) -> Result<RecvPatchPacket, PacketParseError> {
         match data[2] {
             0x02 => Ok(RecvPatchPacket::PatchWelcomeReply(PatchWelcomeReply::from_bytes(data)?)),
             0x04 => Ok(RecvPatchPacket::LoginReply(LoginReply::from_bytes(data)?)),
@@ -178,20 +180,20 @@ impl ServerState for PatchServerState {
         ]
     }
 
-    fn handle(&mut self, pkt: &RecvPatchPacket) -> Box<dyn Iterator<Item = SendPatchPacket>> {
+    fn handle(&mut self, id: ClientId, pkt: &RecvPatchPacket) -> Box<dyn Iterator<Item = (ClientId, SendPatchPacket)>> {
         match pkt {
             RecvPatchPacket::PatchWelcomeReply(_pkt) => {
-                Box::new(vec![SendPatchPacket::RequestLogin(RequestLogin {})].into_iter())
+                Box::new(vec![SendPatchPacket::RequestLogin(RequestLogin {})].into_iter().map(move |pkt| (id, pkt)))
             },
             RecvPatchPacket::LoginReply(_pkt) => {
                 let mut p = vec![SendPatchPacket::Message(Message::new("hello player".to_string()))];
                 p.append(&mut get_file_list_packets(&self.patch_file_tree));
                 p.push(SendPatchPacket::PatchEndList(PatchEndList {}));
-                Box::new(p.into_iter())
+                Box::new(p.into_iter().map(move |pkt| (id, pkt)))
             },
             RecvPatchPacket::FileInfoReply(pkt) => {
                 self.patch_file_info.push(pkt.clone());
-                Box::new(None.into_iter())
+                Box::new(None.into_iter().map(move |pkt| (id, pkt)))
             },
             RecvPatchPacket::FileInfoListEnd(_pkt) => {
                 let need_update = self.patch_file_info.iter()
@@ -204,7 +206,7 @@ impl ServerState for PatchServerState {
                 let p = vec![SendPatchPacket::FilesToPatchMetadata(FilesToPatchMetadata::new(total_size, total_files)),
                              SendPatchPacket::PatchStartList(PatchStartList {})
                 ];
-                Box::new(p.into_iter().chain(SendFileIterator::new(&self)))
+                Box::new(p.into_iter().chain(SendFileIterator::new(&self)).map(move |pkt| (id, pkt)))
             }
         }
     }
@@ -382,13 +384,6 @@ impl Iterator for SendFileIterator {
 }
 
 
-
-fn new_client(socket: net::TcpStream, patch_file_tree: PatchFileTree, patch_file_lookup: HashMap<u32, PatchFile>) {
-    let state = PatchServerState::new(patch_file_tree, patch_file_lookup);
-    let client = Client::new(TcpStream::from_stream(socket).unwrap(), Box::new(state));
-    client.io_loop();
-}
-
 fn main() {
     println!("[patch] starting server");
 
@@ -412,17 +407,8 @@ fn main() {
         }
     }
 
-    let listener = TcpListener::bind(&SocketAddr::from((Ipv4Addr::new(0,0,0,0), PATCH_PORT))).unwrap();
+    let patch_state = PatchServerState::new(patch_file_tree, patch_file_lookup);
+    elseware::common::mainloop::mainloop(patch_state, PATCH_PORT);
     
-    println!("[patch] waiting for connections");
-    while let Ok((socket, addr)) = listener.accept() {
-        let local_patch_file_tree = patch_file_tree.clone();
-        let local_patch_file_lookup = patch_file_lookup.clone();
-        thread::spawn(move || {
-            println!("[patch] accepted connection: {}", addr);
-            new_client(socket, local_patch_file_tree, local_patch_file_lookup);
-        });
-    }
-
     println!("[patch] exiting...");
 }