dkanus/Avarice
2
1
Fork 1
Code Issues 9 Pull Requests 2 Releases Wiki Activity

Compare commits

base: dkanus:0a868521974fd84859506a3f9ef4d427bf05c2b9
Branches Tags
dkanus:master
dkanus:feature_link
dkanus:feature_database
...
compare: dkanus:bb1f73a755753221159eb247553b5b4d0c9c9042
Branches Tags
dkanus:feature_link
dkanus:feature_database
dkanus:master

5 Commits
0a86852197 ... bb1f73a755

Author SHA1 Message Date
Anton Tarasenko
bb1f73a755 Change character byte definitions into b'X' form 2021-07-22 18:54:12 +07:00
Anton Tarasenko
af3341c1e7 Refactor MessageReader for clarity 
Rename `received_bytes` into `ue_received_bytes` and get rid of
`next_received_bytes` by reading relevant data into new byte buffer
instead.
 2021-07-22 18:48:09 +07:00
Anton Tarasenko
0dedd1d1f1 Change MessageReader to use with_capacity 
Some of the collections inside `MessageReader` were created with `new()`
instead of `with_capacity()` call. This patch fixes that or comments why
it was not done in some places.
 2021-07-22 18:18:29 +07:00
Anton Tarasenko
b846fcf55b Fix MessageReader documentation 2021-07-22 18:11:44 +07:00
Anton Tarasenko
ba3ac088dd Refactor numeric constants for MessageReader 2021-07-22 18:03:48 +07:00
1 changed files with 110 additions and 84 deletions
Show Stats Expand all files Collapse all files

194
src/link/mod.rs
View File

@ -4,11 +4,24 @@ use std::str;
extern crate custom_error; extern crate custom_error;
use custom_error::custom_error; use custom_error::custom_error;
const RECEIVED_FIELD_SIZE: usize = 4; // Defines how many bytes is used to encode "AMOUNT" field in the response from ue-server about
const LENGTH_FIELD_SIZE: usize = 4; // amount of bytes it received since the last update
const HEAD_RECEIVED: u8 = 85; const UE_RECEIVED_FIELD_SIZE: usize = 4;
const HEAD_MESSAGE: u8 = 42; // Defines how many bytes is used to encode "LENGTH" field, describing length of
const MAX_MESSAGE_LENGTH: usize = 0xA00000; // next JSON message from ue-server
const UE_LENGTH_FIELD_SIZE: usize = 4;
// Value indicating that next byte sequence from ue-server reports amount of bytes received by
// that server so far. Value itself is arbitrary.
const HEAD_UE_RECEIVED: u8 = 85;
// Value indicating that next byte sequence from ue-server contains JSON message.
// Value itself is arbitrary.
const HEAD_UE_MESSAGE: u8 = 42;
// Maximum allowed size of JSON message sent from ue-server.
const MAX_UE_MESSAGE_LENGTH: usize = 25 * 1024 * 1024;
// We do not expect to receive more that this much messages at once from ue-server
const EXPECTED_LIMIT_TO_UE_MESSAGES: usize = 100;
custom_error! { pub ReadingStreamError custom_error! { pub ReadingStreamError
InvalidHead{input: u8} = "Invalid byte used as a HEAD: {input}", InvalidHead{input: u8} = "Invalid byte used as a HEAD: {input}",
@ -28,20 +41,23 @@ pub struct MessageReader {
is_broken: bool, is_broken: bool,
reading_state: ReadingState, reading_state: ReadingState,
read_bytes: usize, read_bytes: usize,
buffer: [u8; 4],
current_message_length: usize, current_message_length: usize,
current_message: Vec<u8>, current_message: Vec<u8>,
read_messages: VecDeque<String>, read_messages: VecDeque<String>,
next_received_bytes: u32, ue_received_bytes: u64,
received_bytes: u64,
} }
/// For converting byte stream expected to be generated by Acedia mod from the game server into /// For converting byte stream that is expected from the ue-server into actual messages.
/// actual messages. Expected format is a sequence of either: /// Expected format is a sequence of either:
/// [HEAD_RECEIVED: 1 byte] [amount of bytes received by game server since last update: 4 bytes] /// 1. [HEAD_UE_RECEIVED: marker byte | 1 byte]
/// [HEAD_MESSAGE: 1 byte] [length of the message: 4 bytes] [utf8-encoded string: ??? bytes] /// [AMOUNT: amount of bytes received by ue-server since last update | 4 bytes: u32 BE]
/// On any invalid input enters a failure state (can be checked by `is_broken()`) and /// 2. [HEAD_UE_MESSAGE: marker byte | 1 byte]
/// [LENGTH: length of the JSON message in utf8 encoding | 4 bytes: u32 BE]
/// [PAYLOAD: utf8-encoded string | `LENGTH` bytes]
/// On any invalid input enters into a failure state (can be checked by `is_broken()`) and
/// never recovers from it. /// never recovers from it.
/// Use either `push_byte()` or `push()` to input byte stream from game server and `pop()` to /// Use either `push_byte()` or `push()` to input byte stream from ue-server and `pop()` to
/// retrieve resulting messages. /// retrieve resulting messages.
impl MessageReader { impl MessageReader {
pub fn new() -> MessageReader { pub fn new() -> MessageReader {
@ -49,11 +65,12 @@ impl MessageReader {
is_broken: false, is_broken: false,
reading_state: ReadingState::Head, reading_state: ReadingState::Head,
read_bytes: 0, read_bytes: 0,
buffer: [0; 4],
current_message_length: 0, current_message_length: 0,
// Will be recreated with `with_capacity` in `push_byte()`
current_message: Vec::new(), current_message: Vec::new(),
read_messages: VecDeque::new(), read_messages: VecDeque::with_capacity(EXPECTED_LIMIT_TO_UE_MESSAGES),
next_received_bytes: 0, ue_received_bytes: 0,
received_bytes: 0,
} }
} }
@ -63,34 +80,30 @@ impl MessageReader {
} }
match &self.reading_state { match &self.reading_state {
ReadingState::Head => { ReadingState::Head => {
if input == HEAD_RECEIVED { if input == HEAD_UE_RECEIVED {
self.reading_state = ReadingState::ReceivedBytes; self.change_state(ReadingState::ReceivedBytes);
} else if input == HEAD_MESSAGE { } else if input == HEAD_UE_MESSAGE {
self.reading_state = ReadingState::Length; self.change_state(ReadingState::Length);
} else { } else {
self.is_broken = true; self.is_broken = true;
return Err(ReadingStreamError::InvalidHead { input }); return Err(ReadingStreamError::InvalidHead { input });
} }
} }
ReadingState::ReceivedBytes => { ReadingState::ReceivedBytes => {
self.next_received_bytes = self.next_received_bytes << 8; self.buffer[self.read_bytes] = input;
self.next_received_bytes += input as u32;
self.read_bytes += 1; self.read_bytes += 1;
if self.read_bytes >= RECEIVED_FIELD_SIZE { if self.read_bytes >= UE_RECEIVED_FIELD_SIZE {
self.received_bytes += self.next_received_bytes as u64; self.ue_received_bytes += array_of_u8_to_u32(self.buffer) as u64;
self.next_received_bytes = 0; self.change_state(ReadingState::Head);
self.read_bytes = 0;
self.reading_state = ReadingState::Head;
} }
} }
ReadingState::Length => { ReadingState::Length => {
self.current_message_length = self.current_message_length << 8; self.buffer[self.read_bytes] = input;
self.current_message_length += input as usize;
self.read_bytes += 1; self.read_bytes += 1;
if self.read_bytes >= LENGTH_FIELD_SIZE { if self.read_bytes >= UE_LENGTH_FIELD_SIZE {
self.read_bytes = 0; self.current_message_length = array_of_u8_to_u32(self.buffer) as usize;
self.reading_state = ReadingState::Payload; self.change_state(ReadingState::Payload);
if self.current_message_length > MAX_MESSAGE_LENGTH { if self.current_message_length > MAX_UE_MESSAGE_LENGTH {
self.is_broken = true; self.is_broken = true;
return Err(ReadingStreamError::MessageTooLong { return Err(ReadingStreamError::MessageTooLong {
length: self.current_message_length, length: self.current_message_length,
@ -101,7 +114,7 @@ impl MessageReader {
} }
ReadingState::Payload => { ReadingState::Payload => {
self.current_message.push(input); self.current_message.push(input);
self.read_bytes += 1 as usize; self.read_bytes += 1;
if self.read_bytes >= self.current_message_length { if self.read_bytes >= self.current_message_length {
match str::from_utf8(&self.current_message) { match str::from_utf8(&self.current_message) {
Ok(next_message) => self.read_messages.push_front(next_message.to_owned()), Ok(next_message) => self.read_messages.push_front(next_message.to_owned()),
@ -112,8 +125,7 @@ impl MessageReader {
}; };
self.current_message.clear(); self.current_message.clear();
self.current_message_length = 0; self.current_message_length = 0;
self.read_bytes = 0; self.change_state(ReadingState::Head);
self.reading_state = ReadingState::Head;
} }
} }
} }
@ -131,44 +143,56 @@ impl MessageReader {
self.read_messages.pop_back() self.read_messages.pop_back()
} }
pub fn received_bytes(&self) -> u64 { pub fn ue_received_bytes(&self) -> u64 {
self.received_bytes self.ue_received_bytes
} }
pub fn is_broken(&self) -> bool { pub fn is_broken(&self) -> bool {
self.is_broken self.is_broken
} }
fn change_state(&mut self, next_state: ReadingState) {
self.read_bytes = 0;
self.reading_state = next_state;
}
}
fn array_of_u8_to_u32(bytes: [u8; 4]) -> u64 {
(u64::from(bytes[0]) << 24)
+ (u64::from(bytes[1]) << 16)
+ (u64::from(bytes[2]) << 8)
+ (u64::from(bytes[3]))
} }
#[test] #[test]
fn message_push_byte() { fn message_push_byte() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_MESSAGE).unwrap(); reader.push_byte(HEAD_UE_MESSAGE).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(13).unwrap(); reader.push_byte(13).unwrap();
reader.push_byte(0x48).unwrap(); // H reader.push_byte(b'H').unwrap();
reader.push_byte(0x65).unwrap(); // e reader.push_byte(b'e').unwrap();
reader.push_byte(0x6c).unwrap(); // l reader.push_byte(b'l').unwrap();
reader.push_byte(0x6c).unwrap(); // l reader.push_byte(b'l').unwrap();
reader.push_byte(0x6f).unwrap(); // o reader.push_byte(b'o').unwrap();
reader.push_byte(0x2c).unwrap(); // , reader.push_byte(b',').unwrap();
reader.push_byte(0x20).unwrap(); // <space> reader.push_byte(b' ').unwrap();
reader.push_byte(0x77).unwrap(); // w reader.push_byte(b'w').unwrap();
reader.push_byte(0x6f).unwrap(); // o reader.push_byte(b'o').unwrap();
reader.push_byte(0x72).unwrap(); // r reader.push_byte(b'r').unwrap();
reader.push_byte(0x6c).unwrap(); // l reader.push_byte(b'l').unwrap();
reader.push_byte(0x64).unwrap(); // d reader.push_byte(b'd').unwrap();
reader.push_byte(0x21).unwrap(); // <exclamation mark> reader.push_byte(b'!').unwrap();
reader.push_byte(HEAD_MESSAGE).unwrap(); reader.push_byte(HEAD_UE_MESSAGE).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(3).unwrap(); reader.push_byte(3).unwrap();
reader.push_byte(0x59).unwrap(); // Y reader.push_byte(b'Y').unwrap();
reader.push_byte(0x6f).unwrap(); // o reader.push_byte(b'o').unwrap();
reader.push_byte(0x21).unwrap(); // <exclamation mark> reader.push_byte(b'!').unwrap();
assert_eq!(reader.pop().unwrap(), "Hello, world!"); assert_eq!(reader.pop().unwrap(), "Hello, world!");
assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop().unwrap(), "Yo!");
assert_eq!(reader.pop(), None); assert_eq!(reader.pop(), None);
@ -177,47 +201,47 @@ fn message_push_byte() {
#[test] #[test]
fn received_push_byte() { fn received_push_byte() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(243).unwrap(); reader.push_byte(243).unwrap();
assert_eq!(reader.received_bytes(), 243); assert_eq!(reader.ue_received_bytes(), 243);
reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(65).unwrap(); reader.push_byte(65).unwrap();
reader.push_byte(25).unwrap(); reader.push_byte(25).unwrap();
reader.push_byte(178).unwrap(); reader.push_byte(178).unwrap();
reader.push_byte(4).unwrap(); reader.push_byte(4).unwrap();
assert_eq!(reader.received_bytes(), 1092203255); assert_eq!(reader.ue_received_bytes(), 1092203255);
reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(231).unwrap(); reader.push_byte(231).unwrap();
reader.push_byte(34).unwrap(); reader.push_byte(34).unwrap();
reader.push_byte(154).unwrap(); reader.push_byte(154).unwrap();
assert_eq!(reader.received_bytes(), 1092203255); assert_eq!(reader.ue_received_bytes(), 1092203255);
} }
#[test] #[test]
fn mixed_push_byte() { fn mixed_push_byte() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(243).unwrap(); reader.push_byte(243).unwrap();
reader.push_byte(HEAD_MESSAGE).unwrap(); reader.push_byte(HEAD_UE_MESSAGE).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(3).unwrap(); reader.push_byte(3).unwrap();
reader.push_byte(0x59).unwrap(); // Y reader.push_byte(b'Y').unwrap();
reader.push_byte(0x6f).unwrap(); // o reader.push_byte(b'o').unwrap();
reader.push_byte(0x21).unwrap(); // <exclamation mark> reader.push_byte(b'!').unwrap();
reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(65).unwrap(); reader.push_byte(65).unwrap();
reader.push_byte(25).unwrap(); reader.push_byte(25).unwrap();
reader.push_byte(178).unwrap(); reader.push_byte(178).unwrap();
reader.push_byte(4).unwrap(); reader.push_byte(4).unwrap();
assert_eq!(reader.received_bytes(), 1092203255); assert_eq!(reader.ue_received_bytes(), 1092203255);
assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop().unwrap(), "Yo!");
assert_eq!(reader.pop(), None); assert_eq!(reader.pop(), None);
} }
@ -227,27 +251,27 @@ fn pushing_many_bytes_at_once() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader reader
.push(&[ .push(&[
HEAD_RECEIVED, HEAD_UE_RECEIVED,
0, 0,
0, 0,
0, 0,
243, 243,
HEAD_MESSAGE, HEAD_UE_MESSAGE,
0, 0,
0, 0,
0, 0,
3, 3,
0x59, // Y b'Y',
0x6f, // o b'o',
0x21, // <exclamation mark> b'!',
HEAD_RECEIVED, HEAD_UE_RECEIVED,
65, 65,
25, 25,
178, 178,
4, 4,
]) ])
.unwrap(); .unwrap();
assert_eq!(reader.received_bytes(), 1092203255); assert_eq!(reader.ue_received_bytes(), 1092203255);
assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop().unwrap(), "Yo!");
assert_eq!(reader.pop(), None); assert_eq!(reader.pop(), None);
} }
@ -255,7 +279,7 @@ fn pushing_many_bytes_at_once() {
#[test] #[test]
fn generates_error_invalid_head() { fn generates_error_invalid_head() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
@ -270,14 +294,16 @@ fn generates_error_invalid_head() {
#[test] #[test]
fn generates_error_message_too_long() { fn generates_error_message_too_long() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_MESSAGE).unwrap(); let huge_length = MAX_UE_MESSAGE_LENGTH + 1;
// Max length right now is `0xA00000` let bytes = (huge_length as u32).to_be_bytes();
reader.push_byte(0).unwrap();
reader.push_byte(0xA0).unwrap(); reader.push_byte(HEAD_UE_MESSAGE).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(bytes[0]).unwrap();
reader.push_byte(bytes[1]).unwrap();
reader.push_byte(bytes[2]).unwrap();
assert!(!reader.is_broken()); assert!(!reader.is_broken());
reader reader
.push_byte(1) .push_byte(bytes[3])
.expect_err("Testing failing on exceeding allowed message length"); .expect_err("Testing failing on exceeding allowed message length");
assert!(reader.is_broken()); assert!(reader.is_broken());
} }
@ -285,7 +311,7 @@ fn generates_error_message_too_long() {
#[test] #[test]
fn generates_error_invalid_unicode() { fn generates_error_invalid_unicode() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_MESSAGE).unwrap(); reader.push_byte(HEAD_UE_MESSAGE).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();
reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();