dkanus/Avarice
2
1
Fork 1
Code Issues 9 Pull Requests 2 Releases Wiki Activity

Compare commits

base: dkanus:5bfdd612482b02b7f3edf89132948e50369c7135
Branches Tags
dkanus:master
dkanus:feature_link
dkanus:feature_database
..
compare: dkanus:b187041d9e053989957f20df3f8589dd7165eb33
Branches Tags
dkanus:feature_link
dkanus:feature_database
dkanus:master

No commits in common. "5bfdd612482b02b7f3edf89132948e50369c7135" and "b187041d9e053989957f20df3f8589dd7165eb33" have entirely different histories.
5bfdd61248 ... b187041d9e

3 changed files with 310 additions and 483 deletions
Show Stats Expand all files Collapse all files

314
src/link/mod.rs
View File

@ -1,4 +1,310 @@
mod reader; use std::collections::VecDeque;
mod writer; use std::str;
pub use reader::MessageReader;
pub use writer::MessageWriter; 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<u8>,
read_messages: VecDeque<String>,
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<String> {
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());
}

310
src/link/reader.rs
View File

@ -1,310 +0,0 @@
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<u8>,
read_messages: VecDeque<String>,
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<String> {
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());
}

169
src/link/writer.rs
View File

@ -1,169 +0,0 @@
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<u8>,
bytes_sent: usize,
}
pub struct MessageWriter {
sent_bytes: u64,
ue_received_bytes: u64,
pending_messages: VecDeque<PendingMessage>,
}
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<Vec<u8>> {
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