Change character byte definitions into b'X' form

Rename `received_bytes` into `ue_received_bytes` and get rid of
`next_received_bytes` by reading relevant data into new byte buffer
instead.

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<u8>,
     read_messages: VecDeque<String>,
-    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(); // <space>
-    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(); // <exclamation mark>
-    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(); // <exclamation mark>
+    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(); // <exclamation mark>
-    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, // <exclamation mark>
-            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();