Compare commits

..

8 Commits

Author SHA1 Message Date
b187041d9e Move documentation for MessageReader 2021-07-23 02:02:05 +07:00
3f660f54d5 Remove EXPECTED_LIMIT_TO_UE_MESSAGES constant 2021-07-23 02:01:17 +07:00
5ce511c5a7 Refactor main.rs to use if let construction 2021-07-23 01:50:47 +07:00
978a5c4182 Change UE_RECEIVED_FIELD_SIZE to 2
ue-server is only supposed to receive up to 4095 bytes at once and is
expected to report after reading them, therefore 4 bytes for the amount
of received bytes is excessive and 2 will suffice.
2021-07-23 01:48:42 +07:00
e3f554218a Refactor push_byte's code
Move logic below error checking with early exit.
2021-07-23 01:31:40 +07:00
9ca00c5b8c Refactor tests for clarity 2021-07-23 01:28:33 +07:00
f118767e3f Rename buffer into length_buffer 2021-07-23 01:23:13 +07:00
202fa39a42 Fix comments for some MessageReader contants 2021-07-23 01:22:29 +07:00
2 changed files with 50 additions and 67 deletions

View File

@ -6,23 +6,18 @@ use custom_error::custom_error;
// Defines how many bytes is used to encode "AMOUNT" field in the response from ue-server about // 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 // amount of bytes it received since the last update
const UE_RECEIVED_FIELD_SIZE: usize = 4; const UE_RECEIVED_FIELD_SIZE: usize = 2;
// Defines how many bytes is used to encode "LENGTH" field, describing length of // Defines how many bytes is used to encode "LENGTH" field, describing length of
// next JSON message from ue-server // next JSON message from ue-server
const UE_LENGTH_FIELD_SIZE: usize = 4; const UE_LENGTH_FIELD_SIZE: usize = 4;
// Value indicating that next byte sequence from ue-server reports amount of bytes received by // Arbitrary value indicating that next byte sequence from ue-server reports amount of bytes
// that server so far. Value itself is arbitrary. // received by that server so far.
const HEAD_UE_RECEIVED: u8 = 85; const HEAD_UE_RECEIVED: u8 = 85;
// Value indicating that next byte sequence from ue-server contains JSON message. // Arbitrary value indicating that next byte sequence from ue-server contains JSON message.
// Value itself is arbitrary.
const HEAD_UE_MESSAGE: u8 = 42; const HEAD_UE_MESSAGE: u8 = 42;
// Maximum allowed size of JSON message sent from ue-server. // Maximum allowed size of JSON message sent from ue-server.
const MAX_UE_MESSAGE_LENGTH: usize = 25 * 1024 * 1024; 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}",
MessageTooLong{length: usize} = "Message to receive is too long: {length}", MessageTooLong{length: usize} = "Message to receive is too long: {length}",
@ -37,21 +32,10 @@ enum ReadingState {
Payload, Payload,
} }
pub struct MessageReader {
is_broken: bool,
reading_state: ReadingState,
read_bytes: usize,
buffer: [u8; 4],
current_message_length: usize,
current_message: Vec<u8>,
read_messages: VecDeque<String>,
ue_received_bytes: u64,
}
/// For converting byte stream that is expected from the ue-server into actual messages. /// For converting byte stream that is expected from the ue-server into actual messages.
/// Expected format is a sequence of either: /// Expected format is a sequence of either:
/// 1. [HEAD_UE_RECEIVED: marker byte | 1 byte] /// 1. [HEAD_UE_RECEIVED: marker byte | 1 byte]
/// [AMOUNT: amount of bytes received by ue-server since last update | 4 bytes: u32 BE] /// [AMOUNT: amount of bytes received by ue-server since last update | 2 bytes: u16 BE]
/// 2. [HEAD_UE_MESSAGE: marker byte | 1 byte] /// 2. [HEAD_UE_MESSAGE: marker byte | 1 byte]
/// [LENGTH: length of the JSON message in utf8 encoding | 4 bytes: u32 BE] /// [LENGTH: length of the JSON message in utf8 encoding | 4 bytes: u32 BE]
/// [PAYLOAD: utf8-encoded string | `LENGTH` bytes] /// [PAYLOAD: utf8-encoded string | `LENGTH` bytes]
@ -59,17 +43,29 @@ pub struct MessageReader {
/// never recovers from it. /// 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 ue-server and `pop()` to
/// retrieve resulting messages. /// 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<u8>,
read_messages: VecDeque<String>,
ue_received_bytes: u64,
}
impl MessageReader { impl MessageReader {
pub fn new() -> MessageReader { pub fn new() -> MessageReader {
MessageReader { MessageReader {
is_broken: false, is_broken: false,
reading_state: ReadingState::Head, reading_state: ReadingState::Head,
read_bytes: 0, read_bytes: 0,
buffer: [0; 4], length_buffer: [0; 4],
current_message_length: 0, current_message_length: 0,
// Will be recreated with `with_capacity` in `push_byte()` // Will be recreated with `with_capacity` in `push_byte()`
current_message: Vec::new(), current_message: Vec::new(),
read_messages: VecDeque::with_capacity(EXPECTED_LIMIT_TO_UE_MESSAGES), // This value should be more than enough for typical use
read_messages: VecDeque::with_capacity(100),
ue_received_bytes: 0, ue_received_bytes: 0,
} }
} }
@ -90,19 +86,18 @@ impl MessageReader {
} }
} }
ReadingState::ReceivedBytes => { ReadingState::ReceivedBytes => {
self.buffer[self.read_bytes] = input; self.length_buffer[self.read_bytes] = input;
self.read_bytes += 1; self.read_bytes += 1;
if self.read_bytes >= UE_RECEIVED_FIELD_SIZE { if self.read_bytes >= UE_RECEIVED_FIELD_SIZE {
self.ue_received_bytes += array_of_u8_to_u32(self.buffer) as u64; self.ue_received_bytes += array_of_u8_to_u16(self.length_buffer) as u64;
self.change_state(ReadingState::Head); self.change_state(ReadingState::Head);
} }
} }
ReadingState::Length => { ReadingState::Length => {
self.buffer[self.read_bytes] = input; self.length_buffer[self.read_bytes] = input;
self.read_bytes += 1; self.read_bytes += 1;
if self.read_bytes >= UE_LENGTH_FIELD_SIZE { if self.read_bytes >= UE_LENGTH_FIELD_SIZE {
self.current_message_length = array_of_u8_to_u32(self.buffer) as usize; self.current_message_length = array_of_u8_to_u32(self.length_buffer) as usize;
self.change_state(ReadingState::Payload);
if self.current_message_length > MAX_UE_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 {
@ -110,6 +105,7 @@ impl MessageReader {
}); });
} }
self.current_message = Vec::with_capacity(self.current_message_length); self.current_message = Vec::with_capacity(self.current_message_length);
self.change_state(ReadingState::Payload);
} }
} }
ReadingState::Payload => { ReadingState::Payload => {
@ -157,11 +153,15 @@ impl MessageReader {
} }
} }
fn array_of_u8_to_u32(bytes: [u8; 4]) -> u64 { fn array_of_u8_to_u16(bytes: [u8; 4]) -> u16 {
(u64::from(bytes[0]) << 24) (u16::from(bytes[0]) << 8) + u16::from(bytes[1])
+ (u64::from(bytes[1]) << 16) }
+ (u64::from(bytes[2]) << 8)
+ (u64::from(bytes[3])) 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]))
} }
#[test] #[test]
@ -203,21 +203,15 @@ fn received_push_byte() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_UE_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(0xf3).unwrap();
reader.push_byte(0).unwrap(); assert_eq!(reader.ue_received_bytes(), 0xf3);
reader.push_byte(243).unwrap();
assert_eq!(reader.ue_received_bytes(), 243);
reader.push_byte(HEAD_UE_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(65).unwrap(); reader.push_byte(0xb2).unwrap();
reader.push_byte(25).unwrap(); reader.push_byte(0x04).unwrap();
reader.push_byte(178).unwrap(); assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3
reader.push_byte(4).unwrap();
assert_eq!(reader.ue_received_bytes(), 1092203255);
reader.push_byte(HEAD_UE_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(231).unwrap(); reader.push_byte(231).unwrap();
reader.push_byte(34).unwrap(); assert_eq!(reader.ue_received_bytes(), 0xb2_f7);
reader.push_byte(154).unwrap();
assert_eq!(reader.ue_received_bytes(), 1092203255);
} }
#[test] #[test]
@ -225,9 +219,7 @@ fn mixed_push_byte() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_UE_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(0xf3).unwrap();
reader.push_byte(0).unwrap();
reader.push_byte(243).unwrap();
reader.push_byte(HEAD_UE_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();
@ -237,11 +229,9 @@ fn mixed_push_byte() {
reader.push_byte(b'o').unwrap(); reader.push_byte(b'o').unwrap();
reader.push_byte(b'!').unwrap(); reader.push_byte(b'!').unwrap();
reader.push_byte(HEAD_UE_RECEIVED).unwrap(); reader.push_byte(HEAD_UE_RECEIVED).unwrap();
reader.push_byte(65).unwrap(); reader.push_byte(0xb2).unwrap();
reader.push_byte(25).unwrap(); reader.push_byte(0x04).unwrap();
reader.push_byte(178).unwrap(); assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3
reader.push_byte(4).unwrap();
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);
} }
@ -253,9 +243,7 @@ fn pushing_many_bytes_at_once() {
.push(&[ .push(&[
HEAD_UE_RECEIVED, HEAD_UE_RECEIVED,
0, 0,
0, 0xf3,
0,
243,
HEAD_UE_MESSAGE, HEAD_UE_MESSAGE,
0, 0,
0, 0,
@ -265,13 +253,11 @@ fn pushing_many_bytes_at_once() {
b'o', b'o',
b'!', b'!',
HEAD_UE_RECEIVED, HEAD_UE_RECEIVED,
65, 0xb2,
25, 0x04,
178,
4,
]) ])
.unwrap(); .unwrap();
assert_eq!(reader.ue_received_bytes(), 1092203255); assert_eq!(reader.ue_received_bytes(), 0xb2_f7);
assert_eq!(reader.pop().unwrap(), "Yo!"); assert_eq!(reader.pop().unwrap(), "Yo!");
assert_eq!(reader.pop(), None); assert_eq!(reader.pop(), None);
} }
@ -281,9 +267,7 @@ fn generates_error_invalid_head() {
let mut reader = MessageReader::new(); let mut reader = MessageReader::new();
reader.push_byte(HEAD_UE_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(0xf3).unwrap();
reader.push_byte(0).unwrap();
reader.push_byte(243).unwrap();
assert!(!reader.is_broken()); assert!(!reader.is_broken());
reader reader
.push_byte(25) .push_byte(25)

View File

@ -7,8 +7,7 @@ fn main() {
let args: Vec<String> = env::args().collect(); let args: Vec<String> = env::args().collect();
let filename = &args[1]; let filename = &args[1];
let config = unreal_config::load_file(Path::new(filename)); let config = unreal_config::load_file(Path::new(filename));
match config { if let Ok(config) = config {
Ok(config) => print!("{}", config), print!("{}", config);
_ => (),
} }
} }