diff --git a/src/link/mod.rs b/src/link/mod.rs index 26ec5ad..70b5e9e 100644 --- a/src/link/mod.rs +++ b/src/link/mod.rs @@ -1,2 +1,4 @@ mod reader; +mod writer; pub use reader::MessageReader; +pub use writer::MessageWriter; diff --git a/src/link/writer.rs b/src/link/writer.rs new file mode 100644 index 0000000..ff7f9c6 --- /dev/null +++ b/src/link/writer.rs @@ -0,0 +1,169 @@ +use std::cmp::{max, min}; +use std::collections::VecDeque; +use std::convert::TryFrom; + +const LENGTH_FIELD_SIZE: usize = 4; +const UE_INPUT_BUFFER: usize = 4095; + +struct PendingMessage { + contents: Vec, + bytes_sent: usize, +} + +pub struct MessageWriter { + sent_bytes: u64, + ue_received_bytes: u64, + pending_messages: VecDeque, +} + +impl PendingMessage { + fn bytes_left(&self) -> usize { + max(0, self.contents.len() - self.bytes_sent) + } +} + +impl MessageWriter { + pub fn new() -> MessageWriter { + MessageWriter { + sent_bytes: 0, + ue_received_bytes: 0, + pending_messages: VecDeque::with_capacity(100), + } + } + + pub fn push(&mut self, message: &str) { + let message_as_utf8 = message.as_bytes(); + self.pending_messages.push_back(PendingMessage { + contents: [ + &(message_as_utf8.len() as u32).to_be_bytes(), + message_as_utf8, + ] + .concat(), + bytes_sent: 0, + }); + } + + pub fn try_pop(&mut self) -> Option> { + if !self.should_send_now() { + return None; + } + 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_messages.is_empty() + } + + pub fn update_ue_received_bytes(&mut self, ue_received_bytes: u64) { + self.ue_received_bytes = max(ue_received_bytes, self.ue_received_bytes); + } + + fn available_ue_buffer_capacity(&self) -> usize { + match usize::try_from(self.sent_bytes - self.ue_received_bytes).ok() { + Some(ue_buffered_bytes) => max(0, UE_INPUT_BUFFER - ue_buffered_bytes), + _ => 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 new_writer_is_empty() { + let mut writer = MessageWriter::new(); + assert_eq!(writer.is_empty(), true); +} + +#[test] +fn single_short_message() { + let mut writer = MessageWriter::new(); + writer.push("Hello, world!"); + assert_eq!(writer.is_empty(), false); + let resulting_bytes = writer.try_pop().unwrap(); + 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'!', + ]; + assert_eq!(writer.is_empty(), true); + assert_eq!(resulting_bytes, expected_bytes); + assert_eq!(writer.sent_bytes, expected_bytes.len() as u64); +} + +#[test] +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(), 4095); + // Bytes in the chunk = 4095 - 2 = 4093 = 0x0ffd + assert_eq!(resulting_bytes[0], 0x0f); + 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], 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); + writer.update_ue_received_bytes(UE_INPUT_BUFFER as u64); + let resulting_bytes = writer.try_pop().unwrap(); + assert_eq!(writer.is_empty(), true); + // 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], 0x06); + assert_eq!(resulting_bytes[2..], [b'Q', b'Q', b'Q', b'Q', b'Q', b'Q']) +} + +// 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