diff --git a/src/link/mod.rs b/src/link/mod.rs index 511473d..70b5e9e 100644 --- a/src/link/mod.rs +++ b/src/link/mod.rs @@ -1,310 +1,4 @@ -use std::collections::VecDeque; -use std::str; - -extern crate custom_error; -use custom_error::custom_error; - -// 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 = 2; -// 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; -// Arbitrary value indicating that next byte sequence from ue-server reports amount of bytes -// received by that server so far. -const HEAD_UE_RECEIVED: u8 = 85; -// Arbitrary value indicating that next byte sequence from ue-server contains JSON message. -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; - -custom_error! { pub ReadingStreamError - InvalidHead{input: u8} = "Invalid byte used as a HEAD: {input}", - MessageTooLong{length: usize} = "Message to receive is too long: {length}", - InvalidUnicode = "Invalid utf-8 was received", - BrokenStream = "Used stream is broken" -} - -enum ReadingState { - Head, - ReceivedBytes, - Length, - Payload, -} - -/// 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 | 2 bytes: u16 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 ue-server and `pop()` to -/// 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, - read_messages: VecDeque, - ue_received_bytes: u64, -} - -impl MessageReader { - pub fn new() -> MessageReader { - MessageReader { - is_broken: false, - reading_state: ReadingState::Head, - read_bytes: 0, - length_buffer: [0; 4], - current_message_length: 0, - // Will be recreated with `with_capacity` in `push_byte()` - current_message: Vec::new(), - // This value should be more than enough for typical use - read_messages: VecDeque::with_capacity(100), - ue_received_bytes: 0, - } - } - - pub fn push_byte(&mut self, input: u8) -> Result<(), ReadingStreamError> { - if self.is_broken { - return Err(ReadingStreamError::BrokenStream); - } - match &self.reading_state { - ReadingState::Head => { - 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.length_buffer[self.read_bytes] = input; - self.read_bytes += 1; - if self.read_bytes >= UE_RECEIVED_FIELD_SIZE { - self.ue_received_bytes += array_of_u8_to_u16(self.length_buffer) as u64; - self.change_state(ReadingState::Head); - } - } - ReadingState::Length => { - self.length_buffer[self.read_bytes] = input; - self.read_bytes += 1; - if self.read_bytes >= UE_LENGTH_FIELD_SIZE { - self.current_message_length = array_of_u8_to_u32(self.length_buffer) as usize; - if self.current_message_length > MAX_UE_MESSAGE_LENGTH { - self.is_broken = true; - return Err(ReadingStreamError::MessageTooLong { - length: self.current_message_length, - }); - } - self.current_message = Vec::with_capacity(self.current_message_length); - self.change_state(ReadingState::Payload); - } - } - ReadingState::Payload => { - self.current_message.push(input); - 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()), - _ => { - self.is_broken = true; - return Err(ReadingStreamError::InvalidUnicode); - } - }; - self.current_message.clear(); - self.current_message_length = 0; - self.change_state(ReadingState::Head); - } - } - } - Ok(()) - } - - pub fn push(&mut self, input: &[u8]) -> Result<(), ReadingStreamError> { - for &byte in input { - self.push_byte(byte)?; - } - Ok(()) - } - - pub fn pop(&mut self) -> Option { - self.read_messages.pop_back() - } - - 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_u16(bytes: [u8; 4]) -> u16 { - (u16::from(bytes[0]) << 8) + u16::from(bytes[1]) -} - -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] -fn message_push_byte() { - let mut reader = MessageReader::new(); - 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(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(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); -} - -#[test] -fn received_push_byte() { - let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(0).unwrap(); - reader.push_byte(0xf3).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xf3); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(0xb2).unwrap(); - reader.push_byte(0x04).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3 - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(231).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); -} - -#[test] -fn mixed_push_byte() { - let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(0).unwrap(); - reader.push_byte(0xf3).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(b'Y').unwrap(); - reader.push_byte(b'o').unwrap(); - reader.push_byte(b'!').unwrap(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(0xb2).unwrap(); - reader.push_byte(0x04).unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3 - assert_eq!(reader.pop().unwrap(), "Yo!"); - assert_eq!(reader.pop(), None); -} - -#[test] -fn pushing_many_bytes_at_once() { - let mut reader = MessageReader::new(); - reader - .push(&[ - HEAD_UE_RECEIVED, - 0, - 0xf3, - HEAD_UE_MESSAGE, - 0, - 0, - 0, - 3, - b'Y', - b'o', - b'!', - HEAD_UE_RECEIVED, - 0xb2, - 0x04, - ]) - .unwrap(); - assert_eq!(reader.ue_received_bytes(), 0xb2_f7); - assert_eq!(reader.pop().unwrap(), "Yo!"); - assert_eq!(reader.pop(), None); -} - -#[test] -fn generates_error_invalid_head() { - let mut reader = MessageReader::new(); - reader.push_byte(HEAD_UE_RECEIVED).unwrap(); - reader.push_byte(0).unwrap(); - reader.push_byte(0xf3).unwrap(); - assert!(!reader.is_broken()); - reader - .push_byte(25) - .expect_err("Testing failing on incorrect HEAD"); - assert!(reader.is_broken()); -} - -#[test] -fn generates_error_message_too_long() { - let mut reader = MessageReader::new(); - 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(bytes[3]) - .expect_err("Testing failing on exceeding allowed message length"); - assert!(reader.is_broken()); -} - -#[test] -fn generates_error_invalid_unicode() { - let mut reader = MessageReader::new(); - 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(2).unwrap(); - reader.push_byte(0b11010011).unwrap(); // start of 2-byte sequence - assert!(!reader.is_broken()); - // Bytes inside multi-byte code point have to have `1` for their high bit - reader - .push_byte(0b01010011) - .expect_err("Testing failing on incorrect unicode"); - assert!(reader.is_broken()); -} +mod reader; +mod writer; +pub use reader::MessageReader; +pub use writer::MessageWriter; diff --git a/src/link/reader.rs b/src/link/reader.rs new file mode 100644 index 0000000..511473d --- /dev/null +++ b/src/link/reader.rs @@ -0,0 +1,310 @@ +use std::collections::VecDeque; +use std::str; + +extern crate custom_error; +use custom_error::custom_error; + +// 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 = 2; +// 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; +// Arbitrary value indicating that next byte sequence from ue-server reports amount of bytes +// received by that server so far. +const HEAD_UE_RECEIVED: u8 = 85; +// Arbitrary value indicating that next byte sequence from ue-server contains JSON message. +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; + +custom_error! { pub ReadingStreamError + InvalidHead{input: u8} = "Invalid byte used as a HEAD: {input}", + MessageTooLong{length: usize} = "Message to receive is too long: {length}", + InvalidUnicode = "Invalid utf-8 was received", + BrokenStream = "Used stream is broken" +} + +enum ReadingState { + Head, + ReceivedBytes, + Length, + Payload, +} + +/// 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 | 2 bytes: u16 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 ue-server and `pop()` to +/// 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, + read_messages: VecDeque, + ue_received_bytes: u64, +} + +impl MessageReader { + pub fn new() -> MessageReader { + MessageReader { + is_broken: false, + reading_state: ReadingState::Head, + read_bytes: 0, + length_buffer: [0; 4], + current_message_length: 0, + // Will be recreated with `with_capacity` in `push_byte()` + current_message: Vec::new(), + // This value should be more than enough for typical use + read_messages: VecDeque::with_capacity(100), + ue_received_bytes: 0, + } + } + + pub fn push_byte(&mut self, input: u8) -> Result<(), ReadingStreamError> { + if self.is_broken { + return Err(ReadingStreamError::BrokenStream); + } + match &self.reading_state { + ReadingState::Head => { + 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.length_buffer[self.read_bytes] = input; + self.read_bytes += 1; + if self.read_bytes >= UE_RECEIVED_FIELD_SIZE { + self.ue_received_bytes += array_of_u8_to_u16(self.length_buffer) as u64; + self.change_state(ReadingState::Head); + } + } + ReadingState::Length => { + self.length_buffer[self.read_bytes] = input; + self.read_bytes += 1; + if self.read_bytes >= UE_LENGTH_FIELD_SIZE { + self.current_message_length = array_of_u8_to_u32(self.length_buffer) as usize; + if self.current_message_length > MAX_UE_MESSAGE_LENGTH { + self.is_broken = true; + return Err(ReadingStreamError::MessageTooLong { + length: self.current_message_length, + }); + } + self.current_message = Vec::with_capacity(self.current_message_length); + self.change_state(ReadingState::Payload); + } + } + ReadingState::Payload => { + self.current_message.push(input); + 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()), + _ => { + self.is_broken = true; + return Err(ReadingStreamError::InvalidUnicode); + } + }; + self.current_message.clear(); + self.current_message_length = 0; + self.change_state(ReadingState::Head); + } + } + } + Ok(()) + } + + pub fn push(&mut self, input: &[u8]) -> Result<(), ReadingStreamError> { + for &byte in input { + self.push_byte(byte)?; + } + Ok(()) + } + + pub fn pop(&mut self) -> Option { + self.read_messages.pop_back() + } + + 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_u16(bytes: [u8; 4]) -> u16 { + (u16::from(bytes[0]) << 8) + u16::from(bytes[1]) +} + +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] +fn message_push_byte() { + let mut reader = MessageReader::new(); + 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(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(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); +} + +#[test] +fn received_push_byte() { + let mut reader = MessageReader::new(); + reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(0).unwrap(); + reader.push_byte(0xf3).unwrap(); + assert_eq!(reader.ue_received_bytes(), 0xf3); + reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(0xb2).unwrap(); + reader.push_byte(0x04).unwrap(); + assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3 + reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(231).unwrap(); + assert_eq!(reader.ue_received_bytes(), 0xb2_f7); +} + +#[test] +fn mixed_push_byte() { + let mut reader = MessageReader::new(); + reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(0).unwrap(); + reader.push_byte(0xf3).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(b'Y').unwrap(); + reader.push_byte(b'o').unwrap(); + reader.push_byte(b'!').unwrap(); + reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(0xb2).unwrap(); + reader.push_byte(0x04).unwrap(); + assert_eq!(reader.ue_received_bytes(), 0xb2_f7); // 0xf7 = 0x04 + 0xf3 + assert_eq!(reader.pop().unwrap(), "Yo!"); + assert_eq!(reader.pop(), None); +} + +#[test] +fn pushing_many_bytes_at_once() { + let mut reader = MessageReader::new(); + reader + .push(&[ + HEAD_UE_RECEIVED, + 0, + 0xf3, + HEAD_UE_MESSAGE, + 0, + 0, + 0, + 3, + b'Y', + b'o', + b'!', + HEAD_UE_RECEIVED, + 0xb2, + 0x04, + ]) + .unwrap(); + assert_eq!(reader.ue_received_bytes(), 0xb2_f7); + assert_eq!(reader.pop().unwrap(), "Yo!"); + assert_eq!(reader.pop(), None); +} + +#[test] +fn generates_error_invalid_head() { + let mut reader = MessageReader::new(); + reader.push_byte(HEAD_UE_RECEIVED).unwrap(); + reader.push_byte(0).unwrap(); + reader.push_byte(0xf3).unwrap(); + assert!(!reader.is_broken()); + reader + .push_byte(25) + .expect_err("Testing failing on incorrect HEAD"); + assert!(reader.is_broken()); +} + +#[test] +fn generates_error_message_too_long() { + let mut reader = MessageReader::new(); + 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(bytes[3]) + .expect_err("Testing failing on exceeding allowed message length"); + assert!(reader.is_broken()); +} + +#[test] +fn generates_error_invalid_unicode() { + let mut reader = MessageReader::new(); + 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(2).unwrap(); + reader.push_byte(0b11010011).unwrap(); // start of 2-byte sequence + assert!(!reader.is_broken()); + // Bytes inside multi-byte code point have to have `1` for their high bit + reader + .push_byte(0b01010011) + .expect_err("Testing failing on incorrect unicode"); + assert!(reader.is_broken()); +} 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