hedgewars: comparison rust/hedgewars-server/src/server/network.rs

equal deleted inserted replaced

-:a40139603cde
+:92225a708bda
 use log::*;
 use mio::{
 net::{TcpListener, TcpStream},
 Poll, PollOpt, Ready, Token,
 };
+use mio_extras::timer;
 use netbuf;
 use slab::Slab;
 use super::{core::HWServer, coretypes::ClientId, handlers};
 use crate::{
 ssl::{
 HandshakeError, MidHandshakeSslStream, Ssl, SslContext, SslContextBuilder, SslFiletype,
 SslMethod, SslOptions, SslStream, SslStreamBuilder, SslVerifyMode,
 },
 };
+use std::time::Duration;
 const MAX_BYTES_PER_READ: usize = 2048;
+const SEND_PING_TIMEOUT: Duration = Duration::from_secs(30);
+const DROP_CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
 #[derive(Hash, Eq, PartialEq, Copy, Clone)]
 pub enum NetworkClientState {
 Idle,
 NeedsWrite,
 id: ClientId,
 socket: ClientSocket,
 peer_addr: SocketAddr,
 decoder: ProtocolDecoder,
 buf_out: netbuf::Buf,
+timeout: timer::Timeout,
 }
 impl NetworkClient {
-pub fn new(id: ClientId, socket: ClientSocket, peer_addr: SocketAddr) -> NetworkClient {
+pub fn new(
+id: ClientId,
+socket: ClientSocket,
+peer_addr: SocketAddr,
+timeout: timer::Timeout,
+) -> NetworkClient {
 NetworkClient {
 id,
 socket,
 peer_addr,
 decoder: ProtocolDecoder::new(),
 buf_out: netbuf::Buf::new(),
+timeout,
 }
 }
 #[cfg(feature = "tls-connections")]
 fn handshake_impl(
 }
 pub fn send_string(&mut self, msg: &str) {
 self.send_raw_msg(&msg.as_bytes());
 }
+pub fn replace_timeout(&mut self, timeout: timer::Timeout) -> timer::Timeout {
+replace(&mut self.timeout, timeout)
+}
 }
 #[cfg(feature = "tls-connections")]
 struct ServerSsl {
 context: SslContext,
 }
 }
 }
 }
+enum TimeoutEvent {
+SendPing,
+DropClient,
+}
+struct TimerData(TimeoutEvent, ClientId);
 pub struct NetworkLayer {
 listener: TcpListener,
 server: HWServer,
 clients: Slab<NetworkClient>,
 pending: HashSet<(ClientId, NetworkClientState)>,
 pending_cache: Vec<(ClientId, NetworkClientState)>,
 #[cfg(feature = "tls-connections")]
 ssl: ServerSsl,
 #[cfg(feature = "official-server")]
 io: IoLayer,
+timer: timer::Timer<TimerData>,
+}
+fn create_ping_timeout(timer: &mut timer::Timer<TimerData>, client_id: ClientId) -> timer::Timeout {
+timer.set_timeout(
+SEND_PING_TIMEOUT,
+TimerData(TimeoutEvent::SendPing, client_id),
+)
+}
+fn create_drop_timeout(timer: &mut timer::Timer<TimerData>, client_id: ClientId) -> timer::Timeout {
+timer.set_timeout(
+DROP_CLIENT_TIMEOUT,
+TimerData(TimeoutEvent::DropClient, client_id),
+)
 }
 impl NetworkLayer {
 pub fn new(listener: TcpListener, clients_limit: usize, rooms_limit: usize) -> NetworkLayer {
 let server = HWServer::new(clients_limit, rooms_limit);
 let clients = Slab::with_capacity(clients_limit);
 let pending = HashSet::with_capacity(2 * clients_limit);
 let pending_cache = Vec::with_capacity(2 * clients_limit);
+let timer = timer::Builder::default().build();
 NetworkLayer {
 listener,
 server,
 clients,
 pending_cache,
 #[cfg(feature = "tls-connections")]
 ssl: NetworkLayer::create_ssl_context(),
 #[cfg(feature = "official-server")]
 io: IoLayer::new(),
+timer,
 }
 }
 #[cfg(feature = "tls-connections")]
 fn create_ssl_context() -> ServerSsl {
 utils::SERVER_TOKEN,
 Ready::readable(),
 PollOpt::edge(),
 )?;
+poll.register(
+&self.timer,
+utils::TIMER_TOKEN,
+Ready::readable(),
+PollOpt::edge(),
+)?;
 #[cfg(feature = "official-server")]
 self.io.io_thread.register_rx(poll, utils::IO_TOKEN)?;
 Ok(())
 }
 Ready::readable() | Ready::writable(),
 PollOpt::edge(),
 )
 .expect("could not register socket with event loop");
-let client = NetworkClient::new(client_id, client_socket, addr);
+let client = NetworkClient::new(
+client_id,
+client_socket,
+addr,
+create_ping_timeout(&mut self.timer, client_id),
+);
 info!("client {} ({}) added", client.id, client.peer_addr);
 entry.insert(client);
 client_id
 }
 let client_id = response.client_id();
 for task in response.extract_io_tasks() {
 self.io.send(client_id, task);
 }
 }
+}
+pub fn handle_timeout(&mut self, poll: &Poll) -> io::Result<()> {
+while let Some(TimerData(event, client_id)) = self.timer.poll() {
+match event {
+TimeoutEvent::SendPing => {
+if let Some(ref mut client) = self.clients.get_mut(client_id) {
+client.send_string(&HWServerMessage::Ping.to_raw_protocol());
+client.write()?;
+client.replace_timeout(create_drop_timeout(&mut self.timer, client_id));
+}
+}
+TimeoutEvent::DropClient => {
+self.operation_failed(
+poll,
+client_id,
+&ErrorKind::TimedOut.into(),
+"No ping response",
+)?;
+}
+}
+}
+Ok(())
 }
 #[cfg(feature = "official-server")]
 pub fn handle_io_result(&mut self) {
 if let Some((client_id, result)) = self.io.try_recv() {
 self.client_error(poll, client_id)
 }
 pub fn client_readable(&mut self, poll: &Poll, client_id: ClientId) -> io::Result<()> {
 let messages = if let Some(ref mut client) = self.clients.get_mut(client_id) {
+let timeout = client.replace_timeout(create_ping_timeout(&mut self.timer, client_id));
+self.timer.cancel_timeout(&timeout);
 client.read()
 } else {
 warn!("invalid readable client: {}", client_id);
 Ok((Vec::new(), NetworkClientState::Idle))
 };

changeset 14803	92225a708bda
parent 14796	f5d43f007970
child 14807	b2beb784e4b5