use serde::{Deserialize, Serialize};
use std::fmt;
use std::time::{Duration, Instant};
pub const DEFAULT_MAX_BUFFER: usize = 65_536;
pub const REGEX_LOOKBACK: usize = 32;
pub trait Framer: Send {
fn push(&mut self, data: &[u8], now: Instant, out: &mut Vec<Vec<u8>>);
fn poll(&mut self, _now: Instant, _out: &mut Vec<Vec<u8>>) {}
fn deadline(&self) -> Option<Instant> {
None
}
fn flush(&mut self) -> Option<Vec<u8>>;
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfigError(pub String);
impl fmt::Display for ConfigError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "invalid parser options: {}", self.0)
}
}
impl std::error::Error for ConfigError {}
fn bad<T>(msg: &str) -> Result<T, ConfigError> {
Err(ConfigError(msg.into()))
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum ByteSeq {
Text(String),
Bytes(Vec<u8>),
}
impl ByteSeq {
pub fn to_vec(&self) -> Vec<u8> {
match self {
ByteSeq::Text(s) => s.as_bytes().to_vec(),
ByteSeq::Bytes(b) => b.clone(),
}
}
}
impl From<&str> for ByteSeq {
fn from(s: &str) -> Self {
ByteSeq::Text(s.into())
}
}
impl From<&[u8]> for ByteSeq {
fn from(b: &[u8]) -> Self {
ByteSeq::Bytes(b.to_vec())
}
}
fn non_empty(seq: &ByteSeq, name: &str) -> Result<Vec<u8>, ConfigError> {
let v = seq.to_vec();
if v.is_empty() {
return bad(&format!("\"{name}\" has a 0 length"));
}
Ok(v)
}
fn max_buf(v: Option<usize>) -> Result<usize, ConfigError> {
match v {
Some(0) => bad("\"maxBufferSize\" must be greater than 0"),
Some(n) => Ok(n),
None => Ok(DEFAULT_MAX_BUFFER),
}
}
fn find(hay: &[u8], needle: &[u8], from: usize) -> Option<usize> {
if from > hay.len() || needle.len() > hay.len() - from {
return None;
}
hay[from..]
.windows(needle.len())
.position(|w| w == needle)
.map(|p| p + from)
}
fn emit(out: &mut Vec<Vec<u8>>, frame: &[u8]) {
if !frame.is_empty() {
out.push(frame.to_vec());
}
}
fn take_flush(buf: &mut Vec<u8>) -> Option<Vec<u8>> {
let rest = std::mem::take(buf);
(!rest.is_empty()).then_some(rest)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(
tag = "type",
rename_all = "camelCase",
rename_all_fields = "camelCase",
deny_unknown_fields
)]
pub enum FramerConfig {
Delimiter {
delimiter: ByteSeq,
#[serde(default)]
include_delimiter: bool,
#[serde(default)]
max_buffer_size: Option<usize>,
},
Readline {
#[serde(default)]
delimiter: Option<ByteSeq>,
#[serde(default)]
include_delimiter: bool,
#[serde(default)]
max_buffer_size: Option<usize>,
#[serde(default)]
encoding: Option<String>,
},
ByteLength { length: usize },
InterByteTimeout {
#[serde(alias = "interval")]
interval_ms: u64,
#[serde(default)]
max_buffer_size: Option<usize>,
},
PacketLength {
#[serde(default)]
delimiter: Option<OneOrMany>,
#[serde(default)]
delimiter_bytes: Option<usize>,
#[serde(default)]
packet_overhead: Option<usize>,
#[serde(default)]
length_bytes: Option<usize>,
#[serde(default)]
length_offset: Option<usize>,
#[serde(default)]
max_len: Option<usize>,
#[serde(default)]
max_buffer_size: Option<usize>,
},
Ready { delimiter: ByteSeq },
Regex {
regex: String,
#[serde(default)]
max_buffer_size: Option<usize>,
#[serde(default)]
encoding: Option<String>,
},
StartEnd {
start_delimiter: ByteSeq,
end_delimiter: ByteSeq,
#[serde(default)]
include_start_delimiter: bool,
#[serde(default)]
include_end_delimiter: bool,
#[serde(default)]
max_buffer_size: Option<usize>,
},
Slip(SlipOptions),
CcTalk {
#[serde(default, alias = "maxDelayBetweenBytesMs")]
max_delay_ms: Option<u64>,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum OneOrMany {
One(u32),
Many(Vec<u32>),
}
impl FramerConfig {
pub fn build(&self) -> Result<Box<dyn Framer>, ConfigError> {
Ok(match self {
FramerConfig::Delimiter {
delimiter,
include_delimiter,
max_buffer_size,
} => Box::new(DelimiterFramer::new(
non_empty(delimiter, "delimiter")?,
*include_delimiter,
max_buf(*max_buffer_size)?,
)),
FramerConfig::Readline {
delimiter,
include_delimiter,
max_buffer_size,
encoding: _,
} => {
let delimiter = match delimiter {
Some(d) => non_empty(d, "delimiter")?,
None => b"\n".to_vec(),
};
Box::new(DelimiterFramer::new(
delimiter,
*include_delimiter,
max_buf(*max_buffer_size)?,
))
}
FramerConfig::ByteLength { length } => {
if *length == 0 {
return bad("\"length\" must be greater than 0");
}
Box::new(ByteLengthFramer {
length: *length,
buf: Vec::new(),
})
}
FramerConfig::InterByteTimeout {
interval_ms,
max_buffer_size,
} => {
if *interval_ms == 0 {
return bad("\"intervalMs\" must be greater than 0");
}
Box::new(InterByteTimeoutFramer {
interval: Duration::from_millis(*interval_ms),
max: max_buf(*max_buffer_size)?,
buf: Vec::new(),
deadline: None,
})
}
FramerConfig::PacketLength {
delimiter,
delimiter_bytes,
packet_overhead,
length_bytes,
length_offset,
max_len,
max_buffer_size,
} => {
let p = PacketLengthFramer {
delimiters: match delimiter {
None => vec![0xaa],
Some(OneOrMany::One(d)) => vec![*d],
Some(OneOrMany::Many(d)) => d.clone(),
},
delimiter_bytes: delimiter_bytes.unwrap_or(1),
packet_overhead: packet_overhead.unwrap_or(2),
length_bytes: length_bytes.unwrap_or(1),
length_offset: length_offset.unwrap_or(1),
max_len: max_len.unwrap_or(0xff),
max: max_buf(*max_buffer_size)?,
buf: Vec::new(),
started: false,
};
if p.delimiters.is_empty() || !(1..=4).contains(&p.delimiter_bytes) {
return bad("\"delimiter\" needs 1..=4 \"delimiterBytes\"");
}
if !(1..=8).contains(&p.length_bytes) {
return bad("\"lengthBytes\" must be 1..=8");
}
if p.length_offset < p.delimiter_bytes {
return bad("\"lengthOffset\" overlaps the delimiter");
}
if p.length_offset
.checked_add(p.length_bytes)
.is_none_or(|end| end > p.max)
{
return bad("\"lengthOffset\" + \"lengthBytes\" exceeds \"maxBufferSize\"");
}
Box::new(p)
}
FramerConfig::Ready { delimiter } => Box::new(ReadyFramer {
delimiter: non_empty(delimiter, "delimiter")?,
tail: Vec::new(),
ready: false,
}),
FramerConfig::Regex {
regex,
max_buffer_size,
encoding: _,
} => Box::new(RegexFramer {
regex: regex_lite::Regex::new(regex)
.map_err(|e| ConfigError(format!("\"regex\": {e}")))?,
max: max_buf(*max_buffer_size)?,
text: String::new(),
pending: Vec::new(),
scanned: 0,
full_scan: 0,
}),
FramerConfig::StartEnd {
start_delimiter,
end_delimiter,
include_start_delimiter,
include_end_delimiter,
max_buffer_size,
} => Box::new(StartEndFramer {
start: non_empty(start_delimiter, "startDelimiter")?,
end: non_empty(end_delimiter, "endDelimiter")?,
include_start: *include_start_delimiter,
include_end: *include_end_delimiter,
max: max_buf(*max_buffer_size)?,
buf: Vec::new(),
open: None,
scan_from: 0,
}),
FramerConfig::Slip(o) => Box::new(SlipDecoder::new(o.clone())),
FramerConfig::CcTalk { max_delay_ms } => Box::new(CcTalkFramer {
max_delay: Duration::from_millis(max_delay_ms.unwrap_or(50)),
buf: Vec::new(),
last: None,
}),
})
}
}
struct DelimiterFramer {
delimiter: Vec<u8>,
include: bool,
max: usize,
buf: Vec<u8>,
}
impl DelimiterFramer {
fn new(delimiter: Vec<u8>, include: bool, max: usize) -> Self {
Self {
delimiter,
include,
max,
buf: Vec::new(),
}
}
}
impl Framer for DelimiterFramer {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
let mut from = self.buf.len().saturating_sub(self.delimiter.len() - 1);
self.buf.extend_from_slice(data);
let mut start = 0;
while let Some(pos) = find(&self.buf, &self.delimiter, from.max(start)) {
let end = pos
+ if self.include {
self.delimiter.len()
} else {
0
};
emit(out, &self.buf[start..end]);
start = pos + self.delimiter.len();
from = start;
}
self.buf.drain(..start);
while self.buf.len() >= self.max {
let rest = self.buf.split_off(self.max);
out.push(std::mem::replace(&mut self.buf, rest));
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
take_flush(&mut self.buf)
}
}
struct ByteLengthFramer {
length: usize,
buf: Vec<u8>,
}
impl Framer for ByteLengthFramer {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
self.buf.extend_from_slice(data);
let full = self.buf.len() / self.length * self.length;
for frame in self.buf[..full].chunks(self.length) {
out.push(frame.to_vec());
}
self.buf.drain(..full);
}
fn flush(&mut self) -> Option<Vec<u8>> {
take_flush(&mut self.buf)
}
}
struct InterByteTimeoutFramer {
interval: Duration,
max: usize,
buf: Vec<u8>,
deadline: Option<Instant>,
}
impl Framer for InterByteTimeoutFramer {
fn push(&mut self, data: &[u8], now: Instant, out: &mut Vec<Vec<u8>>) {
self.poll(now, out);
for &b in data {
self.buf.push(b);
if self.buf.len() >= self.max {
out.push(std::mem::take(&mut self.buf));
}
}
if !data.is_empty() {
self.deadline = Some(now + self.interval);
}
}
fn poll(&mut self, now: Instant, out: &mut Vec<Vec<u8>>) {
if self.deadline.is_some_and(|d| now >= d) {
self.deadline = None;
emit(out, &std::mem::take(&mut self.buf));
}
}
fn deadline(&self) -> Option<Instant> {
self.deadline.filter(|_| !self.buf.is_empty())
}
fn flush(&mut self) -> Option<Vec<u8>> {
self.deadline = None;
take_flush(&mut self.buf)
}
}
struct PacketLengthFramer {
delimiters: Vec<u32>,
delimiter_bytes: usize,
packet_overhead: usize,
length_bytes: usize,
length_offset: usize,
max_len: usize,
max: usize,
buf: Vec<u8>,
started: bool,
}
impl Framer for PacketLengthFramer {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
for &b in data {
self.buf.push(b);
if !self.started {
if self.buf.len() > self.delimiter_bytes {
self.buf.remove(0);
}
if self.buf.len() == self.delimiter_bytes {
let v = self.buf.iter().fold(0u32, |a, &x| (a << 8) | x as u32);
self.started = self.delimiters.contains(&v);
}
continue;
}
let end = self.length_offset + self.length_bytes;
if self.buf.len() >= end {
let len = self.buf[self.length_offset..end]
.iter()
.rev()
.fold(0u64, |a, &x| (a << 8) | x as u64);
let total = len.saturating_add(self.packet_overhead as u64);
if self.buf.len() as u64 >= total || len > self.max_len as u64 {
out.push(std::mem::take(&mut self.buf));
self.started = false;
continue;
}
}
if self.buf.len() >= self.max {
out.push(std::mem::take(&mut self.buf));
self.started = false;
}
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
let started = std::mem::take(&mut self.started);
let rest = std::mem::take(&mut self.buf);
(started && !rest.is_empty()).then_some(rest)
}
}
struct ReadyFramer {
delimiter: Vec<u8>,
tail: Vec<u8>,
ready: bool,
}
impl Framer for ReadyFramer {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
if self.ready {
emit(out, data);
return;
}
let keep = self.tail.len();
self.tail.extend_from_slice(data);
if let Some(pos) = find(&self.tail, &self.delimiter, 0) {
self.ready = true;
let rest = pos + self.delimiter.len();
emit(out, &self.tail[rest.max(keep)..]);
self.tail = Vec::new();
} else {
let cut = self.tail.len().saturating_sub(self.delimiter.len() - 1);
self.tail.drain(..cut);
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
None
}
}
struct RegexFramer {
regex: regex_lite::Regex,
max: usize,
text: String,
pending: Vec<u8>,
scanned: usize,
full_scan: usize,
}
impl RegexFramer {
fn decode(&mut self, data: &[u8]) {
let mut bytes = std::mem::take(&mut self.pending);
bytes.extend_from_slice(data);
let mut rest: &[u8] = &bytes;
loop {
match std::str::from_utf8(rest) {
Ok(s) => {
self.text.push_str(s);
return;
}
Err(e) => {
let (ok, tail) = rest.split_at(e.valid_up_to());
self.text
.push_str(std::str::from_utf8(ok).unwrap_or_default());
match e.error_len() {
Some(n) => {
self.text.push('\u{FFFD}');
rest = &tail[n..];
}
None => {
self.pending = tail.to_vec();
return;
}
}
}
}
}
}
}
impl Framer for RegexFramer {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
self.decode(data);
let mut from = self.scanned.saturating_sub(REGEX_LOOKBACK);
if self.text.len() >= 2 * self.full_scan {
from = 0;
self.full_scan = self.text.len();
}
while !self.text.is_char_boundary(from) {
from -= 1;
}
if self.regex.find_at(&self.text, from).is_some() {
let mut parts: Vec<&str> = self.regex.split(&self.text).collect();
let last = parts.pop().unwrap_or_default().to_string();
for p in parts {
emit(out, p.as_bytes());
}
self.text = last;
self.full_scan = self.text.len();
}
self.scanned = self.text.len();
if self.text.len() >= self.max {
out.push(std::mem::take(&mut self.text).into_bytes());
self.scanned = 0;
self.full_scan = 0;
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
self.scanned = 0;
self.full_scan = 0;
let mut rest = std::mem::take(&mut self.text).into_bytes();
rest.append(&mut self.pending);
(!rest.is_empty()).then_some(rest)
}
}
struct StartEndFramer {
start: Vec<u8>,
end: Vec<u8>,
include_start: bool,
include_end: bool,
max: usize,
buf: Vec<u8>,
open: Option<usize>,
scan_from: usize,
}
impl Framer for StartEndFramer {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
self.buf.extend_from_slice(data);
let mut consumed = 0;
loop {
let s = match self.open {
Some(s) => s,
None => {
let Some(s) = find(&self.buf, &self.start, self.scan_from) else {
let tail = self.buf.len().saturating_sub(self.start.len() - 1);
self.scan_from = tail.max(consumed);
break;
};
self.open = Some(s);
self.scan_from = s + self.start.len();
s
}
};
let Some(e) = find(&self.buf, &self.end, self.scan_from) else {
let tail = self.buf.len().saturating_sub(self.end.len() - 1);
self.scan_from = tail.max(s + self.start.len());
break;
};
let from = if self.include_start {
s
} else {
s + self.start.len()
};
let to = if self.include_end {
e + self.end.len()
} else {
e
};
emit(out, &self.buf[from..to]);
consumed = e + self.end.len();
self.open = None;
self.scan_from = consumed;
}
self.buf.drain(..consumed);
self.open = self.open.map(|s| s - consumed);
self.scan_from -= consumed;
if self.buf.len() >= self.max {
self.open = None;
self.scan_from = 0;
match self
.buf
.windows(self.start.len())
.rposition(|w| w == self.start)
{
Some(p) if p > 0 => {
self.buf.drain(..p);
}
_ => self.buf.clear(),
}
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
self.open = None;
self.scan_from = 0;
take_flush(&mut self.buf)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", default, deny_unknown_fields)]
pub struct SlipOptions {
#[serde(alias = "START")]
pub start: Option<u8>,
#[serde(alias = "ESC")]
pub esc: u8,
#[serde(alias = "END")]
pub end: u8,
#[serde(alias = "ESC_START")]
pub esc_start: Option<u8>,
#[serde(alias = "ESC_END")]
pub esc_end: u8,
#[serde(alias = "ESC_ESC")]
pub esc_esc: u8,
}
impl Default for SlipOptions {
fn default() -> Self {
Self {
start: None,
esc: 0xdb,
end: 0xc0,
esc_start: None,
esc_end: 0xdc,
esc_esc: 0xdd,
}
}
}
pub fn slip_encode(data: &[u8], o: &SlipOptions, bluetooth_quirk: bool) -> Vec<u8> {
if bluetooth_quirk && data.is_empty() {
return Vec::new();
}
let mut v = Vec::with_capacity(data.len() * 2 + 3);
if bluetooth_quirk {
v.push(o.end);
}
if let Some(s) = o.start {
v.push(s);
}
for &b in data {
match b {
b if Some(b) == o.start && o.esc_start.is_some() => {
v.extend([o.esc, o.esc_start.unwrap_or_default()])
}
b if b == o.end => v.extend([o.esc, o.esc_end]),
b if b == o.esc => v.extend([o.esc, o.esc_esc]),
b => v.push(b),
}
}
v.push(o.end);
v
}
pub struct SlipDecoder {
o: SlipOptions,
buf: Vec<u8>,
escape: bool,
started: bool,
}
impl SlipDecoder {
pub fn new(o: SlipOptions) -> Self {
Self {
o,
buf: Vec::new(),
escape: false,
started: false,
}
}
}
impl Framer for SlipDecoder {
fn push(&mut self, data: &[u8], _now: Instant, out: &mut Vec<Vec<u8>>) {
let o = &self.o;
for &raw in data {
let mut byte = raw;
if Some(byte) == o.start {
self.started = true;
continue;
}
if o.start.is_none() {
self.started = true;
}
if self.escape {
if o.start.is_some() && Some(byte) == o.esc_start {
byte = o.start.unwrap_or_default();
} else if byte == o.esc_esc {
byte = o.esc;
} else if byte == o.esc_end {
byte = o.end;
} else {
emit(out, &std::mem::take(&mut self.buf));
}
} else if byte == o.esc {
self.escape = true;
continue;
} else if byte == o.end {
emit(out, &std::mem::take(&mut self.buf));
self.escape = false;
self.started = false;
continue;
}
self.escape = false;
if self.started {
self.buf.push(byte);
if self.buf.len() >= DEFAULT_MAX_BUFFER {
out.push(std::mem::take(&mut self.buf));
}
}
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
self.escape = false;
self.started = false;
take_flush(&mut self.buf)
}
}
struct CcTalkFramer {
max_delay: Duration,
buf: Vec<u8>,
last: Option<Instant>,
}
impl Framer for CcTalkFramer {
fn push(&mut self, data: &[u8], now: Instant, out: &mut Vec<Vec<u8>>) {
if !self.max_delay.is_zero() {
if self
.last
.is_some_and(|t| now.duration_since(t) > self.max_delay)
{
self.buf.clear();
}
self.last = Some(now);
}
self.buf.extend_from_slice(data);
while self.buf.len() > 1 && self.buf.len() >= self.buf[1] as usize + 5 {
let n = self.buf[1] as usize + 5;
out.push(self.buf.drain(..n).collect());
}
}
fn flush(&mut self) -> Option<Vec<u8>> {
self.buf.clear();
None
}
}
#[cfg(test)]
mod tests;