diff --git a/src/link/mod.rs b/src/link/mod.rs index 610c1a0..70b5e9e 100644 --- a/src/link/mod.rs +++ b/src/link/mod.rs @@ -1,60 +1,4 @@ -use std::error::Error; -use std::io::{Read, Write}; -use std::net::{SocketAddr, TcpListener}; -use std::sync::Arc; -use std::{str, thread}; - mod reader; mod writer; pub use reader::MessageReader; pub use writer::MessageWriter; - -pub struct Link { - reader: MessageReader, - writer: MessageWriter, -} - -impl Link { - pub fn run(port: u16, handler: F) -> Result<(), Box> - where - F: Fn(&mut Link, &str) -> () + Send + Sync + 'static, - { - let address = SocketAddr::from(([0, 0, 0, 0], port)); - let listener = TcpListener::bind(address)?; - let handler = Arc::new(handler); - loop { - // Listen to new (multiple) connection - let mut reading_stream = listener.accept()?.0; - let mut writing_stream = reading_stream.try_clone()?; - let mut avarice_link = Link { - reader: MessageReader::new(), - writer: MessageWriter::new(), - }; - let handler_clone = handler.clone(); - // On connection - spawn a new thread - thread::spawn(move || loop { - let mut buffer = [0; 1024]; - // Reading cycle - match reading_stream.read(&mut buffer) { - Ok(n) => avarice_link.reader.push(&buffer[..n]).unwrap(), - _ => panic!("Connection issue!"), - }; - // Handling cycle - while let Some(message) = avarice_link.reader.pop() { - handler_clone(&mut avarice_link, &message); - } - // Writing - avarice_link - .writer - .update_ue_received_bytes(avarice_link.reader.ue_received_bytes()); - if let Some(bytes) = avarice_link.writer.try_pop() { - writing_stream.write_all(&bytes).unwrap(); - } - }); - } - } - - pub fn write(&mut self, message: &str) { - self.writer.push(message); - } -} diff --git a/src/link/reader.rs b/src/link/reader.rs index 03bb4ce..511473d 100644 --- a/src/link/reader.rs +++ b/src/link/reader.rs @@ -89,7 +89,7 @@ impl MessageReader { self.length_buffer[self.read_bytes] = input; self.read_bytes += 1; if self.read_bytes >= UE_RECEIVED_FIELD_SIZE { - self.ue_received_bytes += u64::from(array_of_u8_to_u16(self.length_buffer)); + self.ue_received_bytes += array_of_u8_to_u16(self.length_buffer) as u64; self.change_state(ReadingState::Head); } } diff --git a/src/link/writer.rs b/src/link/writer.rs index 70ae84b..ff7f9c6 100644 --- a/src/link/writer.rs +++ b/src/link/writer.rs @@ -1,86 +1,79 @@ use std::cmp::{max, min}; use std::collections::VecDeque; use std::convert::TryFrom; -use std::iter::Extend; -// Defines how many bytes is used to encode "LENGTH" field in the chunk sent to ue-server -const CHUNK_LENGTH_FIELD: usize = 2; -// Maximum amount of bytes ue-server is able to receive at once +const LENGTH_FIELD_SIZE: usize = 4; const UE_INPUT_BUFFER: usize = 4095; -// Minimal payload size (in bytes) to send, unless there is not enough data left -const MIN_PAYLOAD_SIZE: usize = 50; + +struct PendingMessage { + contents: Vec, + bytes_sent: usize, +} pub struct MessageWriter { sent_bytes: u64, ue_received_bytes: u64, - pending_data: VecDeque, + pending_messages: VecDeque, +} + +impl PendingMessage { + fn bytes_left(&self) -> usize { + max(0, self.contents.len() - self.bytes_sent) + } } -/// 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 -/// 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 -/// 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: -/// [LENGTH: length of the chunk | 2 bytes: u16 BE] -/// [PAYLOAD: utf8-encoded string | `LENGTH` bytes] -/// Use `push()` to input string messages and `try_pop()` to retrieve next chunk, if ue-server -/// can accept it. Call `update_ue_received_bytes()` to update `MessageWriter`'s information about -/// 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. impl MessageWriter { pub fn new() -> MessageWriter { MessageWriter { sent_bytes: 0, ue_received_bytes: 0, - // This value should be more than enough for typical use: - // it will take at least one second to send this much data to a 30 tick rate ue-server. - pending_data: VecDeque::with_capacity(30 * UE_INPUT_BUFFER), + pending_messages: VecDeque::with_capacity(100), } } pub fn push(&mut self, message: &str) { let message_as_utf8 = message.as_bytes(); - let message_as_utf8 = [ - &(message_as_utf8.len() as u32).to_be_bytes(), - message_as_utf8, - ] - .concat(); - self.pending_data.extend(message_as_utf8.into_iter()); + self.pending_messages.push_back(PendingMessage { + contents: [ + &(message_as_utf8.len() as u32).to_be_bytes(), + message_as_utf8, + ] + .concat(), + bytes_sent: 0, + }); } - /// This method will always return chunk if all remaining data will fit inside it, otherwise it - /// will wait until ue-server's buffer has enough space for at least `MIN_PAYLOAD_SIZE` bytes. pub fn try_pop(&mut self) -> Option> { - if self.is_empty() { + if !self.should_send_now() { return None; } - let required_payload_size = min(self.pending_data.len(), MIN_PAYLOAD_SIZE); - let available_payload_space = self - .available_ue_buffer_capacity() - .checked_sub(CHUNK_LENGTH_FIELD) - .unwrap_or_default(); - if required_payload_size > available_payload_space { - return None; - } - let payload_size = min(available_payload_space, self.pending_data.len()); - let mut bytes_to_send = Vec::with_capacity(CHUNK_LENGTH_FIELD + payload_size); - bytes_to_send.extend((payload_size as u16).to_be_bytes().iter()); - for next_byte in self.pending_data.drain(..payload_size) { - bytes_to_send.push(next_byte); - } + let max_output_bytes = self.available_ue_buffer_capacity(); + let next_payload_size = max_output_bytes - 2; + let next_message = match self.pending_messages.get_mut(0) { + Some(message) => message, + _ => return None, + }; + let first_index = next_message.bytes_sent; + let last_index = min( + next_message.bytes_sent + next_payload_size - 1, + next_message.contents.len() - 1, + ); + next_message.bytes_sent = last_index + 1; + let chunk_length = last_index - first_index + 1; + let bytes_to_send = [ + &(chunk_length as u16).to_be_bytes(), + &next_message.contents[first_index..last_index + 1], + ] + .concat(); self.sent_bytes += bytes_to_send.len() as u64; + if next_message.bytes_sent >= next_message.contents.len() { + self.pending_messages.pop_front(); + } Some(bytes_to_send) } pub fn is_empty(&self) -> bool { - self.pending_data.is_empty() + self.pending_messages.is_empty() } pub fn update_ue_received_bytes(&mut self, ue_received_bytes: u64) { @@ -93,27 +86,39 @@ impl MessageWriter { _ => 0, } } + + fn should_send_now(&self) -> bool { + if self.pending_messages.is_empty() { + return false; + } + let bytes_left_in_message = match self.pending_messages.get(0) { + Some(message) => message.bytes_left(), + _ => return false, + }; + let max_output_bytes = self.available_ue_buffer_capacity(); + if bytes_left_in_message <= max_output_bytes { + return true; + }; + let max_bytes_in_payload = UE_INPUT_BUFFER - LENGTH_FIELD_SIZE; + let next_payload_size = max_output_bytes.checked_sub(4).unwrap_or_default(); + let chunks_in_message = (bytes_left_in_message / max_bytes_in_payload) + 1; + let chunks_in_message_after_sending = + ((bytes_left_in_message - next_payload_size) / max_bytes_in_payload) + 1; + chunks_in_message_after_sending < chunks_in_message + } } #[test] -fn writer_contents_after_creation() { - let writer = MessageWriter::new(); +fn new_writer_is_empty() { + let mut writer = MessageWriter::new(); assert_eq!(writer.is_empty(), true); - assert_eq!(writer.available_ue_buffer_capacity(), UE_INPUT_BUFFER); } #[test] -fn writer_content_after_push() { +fn single_short_message() { let mut writer = MessageWriter::new(); writer.push("Hello, world!"); assert_eq!(writer.is_empty(), false); - assert_eq!(writer.available_ue_buffer_capacity(), UE_INPUT_BUFFER); -} - -#[test] -fn writing_single_short_message() { - let mut writer = MessageWriter::new(); - writer.push("Hello, world!"); let resulting_bytes = writer.try_pop().unwrap(); let expected_bytes = [ 0, 17, // Bytes in the chunk = message length (4 bytes) + message (13 bytes) @@ -121,95 +126,44 @@ fn writing_single_short_message() { b'H', b'e', b'l', b'l', b'o', b',', b' ', b'w', b'o', b'r', b'l', b'd', b'!', ]; assert_eq!(writer.is_empty(), true); - assert_eq!( - writer.available_ue_buffer_capacity(), - UE_INPUT_BUFFER - "Hello, world!".len() - 2 - 4 - ); assert_eq!(resulting_bytes, expected_bytes); assert_eq!(writer.sent_bytes, expected_bytes.len() as u64); } #[test] -fn writing_first_chunk_of_single_long_message() { +fn single_long_message() { let mut writer = MessageWriter::new(); // Because we also have to pass lengths, this will go over the sending limit let long_message = "Q".repeat(UE_INPUT_BUFFER); writer.push(&long_message); + assert_eq!(writer.is_empty(), false); let resulting_bytes = writer.try_pop().unwrap(); assert_eq!(writer.is_empty(), false); - assert_eq!(resulting_bytes.len(), UE_INPUT_BUFFER); - assert_eq!(writer.available_ue_buffer_capacity(), 0); + assert_eq!(resulting_bytes.len(), 4095); // Bytes in the chunk = 4095 - 2 = 4093 = 0x0ffd assert_eq!(resulting_bytes[0], 0x0f); - assert_eq!(resulting_bytes[1], 0xfd); + assert_eq!(resulting_bytes[1], 0x0fd); // Bytes in message = 4095 = 0x0fff assert_eq!(resulting_bytes[2], 0); assert_eq!(resulting_bytes[3], 0); assert_eq!(resulting_bytes[4], 0x0f); - assert_eq!(resulting_bytes[5], 0xff); + assert_eq!(resulting_bytes[5], 0x0ff); for &byte in resulting_bytes[6..].iter() { assert_eq!(byte, b'Q'); } + assert_eq!(writer.try_pop(), None); assert_eq!(writer.is_empty(), false); -} - -#[test] -fn writing_second_chunk_of_single_long_message() { - let mut writer = MessageWriter::new(); - // Because we also have to pass lengths, this will go over the sending limit - let long_message = "Q".repeat(UE_INPUT_BUFFER); - writer.push(&long_message); - // This pops all but 6 bytes of `long_message`, that were required to encode lengths of - // message and first chunk - let first_bytes = writer.try_pop().unwrap(); - writer.update_ue_received_bytes(first_bytes.len() as u64); + writer.update_ue_received_bytes(UE_INPUT_BUFFER as u64); let resulting_bytes = writer.try_pop().unwrap(); - assert_eq!( - writer.available_ue_buffer_capacity(), - UE_INPUT_BUFFER - resulting_bytes.len() - ); assert_eq!(writer.is_empty(), true); - // Bytes in the chunk = 6 + // Last time we have popped all but 6 bytes from `long_message` + // Bytes in the chunk = 6 = 0x06 assert_eq!(resulting_bytes[0], 0); - assert_eq!(resulting_bytes[1], 6); + assert_eq!(resulting_bytes[1], 0x06); assert_eq!(resulting_bytes[2..], [b'Q', b'Q', b'Q', b'Q', b'Q', b'Q']) } -#[test] -fn will_write_small_chunks_if_no_more_data() { - let mut writer = MessageWriter::new(); - // Because we also have to pass lengths (of chunk `CHUNK_LENGTH_FIELD` amd of message `4`), - // sending this will leave us with exactly 10 free bytes in the buffer - let long_message = "Q".repeat(UE_INPUT_BUFFER / 2); - writer.push(&long_message); - writer.try_pop(); - let short_message = "Hello, world!"; - writer.push(&short_message); - let expected_bytes = [ - 0, 17, // Bytes in the chunk = message length (4 bytes) + message (13 bytes) - 0, 0, 0, 13, // Bytes in the message - b'H', b'e', b'l', b'l', b'o', b',', b' ', b'w', b'o', b'r', b'l', b'd', b'!', - ]; - // There should be enough space in the ue-server buffer to send `short_message` - let resulting_bytes = writer.try_pop().unwrap(); - assert_eq!(resulting_bytes, expected_bytes); - assert_eq!(writer.try_pop(), None); - assert_eq!(writer.is_empty(), true); -} - -#[test] -fn will_not_write_small_chunks_if_more_data_remains() { - let mut writer = MessageWriter::new(); - // Because we also have to pass lengths (of chunk `CHUNK_LENGTH_FIELD` amd of message `4`), - // sending this will leave us with exactly 10 free bytes in the buffer - let long_message = "Q".repeat(UE_INPUT_BUFFER - CHUNK_LENGTH_FIELD - 4 - 10); - writer.push(&long_message); - writer.try_pop(); - let short_message = "Hello, world!"; - writer.push(&short_message); - // `MessageWriter` can neither send full message, nor a chunk of size 10 - // (because it is too short) - assert_eq!(writer.try_pop(), None); - assert_eq!(writer.is_empty(), false); -} +// TODO: test that `update_ue_received_bytes()` cannot reduce that value +// TODO: test several messages, message only partially fitting into remaining buffer +// TODO: test giving just barely enough free buffer space diff --git a/src/main.rs b/src/main.rs index 9378c87..7f15ba3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,9 +1,13 @@ +use std::env; +use std::path::Path; mod link; -use link::Link; +mod unreal_config; fn main() { - match Link::run(1234, |link, message| { link.write(message);}) { - Ok(_) => print!("Connect!"), - _ => print!("Connection error!"), - }; + 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); + } }