diff --git a/src/link/mod.rs b/src/link/mod.rs index 87d8533..0a317a6 100644 --- a/src/link/mod.rs +++ b/src/link/mod.rs @@ -4,24 +4,11 @@ use std::str; extern crate custom_error; use custom_error::custom_error; -// 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; +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; custom_error! { pub ReadingStreamError InvalidHead{input: u8} = "Invalid byte used as a HEAD: {input}", @@ -41,23 +28,20 @@ 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, - ue_received_bytes: u64, + next_received_bytes: u32, + received_bytes: u64, } -/// 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 +/// 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 /// never recovers from it. -/// Use either `push_byte()` or `push()` to input byte stream from ue-server and `pop()` to +/// Use either `push_byte()` or `push()` to input byte stream from game server and `pop()` to /// retrieve resulting messages. impl MessageReader { pub fn new() -> MessageReader { @@ -65,12 +49,11 @@ 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::with_capacity(EXPECTED_LIMIT_TO_UE_MESSAGES), - ue_received_bytes: 0, + read_messages: VecDeque::new(), + next_received_bytes: 0, + received_bytes: 0, } } @@ -80,30 +63,34 @@ impl MessageReader { } match &self.reading_state { ReadingState::Head => { - if input == HEAD_UE_RECEIVED { - self.change_state(ReadingState::ReceivedBytes); - } else if input == HEAD_UE_MESSAGE { - self.change_state(ReadingState::Length); + if input == HEAD_RECEIVED { + self.reading_state = ReadingState::ReceivedBytes; + } else if input == HEAD_MESSAGE { + self.reading_state = ReadingState::Length; } else { self.is_broken = true; return Err(ReadingStreamError::InvalidHead { input }); } } ReadingState::ReceivedBytes => { - self.buffer[self.read_bytes] = input; + self.next_received_bytes = self.next_received_bytes << 8; + self.next_received_bytes += input as u32; self.read_bytes += 1; - 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); + 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; } } ReadingState::Length => { - self.buffer[self.read_bytes] = input; + self.current_message_length = self.current_message_length << 8; + self.current_message_length += input as usize; self.read_bytes += 1; - 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 { + if self.read_bytes >= LENGTH_FIELD_SIZE { + self.read_bytes = 0; + self.reading_state = ReadingState::Payload; + if self.current_message_length > MAX_MESSAGE_LENGTH { self.is_broken = true; return Err(ReadingStreamError::MessageTooLong { length: self.current_message_length, @@ -114,7 +101,7 @@ impl MessageReader { } ReadingState::Payload => { self.current_message.push(input); - self.read_bytes += 1; + self.read_bytes += 1 as usize; 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()), @@ -125,7 +112,8 @@ impl MessageReader { }; self.current_message.clear(); self.current_message_length = 0; - self.change_state(ReadingState::Head); + self.read_bytes = 0; + self.reading_state = ReadingState::Head; } } } @@ -143,56 +131,44 @@ impl MessageReader { self.read_messages.pop_back() } - pub fn ue_received_bytes(&self) -> u64 { - self.ue_received_bytes + pub fn received_bytes(&self) -> u64 { + self.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_UE_MESSAGE).unwrap(); + reader.push_byte(HEAD_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(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(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(0).unwrap(); reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap(); reader.push_byte(3).unwrap(); - reader.push_byte(b'Y').unwrap(); - reader.push_byte(b'o').unwrap(); - reader.push_byte(b'!').unwrap(); + reader.push_byte(0x59).unwrap(); // Y + reader.push_byte(0x6f).unwrap(); // o + reader.push_byte(0x21).unwrap(); // assert_eq!(reader.pop().unwrap(), "Hello, world!"); assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop(), None); @@ -201,47 +177,47 @@ fn message_push_byte() { #[test] fn received_push_byte() { let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(HEAD_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.ue_received_bytes(), 243); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + assert_eq!(reader.received_bytes(), 243); + reader.push_byte(HEAD_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.ue_received_bytes(), 1092203255); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + assert_eq!(reader.received_bytes(), 1092203255); + reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(231).unwrap(); reader.push_byte(34).unwrap(); reader.push_byte(154).unwrap(); - assert_eq!(reader.ue_received_bytes(), 1092203255); + assert_eq!(reader.received_bytes(), 1092203255); } #[test] fn mixed_push_byte() { let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(HEAD_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_UE_MESSAGE).unwrap(); + reader.push_byte(HEAD_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(b'Y').unwrap(); - reader.push_byte(b'o').unwrap(); - reader.push_byte(b'!').unwrap(); - reader.push_byte(HEAD_UE_RECEIVED).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(65).unwrap(); reader.push_byte(25).unwrap(); reader.push_byte(178).unwrap(); reader.push_byte(4).unwrap(); - assert_eq!(reader.ue_received_bytes(), 1092203255); + assert_eq!(reader.received_bytes(), 1092203255); assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop(), None); } @@ -251,27 +227,27 @@ fn pushing_many_bytes_at_once() { let mut reader = MessageReader::new(); reader .push(&[ - HEAD_UE_RECEIVED, + HEAD_RECEIVED, 0, 0, 0, 243, - HEAD_UE_MESSAGE, + HEAD_MESSAGE, 0, 0, 0, 3, - b'Y', - b'o', - b'!', - HEAD_UE_RECEIVED, + 0x59, // Y + 0x6f, // o + 0x21, // + HEAD_RECEIVED, 65, 25, 178, 4, ]) .unwrap(); - assert_eq!(reader.ue_received_bytes(), 1092203255); + assert_eq!(reader.received_bytes(), 1092203255); assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop(), None); } @@ -279,7 +255,7 @@ fn pushing_many_bytes_at_once() { #[test] fn generates_error_invalid_head() { let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(HEAD_RECEIVED).unwrap(); reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap(); @@ -294,16 +270,14 @@ fn generates_error_invalid_head() { #[test] fn generates_error_message_too_long() { let mut reader = MessageReader::new(); - 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(); + 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(); assert!(!reader.is_broken()); reader - .push_byte(bytes[3]) + .push_byte(1) .expect_err("Testing failing on exceeding allowed message length"); assert!(reader.is_broken()); } @@ -311,7 +285,7 @@ fn generates_error_message_too_long() { #[test] fn generates_error_invalid_unicode() { let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_MESSAGE).unwrap(); + reader.push_byte(HEAD_MESSAGE).unwrap(); reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap(); reader.push_byte(0).unwrap();