|
15831
|
1 |
use bytes::{Buf, BufMut, BytesMut};
|
|
|
2 |
use log::*;
|
|
15832
|
3 |
use std::{
|
|
|
4 |
error::Error,
|
|
|
5 |
fmt::{Debug, Display, Formatter},
|
|
|
6 |
io,
|
|
|
7 |
io::ErrorKind,
|
|
|
8 |
marker::Unpin,
|
|
|
9 |
time::Duration,
|
|
|
10 |
};
|
|
|
11 |
use tokio::{io::AsyncReadExt, time::timeout};
|
|
15831
|
12 |
|
|
15832
|
13 |
use crate::protocol::ProtocolError::Timeout;
|
|
15804
|
14 |
use hedgewars_network_protocol::{
|
|
|
15 |
messages::HwProtocolMessage,
|
|
15832
|
16 |
parser::HwProtocolError,
|
|
15804
|
17 |
parser::{malformed_message, message},
|
|
|
18 |
};
|
|
12129
|
19 |
|
|
15832
|
20 |
#[derive(Debug)]
|
|
|
21 |
pub enum ProtocolError {
|
|
|
22 |
Eof,
|
|
|
23 |
Timeout,
|
|
|
24 |
Network(Box<dyn Error + Send>),
|
|
|
25 |
}
|
|
|
26 |
|
|
|
27 |
impl Display for ProtocolError {
|
|
|
28 |
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
|
|
29 |
match self {
|
|
|
30 |
ProtocolError::Eof => write!(f, "Connection reset by peer"),
|
|
|
31 |
ProtocolError::Timeout => write!(f, "Read operation timed out"),
|
|
|
32 |
ProtocolError::Network(source) => write!(f, "{:?}", source),
|
|
|
33 |
}
|
|
|
34 |
}
|
|
|
35 |
}
|
|
|
36 |
|
|
|
37 |
impl Error for ProtocolError {
|
|
|
38 |
fn source(&self) -> Option<&(dyn Error + 'static)> {
|
|
|
39 |
if let Self::Network(source) = self {
|
|
|
40 |
Some(source.as_ref())
|
|
|
41 |
} else {
|
|
|
42 |
None
|
|
|
43 |
}
|
|
|
44 |
}
|
|
|
45 |
}
|
|
|
46 |
|
|
|
47 |
pub type Result<T> = std::result::Result<T, ProtocolError>;
|
|
|
48 |
|
|
12136
|
49 |
pub struct ProtocolDecoder {
|
|
15831
|
50 |
buffer: BytesMut,
|
|
15832
|
51 |
read_timeout: Duration,
|
|
14795
|
52 |
is_recovering: bool,
|
|
12129
|
53 |
}
|
|
|
54 |
|
|
12136
|
55 |
impl ProtocolDecoder {
|
|
15832
|
56 |
pub fn new(read_timeout: Duration) -> ProtocolDecoder {
|
|
12136
|
57 |
ProtocolDecoder {
|
|
15831
|
58 |
buffer: BytesMut::with_capacity(1024),
|
|
15832
|
59 |
read_timeout,
|
|
14795
|
60 |
is_recovering: false,
|
|
12129
|
61 |
}
|
|
|
62 |
}
|
|
|
63 |
|
|
14795
|
64 |
fn recover(&mut self) -> bool {
|
|
15831
|
65 |
self.is_recovering = match malformed_message(&self.buffer[..]) {
|
|
14795
|
66 |
Ok((tail, ())) => {
|
|
15831
|
67 |
let remaining = tail.len();
|
|
|
68 |
self.buffer.advance(self.buffer.len() - remaining);
|
|
14795
|
69 |
false
|
|
|
70 |
}
|
|
|
71 |
_ => {
|
|
15831
|
72 |
self.buffer.clear();
|
|
14795
|
73 |
true
|
|
|
74 |
}
|
|
|
75 |
};
|
|
|
76 |
!self.is_recovering
|
|
|
77 |
}
|
|
|
78 |
|
|
15831
|
79 |
fn extract_message(&mut self) -> Option<HwProtocolMessage> {
|
|
|
80 |
if !self.is_recovering || self.recover() {
|
|
|
81 |
match message(&self.buffer[..]) {
|
|
|
82 |
Ok((tail, message)) => {
|
|
|
83 |
let remaining = tail.len();
|
|
|
84 |
self.buffer.advance(self.buffer.len() - remaining);
|
|
|
85 |
return Some(message);
|
|
|
86 |
}
|
|
|
87 |
Err(nom::Err::Incomplete(_)) => {}
|
|
|
88 |
Err(nom::Err::Failure(e) | nom::Err::Error(e)) => {
|
|
|
89 |
debug!("Invalid message: {:?}", e);
|
|
|
90 |
self.recover();
|
|
14795
|
91 |
}
|
|
14457
|
92 |
}
|
|
12136
|
93 |
}
|
|
15831
|
94 |
None
|
|
|
95 |
}
|
|
|
96 |
|
|
|
97 |
pub async fn read_from<R: AsyncReadExt + Unpin>(
|
|
|
98 |
&mut self,
|
|
|
99 |
stream: &mut R,
|
|
15832
|
100 |
) -> Result<HwProtocolMessage> {
|
|
|
101 |
use ProtocolError::*;
|
|
|
102 |
|
|
15831
|
103 |
loop {
|
|
|
104 |
if !self.buffer.has_remaining() {
|
|
15832
|
105 |
match timeout(self.read_timeout, stream.read_buf(&mut self.buffer)).await {
|
|
|
106 |
Err(_) => return Err(Timeout),
|
|
|
107 |
Ok(Err(e)) => return Err(Network(Box::new(e))),
|
|
|
108 |
Ok(Ok(0)) => return Err(Eof),
|
|
|
109 |
Ok(Ok(_)) => (),
|
|
|
110 |
};
|
|
15831
|
111 |
}
|
|
|
112 |
while !self.buffer.is_empty() {
|
|
|
113 |
if let Some(result) = self.extract_message() {
|
|
15832
|
114 |
return Ok(result);
|
|
15831
|
115 |
}
|
|
|
116 |
}
|
|
|
117 |
}
|
|
12129
|
118 |
}
|
|
|
119 |
}
|