diff --git a/src/link/mod.rs b/src/link/mod.rs index 511473d..87d8533 100644 --- a/src/link/mod.rs +++ b/src/link/mod.rs @@ -6,18 +6,23 @@ 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 = 2; +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; -// Arbitrary value indicating that next byte sequence from ue-server reports amount of bytes -// received by that server so far. +// 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; -// Arbitrary value indicating that next byte sequence from ue-server contains JSON message. +// 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}", MessageTooLong{length: usize} = "Message to receive is too long: {length}", @@ -32,10 +37,21 @@ enum ReadingState { Payload, } +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, +} + /// 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 | 2 bytes: u16 BE] +/// [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] @@ -43,29 +59,17 @@ enum ReadingState { /// never recovers from it. /// Use either `push_byte()` or `push()` to input byte stream from ue-server and `pop()` to /// retrieve resulting messages. -pub struct MessageReader { - is_broken: bool, - reading_state: ReadingState, - read_bytes: usize, - length_buffer: [u8; 4], - current_message_length: usize, - current_message: Vec, - read_messages: VecDeque, - ue_received_bytes: u64, -} - impl MessageReader { pub fn new() -> MessageReader { MessageReader { is_broken: false, reading_state: ReadingState::Head, read_bytes: 0, - length_buffer: [0; 4], + buffer: [0; 4], current_message_length: 0, // Will be recreated with `with_capacity` in `push_byte()` current_message: Vec::new(), - // This value should be more than enough for typical use - read_messages: VecDeque::with_capacity(100), + read_messages: VecDeque::with_capacity(EXPECTED_LIMIT_TO_UE_MESSAGES), ue_received_bytes: 0, } } @@ -86,18 +90,19 @@ impl MessageReader { } } ReadingState::ReceivedBytes => { - self.length_buffer[self.read_bytes] = input; + self.buffer[self.read_bytes] = input; self.read_bytes += 1; if self.read_bytes >= UE_RECEIVED_FIELD_SIZE { - self.ue_received_bytes += array_of_u8_to_u16(self.length_buffer) as u64; + self.ue_received_bytes += array_of_u8_to_u32(self.buffer) as u64; self.change_state(ReadingState::Head); } } ReadingState::Length => { - self.length_buffer[self.read_bytes] = input; + self.buffer[self.read_bytes] = input; self.read_bytes += 1; if self.read_bytes >= UE_LENGTH_FIELD_SIZE { - self.current_message_length = array_of_u8_to_u32(self.length_buffer) as usize; + 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 { @@ -105,7 +110,6 @@ impl MessageReader { }); } self.current_message = Vec::with_capacity(self.current_message_length); - self.change_state(ReadingState::Payload); } } ReadingState::Payload => { @@ -153,15 +157,11 @@ impl MessageReader { } } -fn array_of_u8_to_u16(bytes: [u8; 4]) -> u16 { - (u16::from(bytes[0]) << 8) + u16::from(bytes[1]) -} - -fn array_of_u8_to_u32(bytes: [u8; 4]) -> u32 { - (u32::from(bytes[0]) << 24) - + (u32::from(bytes[1]) << 16) - + (u32::from(bytes[2]) << 8) - + (u32::from(bytes[3])) +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] @@ -203,15 +203,21 @@ fn received_push_byte() { let mut reader = MessageReader::new(); reader.push_byte(HEAD_UE_RECEIVED).unwrap(); reader.push_byte(0).unwrap(); - reader.push_byte(0xf3).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xf3); + 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(); - reader.push_byte(0xb2).unwrap(); - reader.push_byte(0x04).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3 + 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(); reader.push_byte(231).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); + reader.push_byte(34).unwrap(); + reader.push_byte(154).unwrap(); + assert_eq!(reader.ue_received_bytes(), 1092203255); } #[test] @@ -219,7 +225,9 @@ fn mixed_push_byte() { let mut reader = MessageReader::new(); reader.push_byte(HEAD_UE_RECEIVED).unwrap(); reader.push_byte(0).unwrap(); - reader.push_byte(0xf3).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(0).unwrap(); reader.push_byte(0).unwrap(); @@ -229,9 +237,11 @@ fn mixed_push_byte() { reader.push_byte(b'o').unwrap(); reader.push_byte(b'!').unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(0xb2).unwrap(); - reader.push_byte(0x04).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3 + 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.pop().unwrap(), "Yo!"); assert_eq!(reader.pop(), None); } @@ -243,7 +253,9 @@ fn pushing_many_bytes_at_once() { .push(&[ HEAD_UE_RECEIVED, 0, - 0xf3, + 0, + 0, + 243, HEAD_UE_MESSAGE, 0, 0, @@ -253,11 +265,13 @@ fn pushing_many_bytes_at_once() { b'o', b'!', HEAD_UE_RECEIVED, - 0xb2, - 0x04, + 65, + 25, + 178, + 4, ]) .unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); + assert_eq!(reader.ue_received_bytes(), 1092203255); assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop(), None); } @@ -267,7 +281,9 @@ fn generates_error_invalid_head() { let mut reader = MessageReader::new(); reader.push_byte(HEAD_UE_RECEIVED).unwrap(); reader.push_byte(0).unwrap(); - reader.push_byte(0xf3).unwrap(); + reader.push_byte(0).unwrap(); + reader.push_byte(0).unwrap(); + reader.push_byte(243).unwrap(); assert!(!reader.is_broken()); reader .push_byte(25) diff --git a/src/main.rs b/src/main.rs index 7f15ba3..604053c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,8 @@ fn main() { let args: Vec = env::args().collect(); let filename = &args[1]; let config = unreal_config::load_file(Path::new(filename)); - if let Ok(config) = config { - print!("{}", config); + match config { + Ok(config) => print!("{}", config), + _ => (), } }