I've decided that it would be a good idea to separete
compoments that transform text messages into byte chunks to send to
Unreal Engine and back. This means that each of them will have only one
responsibility and using them will be easier in case we decide to read
and write via tcp/ip in separate threads. The only interesting to them
both parameter is the amount of bytes that were confirmed to be received
from the Unreal Engine side, which can be easily shared "by hand".

This patch implements "reader" that accepts byte stream expected from
the game server and converts it into:
1. separate string messages;
2. amount of bytes game server reported to already have received.

@@ -0,0 +5,4 @@
+use custom_error::custom_error;
+
+const RECEIVED_FIELD_SIZE: usize = 4;
+const LENGTH_FIELD_SIZE: usize = 4;

@@ -0,0 +7,4 @@
+const RECEIVED_FIELD_SIZE: usize = 4;
+const LENGTH_FIELD_SIZE: usize = 4;
+const HEAD_RECEIVED: u8 = 85;
+const HEAD_MESSAGE: u8 = 42;

@@ -0,0 +8,4 @@
+const LENGTH_FIELD_SIZE: usize = 4;
+const HEAD_RECEIVED: u8 = 85;
+const HEAD_MESSAGE: u8 = 42;
+const MAX_MESSAGE_LENGTH: usize = 0xA00000;

@@ -0,0 +35,4 @@
+    received_bytes: u64,
+}
+
+///     For converting byte stream expected to be generated by Acedia mod from the game server into

@@ -0,0 +37,4 @@
+
+///     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]

@@ -0,0 +50,4 @@
+            reading_state: ReadingState::Head,
+            read_bytes: 0,
+            current_message_length: 0,
+            current_message: Vec::new(),

@@ -0,0 +51,4 @@
+            read_bytes: 0,
+            current_message_length: 0,
+            current_message: Vec::new(),
+            read_messages: VecDeque::new(),

@@ -0,0 +73,4 @@
+                }
+            }
+            ReadingState::ReceivedBytes => {
+                self.next_received_bytes = self.next_received_bytes << 8;

@@ -0,0 +77,4 @@
+                self.next_received_bytes += input as u32;
+                self.read_bytes += 1;
+                if self.read_bytes >= RECEIVED_FIELD_SIZE {
+                    self.received_bytes += self.next_received_bytes as u64;

@@ -0,0 +101,4 @@
+            }
+            ReadingState::Payload => {
+                self.current_message.push(input);
+                self.read_bytes += 1 as usize;

@@ -0,0 +31,4 @@
+    current_message_length: usize,
+    current_message: Vec<u8>,
+    read_messages: VecDeque<String>,
+    next_received_bytes: u32,

@@ -0,0 +148,4 @@
+    reader.push_byte(0).unwrap();
+    reader.push_byte(0).unwrap();
+    reader.push_byte(13).unwrap();
+    reader.push_byte(0x48).unwrap(); // H

@@ -0,0 +149,4 @@
+    reader.push_byte(0).unwrap();
+    reader.push_byte(13).unwrap();
+    reader.push_byte(0x48).unwrap(); // H
+    reader.push_byte(0x65).unwrap(); // e

@@ -0,0 +188,4 @@
+    reader.push_byte(25).unwrap();
+    reader.push_byte(178).unwrap();
+    reader.push_byte(4).unwrap();
+    assert_eq!(reader.received_bytes(), 1092203255);

@@ -0,0 +21,4 @@
+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;

@@ -0,0 +11,4 @@
+//  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.

@@ -0,0 +41,4 @@
+    is_broken: bool,
+    reading_state: ReadingState,
+    read_bytes: usize,
+    buffer: [u8; 4],

@@ -0,0 +212,4 @@
+    reader.push_byte(25).unwrap();
+    reader.push_byte(178).unwrap();
+    reader.push_byte(4).unwrap();
+    assert_eq!(reader.ue_received_bytes(), 1092203255);

@@ -0,0 +102,4 @@
+                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);

@@ -0,0 +12,4 @@
+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.
+

@@ -0,0 +23,4 @@
+        let listener = TcpListener::bind(address)?;
+        let handler = Arc::new(handler);
+        loop {
+            //  Listen to new (multiple) connection

@@ -0,0 +33,4 @@
+            let handler_clone = handler.clone();
+            //  On connection - spawn a new thread
+            thread::spawn(move || loop {
+                let mut buffer = [0; 1024];

@@ -0,0 +24,4 @@
+        let handler = Arc::new(handler);
+        loop {
+            //  Listen to new (multiple) connection
+            let mut reading_stream = listener.accept()?.0;

@@ -0,0 +18,4 @@
+
+///     For converting byte stream that is expected from the ue-server into actual messages.
+///     Conversion process has two steps:
+///     1.  Every string message is converted into it's utf8 representation and is pre-pended with

@@ -0,0 +20,4 @@
+///     Conversion process has two steps:
+///     1.  Every string message is converted into it's utf8 representation and is pre-pended with
+///         it's own length in format:
+///             [MESSAGE_LENGTH: length of the message in utf8 encoding | 4 bytes: u32 BE]

@@ -0,0 +22,4 @@
+///         it's own length in format:
+///             [MESSAGE_LENGTH: length of the message in utf8 encoding | 4 bytes: u32 BE]
+///             [MESSAGE: utf8-encoded string | `MESSAGE_LENGTH` bytes]
+///         Resulting byte sequences from all the messages then concatenated, in order, into

@@ -0,0 +33,4 @@
+/// how many bytes ue-server has received so far.
+///     NOTE: `try_pop()` can return `None` even if not all message data has been transferred,
+/// in case ue-server's buffer does not have enough space. Use `is_empty()` call to check for
+/// whether `MessageWriter` has depleted all it's data.

@@ -0,0 +84,4 @@
+    }
+
+    pub fn update_ue_received_bytes(&mut self, ue_received_bytes: u64) {
+        self.ue_received_bytes = max(ue_received_bytes, self.ue_received_bytes);

@@ -0,0 +88,4 @@
+    }
+
+    fn available_ue_buffer_capacity(&self) -> usize {
+        match usize::try_from(self.sent_bytes - self.ue_received_bytes).ok() {

@@ -0,0 +25,4 @@
+///         Resulting byte sequences from all the messages then concatenated, in order, into
+///         a single data stream.
+///     2.  Resulting data stream is then separated into "chunks" that can be accepted by
+///         the ue-server (each no longer than `UE_INPUT_BUFFER` in total) and are sent in a format:

@@ -0,0 +17,4 @@
+}
+
+///     For converting byte stream that is expected from the ue-server into actual messages.
+///     Conversion process has two steps:

@@ -9,1 +9,4 @@
 [dependencies]
+custom_error="1.*"
+serde ="1.*"
+serde_json="1.*"

@@ -0,0 +28,4 @@
+            f,
+            "{}",
+            json!({
+                "s": self.service,

@@ -0,0 +11,4 @@
+
+impl AvariceMessage {
+    /// Parses JSON form of a message into `AvariceMessage` struct
+    pub fn from(message_str: &str) -> Option<AvariceMessage> {

@@ -0,0 +3,4 @@
+use serde_json::json;
+use std::fmt;
+
+pub struct AvariceMessage {

@@ -0,0 +15,4 @@
+        let mut message_json: serde_json::Value = serde_json::from_str(message_str).unwrap();
+        let message_json = message_json.as_object_mut()?;
+        Some(AvariceMessage {
+            service: message_json.remove("s")?.as_str()?.to_owned(),

@@ -0,0 +1,38 @@
+//! Implements Avarice message: target service, message type, json parameters
+use serde_json;

@@ -0,0 +32,4 @@
+
+impl AvariceServer {
+    /// Blocks until a new message arrives from one of the ue-servers. Returns a pair of `Link`,
+    /// corresponding to the ue-server that sent next message and `AvariceMessage`,

@@ -0,0 +34,4 @@
+    /// Blocks until a new message arrives from one of the ue-servers. Returns a pair of `Link`,
+    /// corresponding to the ue-server that sent next message and `AvariceMessage`,
+    /// representing that message.
+    pub fn next(&mut self) -> Option<(&mut Link, AvariceMessage)> {

@@ -0,0 +49,4 @@
+                            writer: MessageWriter::new(),
+                        },
+                    );
+                    continue;

@@ -0,0 +60,4 @@
+                    continue;
+                }
+                Ok(NetworkMessage::UEReceivedUpdate(ue_server_address, ue_received_bytes)) => {
+                    if let Some(link) = self.connected_links.get_mut(&ue_server_address) {

@@ -0,0 +66,4 @@
+                    continue;
+                }
+                Ok(NetworkMessage::MessageReceived(ue_server_address, message)) => {
+                    //  Not having a link with key `ue_server_address` should be impossible here

@@ -0,0 +88,4 @@
+        self.ue_server_address
+    }
+
+    fn update_ue_received_bytes(&mut self, ue_received_bytes: u64) {

@@ -0,0 +100,4 @@
+    }
+}
+
+/// Creates a new `AvariceServer` that will listen for ue-server connections on the specified port.

@@ -0,0 +3,4 @@
+//! spawns a new thread for every connected ue-server to handle reading data from it.
+//!     Since all reading is handled in ue-servers' own threads, to collect messages they have
+//! received in the main thread we use `std::sync::mpsc::Sender`. Conversely, all the writing to
+//! ue-server is handled in the main thread itself. Writing `TcpStream` is sent to the main thread

@@ -0,0 +146,4 @@
+        Ok(())
+    }
+
+    pub fn pop(&mut self) -> Option<String> {

@@ -0,0 +154,4 @@
+        self.ue_received_bytes
+    }
+
+    pub fn is_broken(&self) -> bool {

@@ -0,0 +60,4 @@
+        self.pending_data.extend(message_as_utf8.into_iter());
+    }
+
+    pub fn try_pop(&mut self) -> Option<Vec<u8>> {