1use crate::config::{StandardSmallPayload, runtime_device_identifier, runtime_string_precision};
10use crate::queue::ByteCost;
11use crate::{
12 MessageClass, MessageDataType, MessageElement, TelemetryError, TelemetryResult,
13 config::{DataEndpoint, DataType},
14 data_type_size, get_data_type, get_message_name, message_meta,
15 router::LeBytes,
16};
17use crate::{impl_data_as_prim, impl_from_prim_slices, impl_ledecode_auto};
18use alloc::{string::String, string::ToString, sync::Arc, vec, vec::Vec};
19use core::any::TypeId;
20use core::fmt::{Formatter, Write};
21use core::sync::atomic::{AtomicU32, Ordering};
22const EPOCH_MS_THRESHOLD: u64 = 1_000_000_000_000; const DEFAULT_STRING_CAPACITY: usize = 96;
35
36static PACKET_NONCE_COUNTER: AtomicU32 = AtomicU32::new(0xA5C3_1F27);
39
40#[inline]
41fn next_packet_nonce() -> u16 {
42 let mut x = PACKET_NONCE_COUNTER.fetch_add(0x9E37_79B9, Ordering::Relaxed);
43 x ^= x << 13;
44 x ^= x >> 17;
45 x ^= x << 5;
46 (x as u16) | 1
47}
48
49#[derive(Clone, Debug, PartialEq, Eq)]
58pub struct Packet {
59 ty: DataType,
61
62 data_size: usize,
67
68 sender: Arc<str>,
70
71 endpoints: Arc<[DataEndpoint]>,
73
74 timestamp: u64,
81
82 nonce: u16,
86
87 payload: StandardSmallPayload,
89
90 wire_shape: Option<MessageElement>,
92
93 wire_target_senders: Option<Arc<[u64]>>,
95}
96
97#[inline]
107const fn element_width(dt: MessageDataType) -> usize {
108 match dt {
109 MessageDataType::UInt8 | MessageDataType::Int8 | MessageDataType::Bool => 1,
110 MessageDataType::UInt16 | MessageDataType::Int16 => 2,
111 MessageDataType::UInt32 | MessageDataType::Int32 | MessageDataType::Float32 => 4,
112 MessageDataType::UInt64 | MessageDataType::Int64 | MessageDataType::Float64 => 8,
113 MessageDataType::UInt128 | MessageDataType::Int128 => 16,
114 MessageDataType::String | MessageDataType::Binary => 1,
116 MessageDataType::NoData => 0,
117 }
118}
119
120#[inline]
124pub fn hash_bytes_u64(mut h: u64, bytes: &[u8]) -> u64 {
125 const PRIME: u64 = 0x9E37_79B1;
127 for &b in bytes {
128 h ^= b as u64;
129 h = h.wrapping_mul(PRIME);
130 h ^= h >> 27;
132 }
133 h
134}
135
136#[inline]
142pub(crate) fn sender_address_u32(sender: &str) -> u32 {
143 if let Some(address) = sender
146 .strip_prefix("@addr:")
147 .and_then(|s| s.parse::<u32>().ok())
148 {
149 return address;
150 }
151 let raw = hash_bytes_u64(0xA6D3_8C21_4B7F_19E5, sender.as_bytes()) as u32;
152 if raw == 0 { 1 } else { raw }
153}
154
155#[inline]
161fn validate_dynamic_len_and_content_for_element(
162 element: MessageElement,
163 bytes: &[u8],
164) -> TelemetryResult<()> {
165 match element.data_type() {
166 MessageDataType::String => {
167 let end = bytes
168 .iter()
169 .rposition(|&b| b != 0)
170 .map(|i| i + 1)
171 .unwrap_or(0);
172 if end > 0 {
173 core::str::from_utf8(&bytes[..end]).map_err(|_| TelemetryError::InvalidUtf8)?;
174 }
175 Ok(())
176 }
177 MessageDataType::Binary => Ok(()),
178 dt => {
179 let w = element_width(dt);
180 if w == 0 || !bytes.len().is_multiple_of(w) {
181 return Err(TelemetryError::SizeMismatch {
182 expected: w,
183 got: bytes.len(),
184 });
185 }
186 Ok(())
187 }
188 }
189}
190
191trait LeDecode: Sized {
196 const WIDTH: usize;
197 fn from_le(slice: &[u8]) -> Self;
198}
199
200impl_ledecode_auto!(f32);
201impl_ledecode_auto!(f64);
202
203impl_ledecode_auto!(u16);
204impl_ledecode_auto!(u32);
205impl_ledecode_auto!(u64);
206impl_ledecode_auto!(u128);
207
208impl_ledecode_auto!(i16);
209impl_ledecode_auto!(i32);
210impl_ledecode_auto!(i64);
211impl_ledecode_auto!(i128);
212
213impl_ledecode_auto!(u8);
215impl_ledecode_auto!(i8);
216
217impl Packet {
222 #[inline]
239 pub(crate) fn validate_payload_against_element(
240 element: MessageElement,
241 payload: &[u8],
242 ) -> TelemetryResult<()> {
243 match element {
244 MessageElement::Static(need, dt, _) => {
245 let need = need * data_type_size(dt);
246 if payload.len() != need {
247 return Err(TelemetryError::SizeMismatch {
248 expected: need,
249 got: payload.len(),
250 });
251 }
252 Ok(())
253 }
254 MessageElement::Dynamic(_, _) => {
255 validate_dynamic_len_and_content_for_element(element, payload)
256 }
257 }
258 }
259
260 #[inline]
289 #[allow(clippy::too_many_arguments)]
290 #[cfg(test)]
291 pub(crate) fn new_with_wire_contract(
292 ty: DataType,
293 endpoints: &[DataEndpoint],
294 sender: &str,
295 timestamp: u64,
296 nonce: u16,
297 payload: Arc<[u8]>,
298 wire_shape: Option<MessageElement>,
299 wire_target_senders: Arc<[u64]>,
300 ) -> TelemetryResult<Self> {
301 if endpoints.is_empty() {
302 return Err(TelemetryError::EmptyEndpoints);
303 }
304
305 let element = wire_shape.unwrap_or_else(|| message_meta(ty).element);
306 Self::validate_payload_against_element(element, &payload)?;
307
308 Ok(Self {
309 ty,
310 data_size: payload.len(),
311 sender: sender.into(),
312 endpoints: Arc::<[DataEndpoint]>::from(endpoints),
313 timestamp,
314 nonce,
315 payload: StandardSmallPayload::from_arc(payload),
316 wire_shape,
317 wire_target_senders: (!wire_target_senders.is_empty()).then_some(wire_target_senders),
318 })
319 }
320
321 #[inline]
323 #[allow(clippy::too_many_arguments)]
324 pub(crate) fn new_with_owned_wire_contract(
325 ty: DataType,
326 endpoints: Arc<[DataEndpoint]>,
327 sender: Arc<str>,
328 timestamp: u64,
329 nonce: u16,
330 payload: StandardSmallPayload,
331 wire_shape: Option<MessageElement>,
332 wire_target_senders: Option<Arc<[u64]>>,
333 ) -> TelemetryResult<Self> {
334 if endpoints.is_empty() {
335 return Err(TelemetryError::EmptyEndpoints);
336 }
337 let element = wire_shape.unwrap_or_else(|| message_meta(ty).element);
338 Self::validate_payload_against_element(element, &payload)?;
339 Ok(Self {
340 ty,
341 data_size: payload.len(),
342 sender,
343 endpoints,
344 timestamp,
345 nonce,
346 payload,
347 wire_shape,
348 wire_target_senders,
349 })
350 }
351
352 pub fn new(
377 ty: DataType,
378 endpoints: &[DataEndpoint],
379 sender: &str,
380 timestamp: u64,
381 payload: Arc<[u8]>,
382 ) -> TelemetryResult<Self> {
383 Self::new_with_nonce(
384 ty,
385 endpoints,
386 sender,
387 timestamp,
388 next_packet_nonce(),
389 payload,
390 )
391 }
392
393 pub(crate) fn new_local(
396 ty: DataType,
397 endpoints: Arc<[DataEndpoint]>,
398 sender: Arc<str>,
399 timestamp: u64,
400 payload: StandardSmallPayload,
401 ) -> TelemetryResult<Self> {
402 Self::new_with_owned_wire_contract(
403 ty,
404 endpoints,
405 sender,
406 timestamp,
407 next_packet_nonce(),
408 payload,
409 None,
410 None,
411 )
412 }
413
414 pub fn new_with_nonce(
420 ty: DataType,
421 endpoints: &[DataEndpoint],
422 sender: &str,
423 timestamp: u64,
424 nonce: u16,
425 payload: Arc<[u8]>,
426 ) -> TelemetryResult<Self> {
427 Self::new_with_owned_wire_contract(
428 ty,
429 Arc::from(endpoints),
430 Arc::from(sender),
431 timestamp,
432 nonce,
433 StandardSmallPayload::from_arc(payload),
434 None,
435 None,
436 )
437 }
438
439 #[inline]
447 fn effective_element(&self) -> MessageElement {
448 self.wire_shape
449 .unwrap_or_else(|| message_meta(self.ty).element)
450 }
451
452 #[inline]
457 fn effective_data_type(&self) -> MessageDataType {
458 self.effective_element().data_type()
459 }
460
461 #[inline]
467 fn effective_message_class(&self) -> MessageClass {
468 self.effective_element().message_type()
469 }
470
471 #[inline]
483 pub fn packet_id(&self) -> u64 {
484 hash_bytes_u64(self.packet_id_header(), self.payload())
485 }
486
487 #[cfg(feature = "latest-value-egress")]
488 pub(crate) fn scoped_packet_id(&self) -> TelemetryResult<u64> {
489 let header = self.packet_id_header();
490 #[cfg(feature = "compact-packet-store")]
491 {
492 self.payload
493 .with_slice(|bytes| hash_bytes_u64(header, bytes))
494 }
495 #[cfg(not(feature = "compact-packet-store"))]
496 {
497 Ok(hash_bytes_u64(header, self.payload()))
498 }
499 }
500
501 #[inline]
502 fn packet_id_header(&self) -> u64 {
503 let mut h: u64 = 0x9E37_79B9_7F4A_7C15;
505
506 h = hash_bytes_u64(h, &sender_address_u32(self.sender.as_ref()).to_le_bytes());
509
510 h = hash_bytes_u64(h, &self.ty.0.to_le_bytes());
512
513 if self
517 .endpoints
518 .windows(2)
519 .all(|pair| pair[0].as_u32() < pair[1].as_u32())
520 {
521 for ep in self.endpoints.iter() {
522 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
523 }
524 } else {
525 for ep in crate::wire_format::unique_sorted_endpoints(&self.endpoints) {
526 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
527 }
528 }
529
530 h = hash_bytes_u64(h, &self.timestamp.to_le_bytes());
532 h = hash_bytes_u64(h, &self.nonce.to_le_bytes());
533 h = hash_bytes_u64(h, &(self.data_size as u64).to_le_bytes());
534
535 h
536 }
537
538 #[inline]
541 fn _as_le_bytes<T>(&self, expected_kind: MessageDataType) -> TelemetryResult<Vec<T>>
542 where
543 T: LeDecode,
544 {
545 self.ensure_kind(expected_kind)?;
546
547 let bytes: &[u8] = self.payload();
548 let width = T::WIDTH;
549
550 if !bytes.len().is_multiple_of(width) {
551 return Err(TelemetryError::SizeMismatch {
553 expected: (bytes.len() / width) * width,
554 got: bytes.len(),
555 });
556 }
557
558 let count = bytes.len() / width;
559 let mut out = Vec::with_capacity(count);
560
561 for chunk in bytes.chunks_exact(width) {
562 out.push(T::from_le(chunk));
563 }
564
565 Ok(out)
566 }
567
568 pub fn validate(&self) -> TelemetryResult<()> {
580 if self.endpoints.is_empty() {
581 return Err(TelemetryError::EmptyEndpoints);
582 }
583 if self.payload.len() != self.data_size {
584 return Err(TelemetryError::SizeMismatch {
585 expected: self.data_size,
586 got: self.payload.len(),
587 });
588 }
589
590 Self::validate_payload_against_element(self.effective_element(), &self.payload)
591 }
592
593 #[inline]
597 pub fn data_type(&self) -> DataType {
598 self.ty
599 }
600
601 #[inline]
604 pub fn sender(&self) -> &str {
605 &self.sender
606 }
607
608 #[inline]
610 pub fn endpoints(&self) -> &[DataEndpoint] {
611 &self.endpoints
612 }
613
614 #[inline]
616 pub fn timestamp(&self) -> u64 {
617 self.timestamp
618 }
619
620 #[inline]
622 pub fn nonce(&self) -> u16 {
623 self.nonce
624 }
625
626 #[inline]
628 pub fn data_size(&self) -> usize {
629 self.data_size
630 }
631
632 #[inline]
634 pub fn payload(&self) -> &[u8] {
635 &self.payload
636 }
637
638 pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
640 #[cfg(feature = "compact-packet-store")]
641 {
642 self.payload.read_range(offset, out)
643 }
644 #[cfg(not(feature = "compact-packet-store"))]
645 {
646 let end = offset
647 .checked_add(out.len())
648 .filter(|&n| n <= self.data_size)
649 .ok_or(TelemetryError::Unpack("payload range out of bounds"))?;
650 out.copy_from_slice(&self.payload()[offset..end]);
651 Ok(())
652 }
653 }
654
655 #[inline]
660 pub(crate) fn wire_shape(&self) -> Option<MessageElement> {
661 self.wire_shape
662 }
663
664 #[inline]
670 pub(crate) fn wire_target_senders(&self) -> &[u64] {
671 self.wire_target_senders.as_deref().unwrap_or(&[])
672 }
673
674 #[inline]
676 pub fn with_nonce(mut self, nonce: u16) -> Self {
677 self.nonce = nonce;
678 self
679 }
680
681 pub fn header_string(&self) -> String {
688 let mut out = String::with_capacity(DEFAULT_STRING_CAPACITY);
689
690 let _ = write!(
691 &mut out,
692 "Type: {}, Data Size: {}, Sender: {}, Endpoints: [",
693 get_message_name(self.ty),
694 self.data_size,
695 self.sender.as_ref(),
696 );
697 for (i, ep) in self.endpoints.iter().enumerate() {
698 if i != 0 {
699 out.push_str(", ");
700 }
701 #[cfg(feature = "std")]
702 {
703 let endpoint_name = ep.as_str();
704 out.push_str(endpoint_name.as_ref());
705 }
706 #[cfg(not(feature = "std"))]
707 out.push_str(ep.as_str());
708 }
709 out.push_str("], Timestamp: ");
710 let _ = write!(&mut out, "{}", self.timestamp);
711
712 out.push_str(" (");
713 append_human_time(&mut out, self.timestamp);
714 out.push(')');
715 out
716 }
717
718 #[inline]
726 pub fn data_as_utf8_ref(&self) -> Option<&str> {
727 if self.effective_data_type() != MessageDataType::String {
728 return None;
729 }
730 let bytes = &self.payload;
731 let end = bytes.iter().rposition(|&b| b != 0).map(|i| i + 1)?;
732 core::str::from_utf8(&bytes[..end]).ok()
733 }
734
735 #[inline]
741 fn data_to_string<T>(&self, s: &mut String)
742 where
743 T: LeBytes + core::fmt::Display + 'static,
744 {
745 let mut first = true;
746
747 for chunk in self.payload.chunks(T::WIDTH) {
748 if chunk.len() != T::WIDTH {
749 break;
750 }
751 if !first {
752 s.push_str(", ");
753 }
754 first = false;
755
756 let v = T::from_le_slice(chunk);
757
758 if TypeId::of::<T>() == TypeId::of::<f32>() || TypeId::of::<T>() == TypeId::of::<f64>()
760 {
761 let _ = write!(s, "{:.*}", runtime_string_precision(), v);
763 } else {
764 let _ = write!(s, "{v}");
765 }
766 }
767 }
768
769 pub fn as_string(&self) -> String {
777 let mut s = String::from("{");
778 s.push_str(&self.header_string());
779
780 if self.payload.is_empty() {
781 s.push_str(", Data: (<NoData>)}");
782 return s;
783 }
784
785 match self.effective_message_class() {
786 MessageClass::Data => {
787 s.push_str(", Data: (");
788 }
789 MessageClass::Error => {
790 s.push_str(", Error: (");
791 }
792 MessageClass::Warning => {
793 s.push_str(", Warning: (");
794 }
795 }
796
797 if let Some(msg) = self.data_as_utf8_ref() {
798 s.push('"');
799 s.push_str(msg);
800 s.push_str("\")}");
801 return s;
802 }
803
804 match self.effective_data_type() {
805 MessageDataType::Float64 => {
806 self.data_to_string::<f64>(&mut s);
807 }
808 MessageDataType::Float32 => {
809 self.data_to_string::<f32>(&mut s);
810 }
811 MessageDataType::UInt128 => {
812 self.data_to_string::<u128>(&mut s);
813 }
814 MessageDataType::UInt64 => {
815 self.data_to_string::<u64>(&mut s);
816 }
817 MessageDataType::UInt32 => {
818 self.data_to_string::<u32>(&mut s);
819 }
820 MessageDataType::UInt16 => {
821 self.data_to_string::<u16>(&mut s);
822 }
823 MessageDataType::UInt8 => {
824 self.data_to_string::<u8>(&mut s);
825 }
826 MessageDataType::Int128 => {
827 self.data_to_string::<i128>(&mut s);
828 }
829 MessageDataType::Int64 => {
830 self.data_to_string::<i64>(&mut s);
831 }
832 MessageDataType::Int32 => {
833 self.data_to_string::<i32>(&mut s);
834 }
835 MessageDataType::Int16 => {
836 self.data_to_string::<i16>(&mut s);
837 }
838 MessageDataType::Int8 => {
839 self.data_to_string::<i8>(&mut s);
840 }
841 MessageDataType::Bool => {
842 let mut it = self.payload.iter().peekable();
844 while let Some(b) = it.next() {
845 let _ = write!(s, "{}", *b != 0);
846 if it.peek().is_some() {
847 s.push_str(", ");
848 }
849 }
850 }
851 MessageDataType::String => {
852 }
854 MessageDataType::Binary => return self.to_hex_string(),
855 MessageDataType::NoData => {
856 s.push_str("<no data>");
857 }
858 }
859
860 s.push_str(")}");
861 s
862 }
863
864 pub fn to_hex_string(&self) -> String {
872 let mut s = self.header_string();
874 s.push_str(", Data (hex):");
875
876 if !self.payload.is_empty() {
877 s.reserve(self.payload.len().saturating_mul(5));
879 for &b in self.payload.iter() {
880 let _ = write!(&mut s, " 0x{:02x}", b);
881 }
882 }
883 s
884 }
885
886 #[inline]
892 fn ensure_kind(&self, expected: MessageDataType) -> TelemetryResult<()> {
893 let dt = self.effective_data_type();
894 if dt != expected {
895 return Err(TelemetryError::TypeMismatch {
896 expected: data_type_size(expected),
897 got: data_type_size(dt),
898 });
899 }
900 Ok(())
901 }
902
903 impl_data_as_prim! {
904 data_as_f32, f32, MessageDataType::Float32;
905 data_as_f64, f64, MessageDataType::Float64;
906
907 data_as_u8, u8, MessageDataType::UInt8;
908 data_as_u16, u16, MessageDataType::UInt16;
909 data_as_u32, u32, MessageDataType::UInt32;
910 data_as_u64, u64, MessageDataType::UInt64;
911 data_as_u128, u128, MessageDataType::UInt128;
912
913 data_as_i8, i8, MessageDataType::Int8;
914 data_as_i16, i16, MessageDataType::Int16;
915 data_as_i32, i32, MessageDataType::Int32;
916 data_as_i64, i64, MessageDataType::Int64;
917 data_as_i128, i128, MessageDataType::Int128;
918 }
919
920 #[inline]
922 pub fn data_as_bool(&self) -> TelemetryResult<Vec<bool>> {
923 self.ensure_kind(MessageDataType::Bool)?;
924 Ok(self.payload.iter().map(|&b| b != 0).collect())
925 }
926
927 #[inline]
929 pub fn data_as_string(&self) -> TelemetryResult<String> {
930 self.ensure_kind(MessageDataType::String)?;
931
932 let bytes = &self.payload;
933 let end = bytes
934 .iter()
935 .rposition(|&b| b != 0)
936 .map(|i| i + 1)
937 .unwrap_or(0);
938
939 if end == 0 {
940 return Ok(String::new());
941 }
942
943 let s = core::str::from_utf8(&bytes[..end])
944 .map_err(|_| TelemetryError::InvalidUtf8)?
945 .to_string();
946 Ok(s)
947 }
948
949 #[inline]
951 pub fn data_as_binary(&self) -> TelemetryResult<Vec<u8>> {
952 self.ensure_kind(MessageDataType::Binary)?;
953 Ok(self.payload.to_vec())
954 }
955
956 fn from_prim_le_slice_with_sender<T>(
962 ty: DataType,
963 values: &[T],
964 endpoints: &[DataEndpoint],
965 sender: &str,
966 timestamp: u64,
967 ) -> TelemetryResult<Self>
968 where
969 T: Copy + 'static,
970 {
971 let dt = get_data_type(ty);
972
973 if dt == MessageDataType::Bool
976 || dt == MessageDataType::String
977 || dt == MessageDataType::Binary
978 {
979 return Err(TelemetryError::BadArg);
981 }
982 let element_size = data_type_size(dt);
983
984 if element_size != size_of::<T>() {
986 return Err(TelemetryError::TypeMismatch {
987 expected: element_size,
988 got: size_of::<T>(),
989 });
990 }
991
992 let total_bytes = values.len() * element_size;
993
994 if let MessageElement::Static(exact, _, _) = message_meta(ty).element {
996 let exact_bytes = exact * element_size;
997 if total_bytes != exact_bytes {
998 return Err(TelemetryError::SizeMismatch {
999 expected: exact_bytes,
1000 got: total_bytes,
1001 });
1002 }
1003 }
1004
1005 let mut bytes = vec![0u8; total_bytes];
1006 unsafe { bytes.set_len(total_bytes) };
1007
1008 for (i, v) in values.iter().copied().enumerate() {
1009 let offset = i * element_size;
1010 let dst = &mut bytes[offset..offset + element_size];
1011
1012 unsafe {
1014 core::ptr::copy_nonoverlapping(
1015 &v as *const T as *const u8,
1016 dst.as_mut_ptr(),
1017 element_size,
1018 );
1019 }
1020
1021 #[cfg(target_endian = "big")]
1024 {
1025 dst.reverse();
1026 }
1027 }
1028
1029 Self::new(ty, endpoints, sender, timestamp, Arc::<[u8]>::from(bytes))
1030 }
1031
1032 #[inline]
1035 pub fn from_prim_le_slice<T>(
1036 ty: DataType,
1037 values: &[T],
1038 endpoints: &[DataEndpoint],
1039 timestamp: u64,
1040 ) -> TelemetryResult<Self>
1041 where
1042 T: Copy + 'static,
1043 {
1044 let sender = runtime_device_identifier();
1045 Self::from_prim_le_slice_with_sender(ty, values, endpoints, &sender, timestamp)
1046 }
1047
1048 impl_from_prim_slices! {
1052 from_u8_slice, u8;
1053 from_u16_slice, u16;
1054 from_i8_slice, i8;
1055 from_i16_slice, i16;
1056 from_u32_slice, u32;
1057 from_i32_slice, i32;
1058 from_u64_slice, u64;
1059 from_i64_slice, i64;
1060 from_u128_slice, u128;
1061 from_i128_slice, i128;
1062 from_f32_slice, f32;
1063 from_f64_slice, f64;
1064 }
1065
1066 #[inline]
1068 pub fn from_no_data(
1069 ty: DataType,
1070 endpoints: &[DataEndpoint],
1071 timestamp: u64,
1072 ) -> TelemetryResult<Self> {
1073 let meta = message_meta(ty);
1074 match meta.element {
1075 MessageElement::Static(need, _, _) => {
1076 if need != 0 {
1077 return Err(TelemetryError::SizeMismatch {
1078 expected: need,
1079 got: 0,
1080 });
1081 }
1082 }
1083 MessageElement::Dynamic(_, _) => {
1084 }
1086 }
1087
1088 let sender = runtime_device_identifier();
1089 Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from([]))
1090 }
1091
1092 #[inline]
1094 pub fn from_bool_slice(
1095 ty: DataType,
1096 values: &[bool],
1097 endpoints: &[DataEndpoint],
1098 timestamp: u64,
1099 ) -> TelemetryResult<Self> {
1100 if get_data_type(ty) != MessageDataType::Bool {
1101 return Err(TelemetryError::TypeMismatch {
1102 expected: data_type_size(get_data_type(ty)),
1103 got: size_of::<bool>(),
1104 });
1105 }
1106
1107 let total_bytes = values.len();
1108 if let MessageElement::Static(exact, _, _) = message_meta(ty).element
1109 && total_bytes != exact
1110 {
1111 return Err(TelemetryError::SizeMismatch {
1112 expected: exact,
1113 got: total_bytes,
1114 });
1115 }
1116
1117 let mut bytes = Vec::with_capacity(total_bytes);
1118 bytes.extend(values.iter().map(|b| if *b { 1u8 } else { 0u8 }));
1119
1120 let sender = runtime_device_identifier();
1121 Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from(bytes))
1122 }
1123
1124 #[inline]
1127 pub fn from_str_slice(
1128 ty: DataType,
1129 s: &str,
1130 endpoints: &[DataEndpoint],
1131 timestamp: u64,
1132 ) -> TelemetryResult<Self> {
1133 if get_data_type(ty) != MessageDataType::String {
1134 return Err(TelemetryError::TypeMismatch {
1135 expected: data_type_size(get_data_type(ty)),
1136 got: 1,
1137 });
1138 }
1139
1140 let bytes: Arc<[u8]> = Arc::from(s.as_bytes());
1141 let sender = runtime_device_identifier();
1142 Self::new(ty, endpoints, &sender, timestamp, bytes)
1143 }
1144}
1145
1146#[inline]
1151fn div_mod_u64(n: u64, d: u64) -> (u64, u64) {
1152 (n / d, n % d)
1153}
1154
1155fn civil_from_days(mut z: i64) -> (i32, u32, u32) {
1157 z += 719_468; let era = (if z >= 0 { z } else { z - 146_096 }) / 146_097;
1160 let doe = z - era * 146_097; let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365; let y = (yoe as i32) + era as i32 * 400;
1163 let doy = (doe - (365 * yoe + yoe / 4 - yoe / 100)) as i32; let mp = (5 * doy + 2) / 153; let d = doy - (153 * mp + 2) / 5 + 1; let m = mp + if mp < 10 { 3 } else { -9 }; let y = y + (m <= 2) as i32; (y, m as u32, d as u32)
1169}
1170
1171fn append_human_time(out: &mut String, total_ms: u64) {
1174 if total_ms >= EPOCH_MS_THRESHOLD {
1175 let (secs, sub_ms) = div_mod_u64(total_ms, 1_000);
1177 let days = (secs / 86_400) as i64;
1178 let sod = (secs % 86_400) as u32; let (year, month, day) = civil_from_days(days);
1180 let hour = sod / 3_600;
1181 let min = (sod % 3_600) / 60;
1182 let sec = sod % 60;
1183 let _ = Write::write_fmt(
1184 out,
1185 format_args!(
1186 "{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}Z",
1187 year, month, day, hour, min, sec, sub_ms as u32
1188 ),
1189 );
1190 } else {
1191 let hours = total_ms / 3_600_000;
1193 let minutes = (total_ms % 3_600_000) / 60_000;
1194 let seconds = (total_ms % 60_000) / 1_000;
1195 let milliseconds = total_ms % 1_000;
1196 if hours > 0 {
1197 let _ = Write::write_fmt(
1198 out,
1199 format_args!("{hours}h {minutes:02}m {seconds:02}s {milliseconds:03}ms"),
1200 );
1201 } else if minutes > 0 {
1202 let _ = Write::write_fmt(
1203 out,
1204 format_args!("{minutes}m {seconds:02}s {milliseconds:03}ms"),
1205 );
1206 } else {
1207 let _ = Write::write_fmt(out, format_args!("{seconds}s {milliseconds:03}ms"));
1208 }
1209 }
1210}
1211
1212impl core::fmt::Display for Packet {
1217 #[inline]
1218 fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
1219 f.write_str(&Packet::as_string(self))
1220 }
1221}
1222
1223impl ByteCost for Packet {
1224 #[cfg(feature = "compact-packet-store")]
1225 fn park_for_queue(&mut self) -> TelemetryResult<()> {
1226 self.payload.park_for_queue()
1227 }
1228 #[inline]
1229 fn byte_cost(&self) -> usize {
1230 size_of::<Self>()
1231 + self.sender.len()
1232 + self.endpoints.len() * size_of::<DataEndpoint>()
1233 + self.payload.byte_cost()
1234 }
1235}
1236
1237#[cfg(feature = "compact-packet-store")]
1240pub struct StoredPacket {
1241 header: Packet,
1242 payload: crate::packet_store::StoredBytes,
1243}
1244#[cfg(feature = "compact-packet-store")]
1245impl StoredPacket {
1246 pub fn data_type(&self) -> DataType {
1247 self.header.data_type()
1248 }
1249 pub fn sender(&self) -> &str {
1250 self.header.sender()
1251 }
1252 pub fn data_size(&self) -> usize {
1253 self.payload.len()
1254 }
1255 pub fn stored_payload_size(&self) -> usize {
1256 self.payload.stored_len()
1257 }
1258 pub fn payload_is_compressed(&self) -> bool {
1259 self.payload.is_compressed()
1260 }
1261 pub fn pin_payload(&self) -> TelemetryResult<crate::packet_store::PinnedBytes> {
1262 self.payload.pin()
1263 }
1264 pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
1265 self.payload.read_range(offset, out)
1266 }
1267 #[cfg(feature = "compact-packet-compression")]
1268 pub fn read_payload_range_bounded(
1269 &self,
1270 codec: &mut crate::packet_store::BoundedPacketCodec,
1271 offset: usize,
1272 out: &mut [u8],
1273 ) -> TelemetryResult<()> {
1274 codec.read_range(&self.payload, offset, out)
1275 }
1276 #[cfg(feature = "compact-packet-compression")]
1277 pub fn read_payload_range_with_codec(
1278 &self,
1279 codec: &mut crate::packet_store::PacketCodec,
1280 offset: usize,
1281 out: &mut [u8],
1282 scratch: &mut [u8],
1283 ) -> TelemetryResult<()> {
1284 codec.read_range(&self.payload, offset, out, scratch)
1285 }
1286}
1287#[cfg(feature = "compact-packet-store")]
1288impl Packet {
1289 pub fn into_stored(
1290 mut self,
1291 store: &Arc<crate::packet_store::PacketStore>,
1292 ) -> TelemetryResult<StoredPacket> {
1293 let payload = store.store(self.payload())?;
1294 self.payload = StandardSmallPayload::new(&[]);
1295 Ok(StoredPacket {
1296 header: self,
1297 payload,
1298 })
1299 }
1300 #[cfg(feature = "compact-packet-compression")]
1301 pub fn into_stored_bounded(
1302 mut self,
1303 store: &Arc<crate::packet_store::PacketStore>,
1304 codec: &mut crate::packet_store::BoundedPacketCodec,
1305 ) -> TelemetryResult<StoredPacket> {
1306 let payload = codec.store(store, self.payload())?;
1307 self.payload = StandardSmallPayload::new(&[]);
1308 Ok(StoredPacket {
1309 header: self,
1310 payload,
1311 })
1312 }
1313 #[cfg(feature = "compact-packet-compression")]
1314 pub fn into_stored_compressed(
1315 mut self,
1316 store: &Arc<crate::packet_store::PacketStore>,
1317 codec: &mut crate::packet_store::PacketCodec,
1318 scratch: &mut [u8],
1319 ) -> TelemetryResult<StoredPacket> {
1320 let payload = codec.store(store, self.payload(), scratch)?;
1321 self.payload = StandardSmallPayload::new(&[]);
1322 Ok(StoredPacket {
1323 header: self,
1324 payload,
1325 })
1326 }
1327}