use std::time::Duration; use std::pin::Pin; use futures::future::Future; use log::{info, warn}; use async_std::sync::{Arc, Mutex}; use async_std::io::prelude::{ReadExt, WriteExt}; use std::collections::HashMap; use serde::Serialize; use serde::de::DeserializeOwned; use crate::common::interserver::{ServerId, InterserverActor}; use libpso::crypto::{PSOCipher, NullCipher, CipherError}; use crate::common::serverstate::{ServerState, SendServerPacket, RecvServerPacket}; use crate::login::character::CharacterServerState; //use crate::ship::ship::ShipServerState; use crate::entity::gateway::entitygateway::EntityGateway; use async_std::channel; #[derive(Debug)] enum MessageReceiverError { //InvalidSize, InvalidPayload, //NetworkError(std::io::Error), Disconnected, } struct MessageReceiver { socket: async_std::net::TcpStream, } impl MessageReceiver { fn new(socket: async_std::net::TcpStream) -> MessageReceiver { MessageReceiver { socket, } } async fn recv(&mut self) -> Result { let mut size_buf = [0u8; 4]; self.socket.read_exact(&mut size_buf).await.map_err(|_| MessageReceiverError::Disconnected)?; let size = u32::from_le_bytes(size_buf) as usize; let mut payload = vec![0u8; size]; self.socket.read_exact(&mut payload).await.map_err(|_| MessageReceiverError::Disconnected)?; let payload = String::from_utf8(payload).map_err(|_| MessageReceiverError::InvalidPayload)?; let msg = serde_json::from_str(&payload).map_err(|_| MessageReceiverError::InvalidPayload)?; Ok(msg) } } /* #[derive(Debug)] enum InterserverInputAction { NewConnection(ServerId, async_std::channel::Sender), Message(ServerId, R), Disconnect(ServerId), } async fn interserver_state_loop(state: Arc>, action_receiver: async_std::channel::Receiver>) where A: InterserverActor + Send + 'static, S: Serialize + Send + 'static, R: DeserializeOwned + Send + 'static, { async_std::task::spawn(async move { let mut ships = HashMap::new(); loop { info!("interserver loop"); let action = match action_receiver.recv().await { Ok(action) => action, Err(err) => { warn!("error in iterserver state loop {:?}", err); continue; } }; let mut state = state.lock().await; match action { InterserverInputAction::NewConnection(server_id, ship_action_sender) => { ships.insert(server_id, ship_action_sender); for (server, action) in state.on_connect(server_id).await { if let Some(sender) = ships.get_mut(&server) { sender.send(action).await.unwrap(); } } }, InterserverInputAction::Message(server_id, message) => { let actions = state.action(server_id, message).await; match actions { Ok(actions) => { for (server, action) in actions{ if let Some(sender) = ships.get_mut(&server) { sender.send(action).await.unwrap(); } } }, Err(err) => { warn!("[server {:?} state handler error] {:?}", server_id, err); } } }, InterserverInputAction::Disconnect(server_id) => { let actions = state.on_disconnect(server_id).await; ships.remove(&server_id); for (server, action) in actions { if let Some(sender) = ships.get_mut(&server) { sender.send(action).await.unwrap(); } } } } } }); } async fn login_recv_loop(server_id: ServerId, socket: async_std::net::TcpStream, state_loop_sender: async_std::channel::Sender>, output_loop_sender: async_std::channel::Sender) where S: Serialize + std::fmt::Debug + Send + 'static, R: DeserializeOwned + std::fmt::Debug + Send + 'static, { async_std::task::spawn(async move { state_loop_sender.send(InterserverInputAction::NewConnection(server_id, output_loop_sender)).await.unwrap(); let mut msg_receiver = MessageReceiver::new(socket); loop { info!("login recv loop"); match msg_receiver.recv().await { Ok(msg) => { info!("[login recv loop msg] {:?}", msg); state_loop_sender.send(InterserverInputAction::Message(server_id, msg)).await.unwrap(); }, Err(err) => { if let MessageReceiverError::Disconnected = err { info!("[login recv loop disconnect] {:?}", server_id); state_loop_sender.send(InterserverInputAction::Disconnect(server_id)).await.unwrap(); break; } info!("[login recv loop err] {:?}", err); } } } }); } async fn interserver_send_loop(server_id: ServerId, mut socket: async_std::net::TcpStream, output_loop_receiver: async_std::channel::Receiver) where S: Serialize + std::fmt::Debug + Send + 'static, { async_std::task::spawn(async move { loop { info!("login send loop"); match output_loop_receiver.recv().await { Ok(msg) => { let payload = serde_json::to_string(&msg); if let Ok(payload) = payload { let len_bytes = u32::to_le_bytes(payload.len() as u32); if let Err(err) = socket.write_all(&len_bytes).await { warn!("interserver send failed: {:?}", err); break; } if let Err(err) = socket.write_all(payload.as_bytes()).await { warn!("intserserver send failed: {:?}", err); break; } } }, Err(err) => { warn!("error in send_loop: {:?}, {:?}", server_id, err); break; } } } }); } pub fn login_listen_mainloop(state: Arc>>, port: u16) -> Pin>> { Box::pin(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 = 0; let (server_state_sender, server_state_receiver) = async_std::channel::bounded(1024); interserver_state_loop(state.clone(), server_state_receiver).await; loop { let (socket, addr) = listener.accept().await.unwrap(); info!("new ship server: {:?} {:?}", socket, addr); id += 1; let server_id = crate::common::interserver::ServerId(id); let (client_sender, client_receiver) = async_std::channel::bounded(64); { let mut state = state.lock().await; let local_sender = client_sender.clone(); state.set_sender(server_id, Box::new(move |message| { async_std::task::block_on(local_sender.send(message)).unwrap(); })) } login_recv_loop(server_id, socket.clone(), server_state_sender.clone(), client_sender).await; interserver_send_loop(server_id, socket.clone(), client_receiver).await; } })) } */ /* pub fn ship_connect_mainloop(state: Arc>>, ip: std::net::Ipv4Addr, port: u16) -> Pin>> { Box::pin(async_std::task::spawn(async move { let mut id = 0; let (server_state_sender, server_state_receiver) = async_std::channel::bounded(1024); interserver_state_loop(state.clone(), server_state_receiver).await; loop { info!("trying to connect to loginserv"); let socket = match async_std::net::TcpStream::connect((ip, port)).await { Ok(socket) => socket, Err(err) => { info!("err trying to connect to loginserv {:?}", err); async_std::task::sleep(Duration::from_secs(10)).await; continue; } }; id += 1; let server_id = crate::common::interserver::ServerId(id); info!("found loginserv: {:?} {:?}", server_id, socket); let (client_sender, client_receiver) = async_std::channel::bounded(64); { let mut state = state.lock().await; let local_sender = client_sender.clone(); state.set_sender(Box::new(move |message| { async_std::task::block_on(local_sender.send(message)).unwrap(); })) } login_recv_loop(server_id, socket.clone(), server_state_sender.clone(), client_sender).await; interserver_send_loop(server_id, socket.clone(), client_receiver).await; let mut buf = [0u8; 1]; loop { let peek = socket.peek(&mut buf).await; match peek { Ok(len) if len == 0 => { break }, _ => { } } } } })) } */