author | alfadur |
Mon, 15 Apr 2019 21:22:51 +0300 | |
changeset 14808 | 92225a708bda |
parent 14801 | f5d43f007970 |
child 14812 | b2beb784e4b5 |
permissions | -rw-r--r-- |
13124 | 1 |
extern crate slab; |
2 |
||
13419 | 3 |
use std::{ |
13420 | 4 |
collections::HashSet, |
14462 | 5 |
io, |
6 |
io::{Error, ErrorKind, Read, Write}, |
|
7 |
mem::{replace, swap}, |
|
8 |
net::{IpAddr, Ipv4Addr, SocketAddr}, |
|
13419 | 9 |
}; |
10 |
||
14462 | 11 |
use log::*; |
13419 | 12 |
use mio::{ |
14462 | 13 |
net::{TcpListener, TcpStream}, |
14 |
Poll, PollOpt, Ready, Token, |
|
13419 | 15 |
}; |
14808 | 16 |
use mio_extras::timer; |
13419 | 17 |
use netbuf; |
13124 | 18 |
use slab::Slab; |
19 |
||
14785 | 20 |
use super::{core::HWServer, coretypes::ClientId, handlers}; |
13671 | 21 |
use crate::{ |
14462 | 22 |
protocol::{messages::*, ProtocolDecoder}, |
13671 | 23 |
utils, |
13419 | 24 |
}; |
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
25 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
26 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
27 |
use super::io::{IOThread, RequestId}; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
28 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
29 |
use crate::server::handlers::{IoResult, IoTask}; |
13804 | 30 |
#[cfg(feature = "tls-connections")] |
31 |
use openssl::{ |
|
14462 | 32 |
error::ErrorStack, |
13804 | 33 |
ssl::{ |
14462 | 34 |
HandshakeError, MidHandshakeSslStream, Ssl, SslContext, SslContextBuilder, SslFiletype, |
35 |
SslMethod, SslOptions, SslStream, SslStreamBuilder, SslVerifyMode, |
|
13804 | 36 |
}, |
37 |
}; |
|
14808 | 38 |
use std::time::Duration; |
13419 | 39 |
|
40 |
const MAX_BYTES_PER_READ: usize = 2048; |
|
14808 | 41 |
const SEND_PING_TIMEOUT: Duration = Duration::from_secs(30); |
42 |
const DROP_CLIENT_TIMEOUT: Duration = Duration::from_secs(10); |
|
13124 | 43 |
|
13420 | 44 |
#[derive(Hash, Eq, PartialEq, Copy, Clone)] |
13419 | 45 |
pub enum NetworkClientState { |
46 |
Idle, |
|
47 |
NeedsWrite, |
|
48 |
NeedsRead, |
|
49 |
Closed, |
|
50 |
} |
|
51 |
||
52 |
type NetworkResult<T> = io::Result<(T, NetworkClientState)>; |
|
13124 | 53 |
|
13804 | 54 |
#[cfg(not(feature = "tls-connections"))] |
55 |
pub enum ClientSocket { |
|
14462 | 56 |
Plain(TcpStream), |
13804 | 57 |
} |
58 |
||
59 |
#[cfg(feature = "tls-connections")] |
|
60 |
pub enum ClientSocket { |
|
61 |
SslHandshake(Option<MidHandshakeSslStream<TcpStream>>), |
|
14462 | 62 |
SslStream(SslStream<TcpStream>), |
13804 | 63 |
} |
64 |
||
65 |
impl ClientSocket { |
|
66 |
fn inner(&self) -> &TcpStream { |
|
67 |
#[cfg(not(feature = "tls-connections"))] |
|
68 |
match self { |
|
69 |
ClientSocket::Plain(stream) => stream, |
|
70 |
} |
|
71 |
||
72 |
#[cfg(feature = "tls-connections")] |
|
73 |
match self { |
|
74 |
ClientSocket::SslHandshake(Some(builder)) => builder.get_ref(), |
|
75 |
ClientSocket::SslHandshake(None) => unreachable!(), |
|
14462 | 76 |
ClientSocket::SslStream(ssl_stream) => ssl_stream.get_ref(), |
13804 | 77 |
} |
78 |
} |
|
79 |
} |
|
80 |
||
13124 | 81 |
pub struct NetworkClient { |
82 |
id: ClientId, |
|
13804 | 83 |
socket: ClientSocket, |
13124 | 84 |
peer_addr: SocketAddr, |
85 |
decoder: ProtocolDecoder, |
|
14462 | 86 |
buf_out: netbuf::Buf, |
14808 | 87 |
timeout: timer::Timeout, |
13124 | 88 |
} |
89 |
||
90 |
impl NetworkClient { |
|
14808 | 91 |
pub fn new( |
92 |
id: ClientId, |
|
93 |
socket: ClientSocket, |
|
94 |
peer_addr: SocketAddr, |
|
95 |
timeout: timer::Timeout, |
|
96 |
) -> NetworkClient { |
|
13124 | 97 |
NetworkClient { |
14462 | 98 |
id, |
99 |
socket, |
|
100 |
peer_addr, |
|
13124 | 101 |
decoder: ProtocolDecoder::new(), |
14462 | 102 |
buf_out: netbuf::Buf::new(), |
14808 | 103 |
timeout, |
13124 | 104 |
} |
105 |
} |
|
106 |
||
13807 | 107 |
#[cfg(feature = "tls-connections")] |
14462 | 108 |
fn handshake_impl( |
109 |
&mut self, |
|
110 |
handshake: MidHandshakeSslStream<TcpStream>, |
|
111 |
) -> io::Result<NetworkClientState> { |
|
13807 | 112 |
match handshake.handshake() { |
113 |
Ok(stream) => { |
|
114 |
self.socket = ClientSocket::SslStream(stream); |
|
14462 | 115 |
debug!( |
116 |
"TLS handshake with {} ({}) completed", |
|
117 |
self.id, self.peer_addr |
|
118 |
); |
|
13807 | 119 |
Ok(NetworkClientState::Idle) |
120 |
} |
|
121 |
Err(HandshakeError::WouldBlock(new_handshake)) => { |
|
122 |
self.socket = ClientSocket::SslHandshake(Some(new_handshake)); |
|
123 |
Ok(NetworkClientState::Idle) |
|
124 |
} |
|
13808 | 125 |
Err(HandshakeError::Failure(new_handshake)) => { |
126 |
self.socket = ClientSocket::SslHandshake(Some(new_handshake)); |
|
13807 | 127 |
debug!("TLS handshake with {} ({}) failed", self.id, self.peer_addr); |
128 |
Err(Error::new(ErrorKind::Other, "Connection failure")) |
|
129 |
} |
|
14462 | 130 |
Err(HandshakeError::SetupFailure(_)) => unreachable!(), |
13807 | 131 |
} |
132 |
} |
|
133 |
||
14462 | 134 |
fn read_impl<R: Read>( |
135 |
decoder: &mut ProtocolDecoder, |
|
136 |
source: &mut R, |
|
137 |
id: ClientId, |
|
138 |
addr: &SocketAddr, |
|
139 |
) -> NetworkResult<Vec<HWProtocolMessage>> { |
|
13419 | 140 |
let mut bytes_read = 0; |
141 |
let result = loop { |
|
13804 | 142 |
match decoder.read_from(source) { |
13419 | 143 |
Ok(bytes) => { |
13804 | 144 |
debug!("Client {}: read {} bytes", id, bytes); |
13419 | 145 |
bytes_read += bytes; |
146 |
if bytes == 0 { |
|
147 |
let result = if bytes_read == 0 { |
|
13804 | 148 |
info!("EOF for client {} ({})", id, addr); |
13419 | 149 |
(Vec::new(), NetworkClientState::Closed) |
150 |
} else { |
|
13804 | 151 |
(decoder.extract_messages(), NetworkClientState::NeedsRead) |
13419 | 152 |
}; |
153 |
break Ok(result); |
|
14462 | 154 |
} else if bytes_read >= MAX_BYTES_PER_READ { |
155 |
break Ok((decoder.extract_messages(), NetworkClientState::NeedsRead)); |
|
13419 | 156 |
} |
157 |
} |
|
158 |
Err(ref error) if error.kind() == ErrorKind::WouldBlock => { |
|
14462 | 159 |
let messages = if bytes_read == 0 { |
13419 | 160 |
Vec::new() |
161 |
} else { |
|
13804 | 162 |
decoder.extract_messages() |
13419 | 163 |
}; |
164 |
break Ok((messages, NetworkClientState::Idle)); |
|
165 |
} |
|
14462 | 166 |
Err(error) => break Err(error), |
13419 | 167 |
} |
168 |
}; |
|
169 |
result |
|
170 |
} |
|
171 |
||
13804 | 172 |
pub fn read(&mut self) -> NetworkResult<Vec<HWProtocolMessage>> { |
173 |
#[cfg(not(feature = "tls-connections"))] |
|
174 |
match self.socket { |
|
14462 | 175 |
ClientSocket::Plain(ref mut stream) => { |
176 |
NetworkClient::read_impl(&mut self.decoder, stream, self.id, &self.peer_addr) |
|
177 |
} |
|
13804 | 178 |
} |
179 |
||
180 |
#[cfg(feature = "tls-connections")] |
|
181 |
match self.socket { |
|
182 |
ClientSocket::SslHandshake(ref mut handshake_opt) => { |
|
13807 | 183 |
let handshake = std::mem::replace(handshake_opt, None).unwrap(); |
184 |
Ok((Vec::new(), self.handshake_impl(handshake)?)) |
|
14462 | 185 |
} |
186 |
ClientSocket::SslStream(ref mut stream) => { |
|
13804 | 187 |
NetworkClient::read_impl(&mut self.decoder, stream, self.id, &self.peer_addr) |
14462 | 188 |
} |
13804 | 189 |
} |
190 |
} |
|
191 |
||
192 |
fn write_impl<W: Write>(buf_out: &mut netbuf::Buf, destination: &mut W) -> NetworkResult<()> { |
|
13419 | 193 |
let result = loop { |
13804 | 194 |
match buf_out.write_to(destination) { |
14462 | 195 |
Ok(bytes) if buf_out.is_empty() || bytes == 0 => { |
14676
455865ccd36c
Server action refactoring part 2 of N
alfadur <mail@none>
parents:
14462
diff
changeset
|
196 |
break Ok(((), NetworkClientState::Idle)); |
14462 | 197 |
} |
13420 | 198 |
Ok(_) => (), |
14462 | 199 |
Err(ref error) |
200 |
if error.kind() == ErrorKind::Interrupted |
|
201 |
|| error.kind() == ErrorKind::WouldBlock => |
|
202 |
{ |
|
13419 | 203 |
break Ok(((), NetworkClientState::NeedsWrite)); |
14462 | 204 |
} |
205 |
Err(error) => break Err(error), |
|
13419 | 206 |
} |
207 |
}; |
|
13804 | 208 |
result |
209 |
} |
|
210 |
||
211 |
pub fn write(&mut self) -> NetworkResult<()> { |
|
212 |
let result = { |
|
213 |
#[cfg(not(feature = "tls-connections"))] |
|
214 |
match self.socket { |
|
14462 | 215 |
ClientSocket::Plain(ref mut stream) => { |
13804 | 216 |
NetworkClient::write_impl(&mut self.buf_out, stream) |
14462 | 217 |
} |
13804 | 218 |
} |
219 |
||
14462 | 220 |
#[cfg(feature = "tls-connections")] |
221 |
{ |
|
13804 | 222 |
match self.socket { |
13807 | 223 |
ClientSocket::SslHandshake(ref mut handshake_opt) => { |
224 |
let handshake = std::mem::replace(handshake_opt, None).unwrap(); |
|
225 |
Ok(((), self.handshake_impl(handshake)?)) |
|
226 |
} |
|
14462 | 227 |
ClientSocket::SslStream(ref mut stream) => { |
13804 | 228 |
NetworkClient::write_impl(&mut self.buf_out, stream) |
14462 | 229 |
} |
13804 | 230 |
} |
231 |
} |
|
232 |
}; |
|
233 |
||
234 |
self.socket.inner().flush()?; |
|
13419 | 235 |
result |
236 |
} |
|
237 |
||
13124 | 238 |
pub fn send_raw_msg(&mut self, msg: &[u8]) { |
13529 | 239 |
self.buf_out.write_all(msg).unwrap(); |
13124 | 240 |
} |
241 |
||
13529 | 242 |
pub fn send_string(&mut self, msg: &str) { |
13124 | 243 |
self.send_raw_msg(&msg.as_bytes()); |
244 |
} |
|
14808 | 245 |
|
246 |
pub fn replace_timeout(&mut self, timeout: timer::Timeout) -> timer::Timeout { |
|
247 |
replace(&mut self.timeout, timeout) |
|
248 |
} |
|
13124 | 249 |
} |
250 |
||
13804 | 251 |
#[cfg(feature = "tls-connections")] |
252 |
struct ServerSsl { |
|
14462 | 253 |
context: SslContext, |
13804 | 254 |
} |
255 |
||
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
256 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
257 |
pub struct IoLayer { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
258 |
next_request_id: RequestId, |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
259 |
request_queue: Vec<(RequestId, ClientId)>, |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
260 |
io_thread: IOThread, |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
261 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
262 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
263 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
264 |
impl IoLayer { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
265 |
fn new() -> Self { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
266 |
Self { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
267 |
next_request_id: 0, |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
268 |
request_queue: vec![], |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
269 |
io_thread: IOThread::new(), |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
270 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
271 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
272 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
273 |
fn send(&mut self, client_id: ClientId, task: IoTask) { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
274 |
let request_id = self.next_request_id; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
275 |
self.next_request_id += 1; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
276 |
self.request_queue.push((request_id, client_id)); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
277 |
self.io_thread.send(request_id, task); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
278 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
279 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
280 |
fn try_recv(&mut self) -> Option<(ClientId, IoResult)> { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
281 |
let (request_id, result) = self.io_thread.try_recv()?; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
282 |
if let Some(index) = self |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
283 |
.request_queue |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
284 |
.iter() |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
285 |
.position(|(id, _)| *id == request_id) |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
286 |
{ |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
287 |
let (_, client_id) = self.request_queue.swap_remove(index); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
288 |
Some((client_id, result)) |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
289 |
} else { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
290 |
None |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
291 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
292 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
293 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
294 |
fn cancel(&mut self, client_id: ClientId) { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
295 |
let mut index = 0; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
296 |
while index < self.request_queue.len() { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
297 |
if self.request_queue[index].1 == client_id { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
298 |
self.request_queue.swap_remove(index); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
299 |
} else { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
300 |
index += 1; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
301 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
302 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
303 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
304 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
305 |
|
14808 | 306 |
enum TimeoutEvent { |
307 |
SendPing, |
|
308 |
DropClient, |
|
309 |
} |
|
310 |
||
311 |
struct TimerData(TimeoutEvent, ClientId); |
|
312 |
||
13124 | 313 |
pub struct NetworkLayer { |
314 |
listener: TcpListener, |
|
315 |
server: HWServer, |
|
13419 | 316 |
clients: Slab<NetworkClient>, |
13420 | 317 |
pending: HashSet<(ClientId, NetworkClientState)>, |
13804 | 318 |
pending_cache: Vec<(ClientId, NetworkClientState)>, |
319 |
#[cfg(feature = "tls-connections")] |
|
14462 | 320 |
ssl: ServerSsl, |
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
321 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
322 |
io: IoLayer, |
14808 | 323 |
timer: timer::Timer<TimerData>, |
324 |
} |
|
325 |
||
326 |
fn create_ping_timeout(timer: &mut timer::Timer<TimerData>, client_id: ClientId) -> timer::Timeout { |
|
327 |
timer.set_timeout( |
|
328 |
SEND_PING_TIMEOUT, |
|
329 |
TimerData(TimeoutEvent::SendPing, client_id), |
|
330 |
) |
|
331 |
} |
|
332 |
||
333 |
fn create_drop_timeout(timer: &mut timer::Timer<TimerData>, client_id: ClientId) -> timer::Timeout { |
|
334 |
timer.set_timeout( |
|
335 |
DROP_CLIENT_TIMEOUT, |
|
336 |
TimerData(TimeoutEvent::DropClient, client_id), |
|
337 |
) |
|
13124 | 338 |
} |
339 |
||
340 |
impl NetworkLayer { |
|
341 |
pub fn new(listener: TcpListener, clients_limit: usize, rooms_limit: usize) -> NetworkLayer { |
|
14785 | 342 |
let server = HWServer::new(clients_limit, rooms_limit); |
13124 | 343 |
let clients = Slab::with_capacity(clients_limit); |
13420 | 344 |
let pending = HashSet::with_capacity(2 * clients_limit); |
345 |
let pending_cache = Vec::with_capacity(2 * clients_limit); |
|
14808 | 346 |
let timer = timer::Builder::default().build(); |
13804 | 347 |
|
348 |
NetworkLayer { |
|
14462 | 349 |
listener, |
350 |
server, |
|
351 |
clients, |
|
352 |
pending, |
|
353 |
pending_cache, |
|
13804 | 354 |
#[cfg(feature = "tls-connections")] |
14462 | 355 |
ssl: NetworkLayer::create_ssl_context(), |
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
356 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
357 |
io: IoLayer::new(), |
14808 | 358 |
timer, |
13804 | 359 |
} |
360 |
} |
|
361 |
||
362 |
#[cfg(feature = "tls-connections")] |
|
363 |
fn create_ssl_context() -> ServerSsl { |
|
364 |
let mut builder = SslContextBuilder::new(SslMethod::tls()).unwrap(); |
|
365 |
builder.set_verify(SslVerifyMode::NONE); |
|
366 |
builder.set_read_ahead(true); |
|
14462 | 367 |
builder |
368 |
.set_certificate_file("ssl/cert.pem", SslFiletype::PEM) |
|
369 |
.unwrap(); |
|
370 |
builder |
|
371 |
.set_private_key_file("ssl/key.pem", SslFiletype::PEM) |
|
372 |
.unwrap(); |
|
13804 | 373 |
builder.set_options(SslOptions::NO_COMPRESSION); |
374 |
builder.set_cipher_list("DEFAULT:!LOW:!RC4:!EXP").unwrap(); |
|
14462 | 375 |
ServerSsl { |
376 |
context: builder.build(), |
|
377 |
} |
|
13124 | 378 |
} |
379 |
||
380 |
pub fn register_server(&self, poll: &Poll) -> io::Result<()> { |
|
14462 | 381 |
poll.register( |
382 |
&self.listener, |
|
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
383 |
utils::SERVER_TOKEN, |
14462 | 384 |
Ready::readable(), |
385 |
PollOpt::edge(), |
|
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
386 |
)?; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
387 |
|
14808 | 388 |
poll.register( |
389 |
&self.timer, |
|
390 |
utils::TIMER_TOKEN, |
|
391 |
Ready::readable(), |
|
392 |
PollOpt::edge(), |
|
393 |
)?; |
|
394 |
||
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
395 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
396 |
self.io.io_thread.register_rx(poll, utils::IO_TOKEN)?; |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
397 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
398 |
Ok(()) |
13124 | 399 |
} |
400 |
||
401 |
fn deregister_client(&mut self, poll: &Poll, id: ClientId) { |
|
402 |
let mut client_exists = false; |
|
13419 | 403 |
if let Some(ref client) = self.clients.get(id) { |
13804 | 404 |
poll.deregister(client.socket.inner()) |
13529 | 405 |
.expect("could not deregister socket"); |
13124 | 406 |
info!("client {} ({}) removed", client.id, client.peer_addr); |
407 |
client_exists = true; |
|
408 |
} |
|
409 |
if client_exists { |
|
410 |
self.clients.remove(id); |
|
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
411 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
412 |
self.io.cancel(id); |
13124 | 413 |
} |
414 |
} |
|
415 |
||
14462 | 416 |
fn register_client( |
417 |
&mut self, |
|
418 |
poll: &Poll, |
|
419 |
client_socket: ClientSocket, |
|
420 |
addr: SocketAddr, |
|
14698 | 421 |
) -> ClientId { |
422 |
let entry = self.clients.vacant_entry(); |
|
423 |
let client_id = entry.key(); |
|
424 |
||
14462 | 425 |
poll.register( |
426 |
client_socket.inner(), |
|
14698 | 427 |
Token(client_id), |
14462 | 428 |
Ready::readable() | Ready::writable(), |
429 |
PollOpt::edge(), |
|
430 |
) |
|
431 |
.expect("could not register socket with event loop"); |
|
13124 | 432 |
|
14808 | 433 |
let client = NetworkClient::new( |
434 |
client_id, |
|
435 |
client_socket, |
|
436 |
addr, |
|
437 |
create_ping_timeout(&mut self.timer, client_id), |
|
438 |
); |
|
13124 | 439 |
info!("client {} ({}) added", client.id, client.peer_addr); |
440 |
entry.insert(client); |
|
14698 | 441 |
|
442 |
client_id |
|
13124 | 443 |
} |
444 |
||
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
445 |
fn handle_response(&mut self, mut response: handlers::Response, poll: &Poll) { |
14677
6e6632068a33
Server action refactoring part 3 of N
alfadur <mail@none>
parents:
14676
diff
changeset
|
446 |
debug!("{} pending server messages", response.len()); |
6e6632068a33
Server action refactoring part 3 of N
alfadur <mail@none>
parents:
14676
diff
changeset
|
447 |
let output = response.extract_messages(&mut self.server); |
6e6632068a33
Server action refactoring part 3 of N
alfadur <mail@none>
parents:
14676
diff
changeset
|
448 |
for (clients, message) in output { |
13424 | 449 |
debug!("Message {:?} to {:?}", message, clients); |
450 |
let msg_string = message.to_raw_protocol(); |
|
451 |
for client_id in clients { |
|
452 |
if let Some(client) = self.clients.get_mut(client_id) { |
|
453 |
client.send_string(&msg_string); |
|
14462 | 454 |
self.pending |
455 |
.insert((client_id, NetworkClientState::NeedsWrite)); |
|
13419 | 456 |
} |
457 |
} |
|
458 |
} |
|
14701 | 459 |
|
460 |
for client_id in response.extract_removed_clients() { |
|
461 |
self.deregister_client(poll, client_id); |
|
462 |
} |
|
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
463 |
|
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
464 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
465 |
{ |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
466 |
let client_id = response.client_id(); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
467 |
for task in response.extract_io_tasks() { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
468 |
self.io.send(client_id, task); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
469 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
470 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
471 |
} |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
472 |
|
14808 | 473 |
pub fn handle_timeout(&mut self, poll: &Poll) -> io::Result<()> { |
474 |
while let Some(TimerData(event, client_id)) = self.timer.poll() { |
|
475 |
match event { |
|
476 |
TimeoutEvent::SendPing => { |
|
477 |
if let Some(ref mut client) = self.clients.get_mut(client_id) { |
|
478 |
client.send_string(&HWServerMessage::Ping.to_raw_protocol()); |
|
479 |
client.write()?; |
|
480 |
client.replace_timeout(create_drop_timeout(&mut self.timer, client_id)); |
|
481 |
} |
|
482 |
} |
|
483 |
TimeoutEvent::DropClient => { |
|
484 |
self.operation_failed( |
|
485 |
poll, |
|
486 |
client_id, |
|
487 |
&ErrorKind::TimedOut.into(), |
|
488 |
"No ping response", |
|
489 |
)?; |
|
490 |
} |
|
491 |
} |
|
492 |
} |
|
493 |
Ok(()) |
|
494 |
} |
|
495 |
||
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
496 |
#[cfg(feature = "official-server")] |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
497 |
pub fn handle_io_result(&mut self) { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
498 |
if let Some((client_id, result)) = self.io.try_recv() { |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
499 |
let mut response = handlers::Response::new(client_id); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
500 |
handlers::handle_io_result(&mut self.server, client_id, &mut response, result); |
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
501 |
} |
13419 | 502 |
} |
503 |
||
13804 | 504 |
fn create_client_socket(&self, socket: TcpStream) -> io::Result<ClientSocket> { |
14462 | 505 |
#[cfg(not(feature = "tls-connections"))] |
506 |
{ |
|
13804 | 507 |
Ok(ClientSocket::Plain(socket)) |
508 |
} |
|
509 |
||
14462 | 510 |
#[cfg(feature = "tls-connections")] |
511 |
{ |
|
13804 | 512 |
let ssl = Ssl::new(&self.ssl.context).unwrap(); |
513 |
let mut builder = SslStreamBuilder::new(ssl, socket); |
|
514 |
builder.set_accept_state(); |
|
515 |
match builder.handshake() { |
|
14462 | 516 |
Ok(stream) => Ok(ClientSocket::SslStream(stream)), |
517 |
Err(HandshakeError::WouldBlock(stream)) => { |
|
518 |
Ok(ClientSocket::SslHandshake(Some(stream))) |
|
519 |
} |
|
13804 | 520 |
Err(e) => { |
521 |
debug!("OpenSSL handshake failed: {}", e); |
|
522 |
Err(Error::new(ErrorKind::Other, "Connection failure")) |
|
523 |
} |
|
524 |
} |
|
525 |
} |
|
526 |
} |
|
527 |
||
13124 | 528 |
pub fn accept_client(&mut self, poll: &Poll) -> io::Result<()> { |
529 |
let (client_socket, addr) = self.listener.accept()?; |
|
530 |
info!("Connected: {}", addr); |
|
531 |
||
14698 | 532 |
let client_id = self.register_client(poll, self.create_client_socket(client_socket)?, addr); |
533 |
||
534 |
let mut response = handlers::Response::new(client_id); |
|
535 |
||
536 |
handlers::handle_client_accept(&mut self.server, client_id, &mut response); |
|
537 |
||
538 |
if !response.is_empty() { |
|
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
539 |
self.handle_response(response, poll); |
14698 | 540 |
} |
13124 | 541 |
|
542 |
Ok(()) |
|
543 |
} |
|
544 |
||
14462 | 545 |
fn operation_failed( |
546 |
&mut self, |
|
547 |
poll: &Poll, |
|
548 |
client_id: ClientId, |
|
549 |
error: &Error, |
|
550 |
msg: &str, |
|
551 |
) -> io::Result<()> { |
|
13419 | 552 |
let addr = if let Some(ref mut client) = self.clients.get_mut(client_id) { |
553 |
client.peer_addr |
|
554 |
} else { |
|
555 |
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0) |
|
556 |
}; |
|
557 |
debug!("{}({}): {}", msg, addr, error); |
|
558 |
self.client_error(poll, client_id) |
|
13124 | 559 |
} |
560 |
||
14462 | 561 |
pub fn client_readable(&mut self, poll: &Poll, client_id: ClientId) -> io::Result<()> { |
562 |
let messages = if let Some(ref mut client) = self.clients.get_mut(client_id) { |
|
14808 | 563 |
let timeout = client.replace_timeout(create_ping_timeout(&mut self.timer, client_id)); |
564 |
self.timer.cancel_timeout(&timeout); |
|
14462 | 565 |
client.read() |
566 |
} else { |
|
567 |
warn!("invalid readable client: {}", client_id); |
|
568 |
Ok((Vec::new(), NetworkClientState::Idle)) |
|
569 |
}; |
|
13419 | 570 |
|
14676
455865ccd36c
Server action refactoring part 2 of N
alfadur <mail@none>
parents:
14462
diff
changeset
|
571 |
let mut response = handlers::Response::new(client_id); |
455865ccd36c
Server action refactoring part 2 of N
alfadur <mail@none>
parents:
14462
diff
changeset
|
572 |
|
13419 | 573 |
match messages { |
574 |
Ok((messages, state)) => { |
|
575 |
for message in messages { |
|
14676
455865ccd36c
Server action refactoring part 2 of N
alfadur <mail@none>
parents:
14462
diff
changeset
|
576 |
debug!("Handling message {:?} for client {}", message, client_id); |
14801 | 577 |
handlers::handle(&mut self.server, client_id, &mut response, message); |
13419 | 578 |
} |
579 |
match state { |
|
13420 | 580 |
NetworkClientState::NeedsRead => { |
581 |
self.pending.insert((client_id, state)); |
|
14462 | 582 |
} |
583 |
NetworkClientState::Closed => self.client_error(&poll, client_id)?, |
|
13419 | 584 |
_ => {} |
585 |
}; |
|
13124 | 586 |
} |
13419 | 587 |
Err(e) => self.operation_failed( |
14462 | 588 |
poll, |
589 |
client_id, |
|
590 |
&e, |
|
591 |
"Error while reading from client socket", |
|
592 |
)?, |
|
13124 | 593 |
} |
594 |
||
14677
6e6632068a33
Server action refactoring part 3 of N
alfadur <mail@none>
parents:
14676
diff
changeset
|
595 |
if !response.is_empty() { |
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
596 |
self.handle_response(response, poll); |
13124 | 597 |
} |
598 |
||
599 |
Ok(()) |
|
600 |
} |
|
601 |
||
14462 | 602 |
pub fn client_writable(&mut self, poll: &Poll, client_id: ClientId) -> io::Result<()> { |
603 |
let result = if let Some(ref mut client) = self.clients.get_mut(client_id) { |
|
604 |
client.write() |
|
605 |
} else { |
|
606 |
warn!("invalid writable client: {}", client_id); |
|
607 |
Ok(((), NetworkClientState::Idle)) |
|
608 |
}; |
|
13419 | 609 |
|
610 |
match result { |
|
13420 | 611 |
Ok(((), state)) if state == NetworkClientState::NeedsWrite => { |
612 |
self.pending.insert((client_id, state)); |
|
14462 | 613 |
} |
13420 | 614 |
Ok(_) => {} |
14462 | 615 |
Err(e) => { |
616 |
self.operation_failed(poll, client_id, &e, "Error while writing to client socket")? |
|
617 |
} |
|
13124 | 618 |
} |
619 |
||
620 |
Ok(()) |
|
621 |
} |
|
622 |
||
14462 | 623 |
pub fn client_error(&mut self, poll: &Poll, client_id: ClientId) -> io::Result<()> { |
13124 | 624 |
self.deregister_client(poll, client_id); |
14678
08a8605bafaf
Server action refactoring part 4 of N
alfadur <mail@none>
parents:
14677
diff
changeset
|
625 |
let mut response = handlers::Response::new(client_id); |
08a8605bafaf
Server action refactoring part 4 of N
alfadur <mail@none>
parents:
14677
diff
changeset
|
626 |
handlers::handle_client_loss(&mut self.server, client_id, &mut response); |
14784
f43ab2bd76ae
add a thread for internal server IO and implement account checking with it
alfadur
parents:
14702
diff
changeset
|
627 |
self.handle_response(response, poll); |
13124 | 628 |
|
629 |
Ok(()) |
|
630 |
} |
|
13419 | 631 |
|
632 |
pub fn has_pending_operations(&self) -> bool { |
|
633 |
!self.pending.is_empty() |
|
634 |
} |
|
635 |
||
636 |
pub fn on_idle(&mut self, poll: &Poll) -> io::Result<()> { |
|
13420 | 637 |
if self.has_pending_operations() { |
13483 | 638 |
let mut cache = replace(&mut self.pending_cache, Vec::new()); |
13420 | 639 |
cache.extend(self.pending.drain()); |
640 |
for (id, state) in cache.drain(..) { |
|
641 |
match state { |
|
14462 | 642 |
NetworkClientState::NeedsRead => self.client_readable(poll, id)?, |
643 |
NetworkClientState::NeedsWrite => self.client_writable(poll, id)?, |
|
13420 | 644 |
_ => {} |
645 |
} |
|
13419 | 646 |
} |
13420 | 647 |
swap(&mut cache, &mut self.pending_cache); |
13419 | 648 |
} |
649 |
Ok(()) |
|
650 |
} |
|
13124 | 651 |
} |