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