#![forbid(unsafe_code)]
use std::collections::HashMap;
use matter_codec::Value;
use crate::error::ImError;
use crate::path::AttributePath;
use crate::read::{ReportData, ReportOp};
pub const DEFAULT_MAX_ELEMENTS: usize = 100_000;
pub const DEFAULT_MAX_BYTES: usize = 4 * 1024 * 1024;
fn estimate_value_bytes(v: &Value) -> usize {
const SCALAR: usize = 8;
match v {
Value::Utf8(s) => s.len(),
Value::Bytes(b) => b.len(),
Value::Array(items) => items.iter().map(estimate_value_bytes).sum::<usize>() + SCALAR,
Value::Structure(members) | Value::List(members) => {
members
.iter()
.map(|(_, mv)| estimate_value_bytes(mv))
.sum::<usize>()
+ SCALAR
}
_ => SCALAR,
}
}
pub struct ReportAccumulator {
order: Vec<AttributePath>,
values: HashMap<(u16, u32, u32), Value>,
versions: HashMap<(u16, u32, u32), Option<u32>>,
bytes: usize,
max_elements: usize,
max_bytes: usize,
}
impl Default for ReportAccumulator {
fn default() -> Self {
Self::with_limits(DEFAULT_MAX_ELEMENTS, DEFAULT_MAX_BYTES)
}
}
impl ReportAccumulator {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn with_limits(max_elements: usize, max_bytes: usize) -> Self {
Self {
order: Vec::new(),
values: HashMap::new(),
versions: HashMap::new(),
bytes: 0,
max_elements,
max_bytes,
}
}
fn overflow(&self) -> ImError {
ImError::AccumulatorOverflow {
elements: self.order.len(),
bytes: self.bytes,
max_elements: self.max_elements,
max_bytes: self.max_bytes,
}
}
pub fn push(&mut self, report: ReportData) -> Result<(), ImError> {
for item in report.items {
let key = (item.path.endpoint, item.path.cluster, item.path.attribute);
let item_bytes = estimate_value_bytes(&item.value);
if !self.values.contains_key(&key) && self.order.len() >= self.max_elements {
return Err(self.overflow());
}
if self.bytes.saturating_add(item_bytes) > self.max_bytes {
return Err(self.overflow());
}
match item.op {
ReportOp::Replace => {
let newer = match (self.versions.get(&key), item.data_version) {
(Some(Some(old)), Some(new)) => new >= *old,
_ => true, };
if newer {
if !self.values.contains_key(&key) {
self.order.push(item.path);
} else if let Some(prev) = self.values.get(&key) {
self.bytes = self.bytes.saturating_sub(estimate_value_bytes(prev));
}
self.bytes = self.bytes.saturating_add(item_bytes);
self.values.insert(key, item.value);
self.versions.insert(key, item.data_version);
}
}
ReportOp::Append => {
if !self.values.contains_key(&key) {
self.order.push(item.path);
self.values.insert(key, Value::Array(Vec::new()));
self.versions.insert(key, item.data_version);
}
self.bytes = self.bytes.saturating_add(item_bytes);
match self.values.get_mut(&key) {
Some(Value::Array(list)) => list.push(item.value),
Some(slot) => *slot = Value::Array(vec![item.value]),
None => {}
}
}
}
}
Ok(())
}
#[must_use]
pub fn finish(mut self) -> Vec<(AttributePath, Value)> {
let mut out = Vec::with_capacity(self.order.len());
for path in std::mem::take(&mut self.order) {
let key = (path.endpoint, path.cluster, path.attribute);
if let Some(v) = self.values.remove(&key) {
out.push((path, v));
}
}
out
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used, clippy::expect_used)]
use super::*;
use crate::read::AttributeReportItem;
fn report(items: Vec<AttributeReportItem>) -> ReportData {
ReportData {
items,
subscription_id: None,
more_chunked_messages: false,
suppress_response: false,
events: Vec::new(),
}
}
fn ap(endpoint: u16, cluster: u32, attribute: u32) -> AttributePath {
AttributePath {
endpoint,
cluster,
attribute,
}
}
fn replace(p: AttributePath, v: Value) -> AttributeReportItem {
AttributeReportItem {
path: p,
op: ReportOp::Replace,
value: v,
data_version: None,
}
}
fn append(p: AttributePath, v: Value) -> AttributeReportItem {
AttributeReportItem {
path: p,
op: ReportOp::Append,
value: v,
data_version: None,
}
}
#[test]
fn message_level_merge_preserves_order() {
let mut acc = ReportAccumulator::new();
acc.push(report(vec![replace(
ap(0, 0x28, 0x0002),
Value::Uint(5010),
)]))
.unwrap();
acc.push(report(vec![replace(
ap(1, 0x06, 0x0000),
Value::Bool(true),
)]))
.unwrap();
let out = acc.finish();
assert_eq!(out.len(), 2);
assert_eq!(out[0].0, ap(0, 0x28, 0x0002));
assert_eq!(out[0].1, Value::Uint(5010));
assert_eq!(out[1].0, ap(1, 0x06, 0x0000));
assert_eq!(out[1].1, Value::Bool(true));
}
#[test]
fn list_append_after_empty_replace() {
let mut acc = ReportAccumulator::new();
let p = ap(0, 0x1d, 0x0003);
acc.push(report(vec![replace(p, Value::Array(Vec::new()))]))
.unwrap();
acc.push(report(vec![append(p, Value::Uint(1))])).unwrap();
acc.push(report(vec![append(p, Value::Uint(2))])).unwrap();
let out = acc.finish();
assert_eq!(out.len(), 1);
assert_eq!(out[0].1, Value::Array(vec![Value::Uint(1), Value::Uint(2)]));
}
#[test]
fn append_without_base_starts_empty() {
let mut acc = ReportAccumulator::new();
let p = ap(0, 0x1d, 0x0003);
acc.push(report(vec![append(p, Value::Uint(9))])).unwrap();
assert_eq!(acc.finish()[0].1, Value::Array(vec![Value::Uint(9)]));
}
#[test]
fn append_onto_non_array_coerces_instead_of_dropping() {
let mut acc = ReportAccumulator::new();
let p = ap(0, 0x1d, 0x0003);
acc.push(report(vec![replace(p, Value::Uint(1))])).unwrap();
acc.push(report(vec![append(p, Value::Uint(2))])).unwrap();
assert_eq!(acc.finish()[0].1, Value::Array(vec![Value::Uint(2)]));
}
#[test]
fn newest_data_version_wins() {
let mut acc = ReportAccumulator::new();
let p = ap(0, 0x28, 0x0000);
acc.push(report(vec![AttributeReportItem {
path: p,
op: ReportOp::Replace,
value: Value::Uint(1),
data_version: Some(5),
}]))
.unwrap();
acc.push(report(vec![AttributeReportItem {
path: p,
op: ReportOp::Replace,
value: Value::Uint(2),
data_version: Some(3),
}]))
.unwrap();
assert_eq!(
acc.finish()[0].1,
Value::Uint(1),
"older DataVersion must not overwrite"
);
}
#[test]
fn finish_moves_values_preserving_content_and_order() {
let mut acc = ReportAccumulator::new();
let p0 = ap(0, 0x28, 0x0001);
let p1 = ap(1, 0x06, 0x0000);
let p2 = ap(2, 0x1d, 0x0003);
let v0 = Value::Utf8(String::from("VendorName"));
let v1 = Value::Bytes(vec![0xde, 0xad, 0xbe, 0xef]);
let v2 = Value::Array(vec![Value::Uint(1), Value::Utf8(String::from("x"))]);
acc.push(report(vec![
replace(p0, v0.clone()),
replace(p1, v1.clone()),
replace(p2, v2.clone()),
]))
.unwrap();
let out = acc.finish();
assert_eq!(
out,
vec![(p0, v0), (p1, v1), (p2, v2)],
"moved-out set must match inserted (path, value) pairs in first-seen order"
);
}
use proptest::prelude::*;
#[test]
fn element_ceiling_is_enforced() {
let mut acc = ReportAccumulator::with_limits(3, usize::MAX);
for i in 0..3u32 {
acc.push(report(vec![replace(
ap(0, 0x06, i),
Value::Uint(u64::from(i)),
)]))
.expect("within element cap");
}
let err = acc
.push(report(vec![replace(ap(0, 0x06, 99), Value::Uint(1))]))
.expect_err("4th distinct element must exceed the cap");
assert!(
matches!(
err,
ImError::AccumulatorOverflow {
max_elements: 3,
..
}
),
"expected AccumulatorOverflow, got {err:?}"
);
}
#[test]
fn byte_ceiling_is_enforced() {
let mut acc = ReportAccumulator::with_limits(usize::MAX, 16);
let err = acc
.push(report(vec![replace(
ap(0, 0x28, 0x0001),
Value::Bytes(vec![0u8; 1024]),
)]))
.expect_err("1 KiB value must exceed a 16-byte cap");
assert!(
matches!(err, ImError::AccumulatorOverflow { max_bytes: 16, .. }),
"expected AccumulatorOverflow, got {err:?}"
);
}
#[test]
fn normal_sized_report_set_is_ok() {
let mut acc = ReportAccumulator::new();
for i in 0..200u32 {
acc.push(report(vec![replace(
ap(0, 0x28, i),
Value::Utf8(String::from("a-realistic-attribute-value")),
)]))
.expect("200 small attributes are well within the default ceiling");
}
assert_eq!(acc.finish().len(), 200);
}
proptest! {
#[test]
fn message_chunking_is_order_preserving(
attrs in proptest::collection::vec((0u16..4, 0u32..8, 0u32..8, 0u64..1000), 1..20),
) {
let mut seen = std::collections::HashSet::new();
let unique: Vec<_> = attrs.into_iter()
.filter(|(e, c, a, _)| seen.insert((*e, *c, *a)))
.collect();
let mut whole = ReportAccumulator::new();
whole.push(report(
unique.iter().map(|&(e, c, a, v)| replace(ap(e, c, a), Value::Uint(v))).collect(),
)).unwrap();
let whole_out = whole.finish();
let mut split = ReportAccumulator::new();
for &(e, c, a, v) in &unique {
split.push(report(vec![replace(ap(e, c, a), Value::Uint(v))])).unwrap();
}
let split_out = split.finish();
prop_assert_eq!(&whole_out, &split_out);
for (i, &(e, c, a, _)) in unique.iter().enumerate() {
prop_assert_eq!(split_out[i].0, ap(e, c, a));
}
}
#[test]
fn appends_build_list_in_order(elems in proptest::collection::vec(0u64..1000, 0..30)) {
let p = ap(0, 0x1d, 0x0003);
let mut acc = ReportAccumulator::new();
acc.push(report(vec![replace(p, Value::Array(Vec::new()))])).unwrap();
for &v in &elems {
acc.push(report(vec![append(p, Value::Uint(v))])).unwrap();
}
let out = acc.finish();
let want: Vec<Value> = elems.iter().map(|&v| Value::Uint(v)).collect();
prop_assert_eq!(&out[0].1, &Value::Array(want));
}
}
}