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 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(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(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 fn new_with_nonce(
399 ty: DataType,
400 endpoints: &[DataEndpoint],
401 sender: &str,
402 timestamp: u64,
403 nonce: u16,
404 payload: Arc<[u8]>,
405 ) -> TelemetryResult<Self> {
406 Self::new_with_owned_wire_contract(
407 ty,
408 Arc::from(endpoints),
409 Arc::from(sender),
410 timestamp,
411 nonce,
412 StandardSmallPayload::from_arc(payload),
413 None,
414 None,
415 )
416 }
417
418 #[inline]
426 fn effective_element(&self) -> MessageElement {
427 self.wire_shape.unwrap_or(message_meta(self.ty).element)
428 }
429
430 #[inline]
435 fn effective_data_type(&self) -> MessageDataType {
436 self.effective_element().data_type()
437 }
438
439 #[inline]
445 fn effective_message_class(&self) -> MessageClass {
446 self.effective_element().message_type()
447 }
448
449 #[inline]
461 pub fn packet_id(&self) -> u64 {
462 let mut h: u64 = 0x9E37_79B9_7F4A_7C15;
464
465 h = hash_bytes_u64(h, &sender_address_u32(self.sender.as_ref()).to_le_bytes());
468
469 h = hash_bytes_u64(h, &self.ty.0.to_le_bytes());
471
472 if self
476 .endpoints
477 .windows(2)
478 .all(|pair| pair[0].as_u32() < pair[1].as_u32())
479 {
480 for ep in self.endpoints.iter() {
481 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
482 }
483 } else {
484 for ep in crate::wire_format::unique_sorted_endpoints(&self.endpoints) {
485 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
486 }
487 }
488
489 h = hash_bytes_u64(h, &self.timestamp.to_le_bytes());
491 h = hash_bytes_u64(h, &self.nonce.to_le_bytes());
492 h = hash_bytes_u64(h, &(self.data_size as u64).to_le_bytes());
493
494 h = hash_bytes_u64(h, self.payload());
496
497 h
498 }
499
500 #[inline]
503 fn _as_le_bytes<T>(&self, expected_kind: MessageDataType) -> TelemetryResult<Vec<T>>
504 where
505 T: LeDecode,
506 {
507 self.ensure_kind(expected_kind)?;
508
509 let bytes: &[u8] = self.payload();
510 let width = T::WIDTH;
511
512 if !bytes.len().is_multiple_of(width) {
513 return Err(TelemetryError::SizeMismatch {
515 expected: (bytes.len() / width) * width,
516 got: bytes.len(),
517 });
518 }
519
520 let count = bytes.len() / width;
521 let mut out = Vec::with_capacity(count);
522
523 for chunk in bytes.chunks_exact(width) {
524 out.push(T::from_le(chunk));
525 }
526
527 Ok(out)
528 }
529
530 pub fn validate(&self) -> TelemetryResult<()> {
542 if self.endpoints.is_empty() {
543 return Err(TelemetryError::EmptyEndpoints);
544 }
545 if self.payload.len() != self.data_size {
546 return Err(TelemetryError::SizeMismatch {
547 expected: self.data_size,
548 got: self.payload.len(),
549 });
550 }
551
552 Self::validate_payload_against_element(self.effective_element(), &self.payload)
553 }
554
555 #[inline]
559 pub fn data_type(&self) -> DataType {
560 self.ty
561 }
562
563 #[inline]
566 pub fn sender(&self) -> &str {
567 &self.sender
568 }
569
570 #[inline]
572 pub fn endpoints(&self) -> &[DataEndpoint] {
573 &self.endpoints
574 }
575
576 #[inline]
578 pub fn timestamp(&self) -> u64 {
579 self.timestamp
580 }
581
582 #[inline]
584 pub fn nonce(&self) -> u16 {
585 self.nonce
586 }
587
588 #[inline]
590 pub fn data_size(&self) -> usize {
591 self.data_size
592 }
593
594 #[inline]
596 pub fn payload(&self) -> &[u8] {
597 &self.payload
598 }
599
600 pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
602 #[cfg(feature = "compact-packet-store")]
603 {
604 self.payload.read_range(offset, out)
605 }
606 #[cfg(not(feature = "compact-packet-store"))]
607 {
608 let end = offset
609 .checked_add(out.len())
610 .filter(|&n| n <= self.data_size)
611 .ok_or(TelemetryError::Unpack("payload range out of bounds"))?;
612 out.copy_from_slice(&self.payload()[offset..end]);
613 Ok(())
614 }
615 }
616
617 #[inline]
622 pub(crate) fn wire_shape(&self) -> Option<MessageElement> {
623 self.wire_shape
624 }
625
626 #[inline]
632 pub(crate) fn wire_target_senders(&self) -> &[u64] {
633 self.wire_target_senders.as_deref().unwrap_or(&[])
634 }
635
636 #[inline]
638 pub fn with_nonce(mut self, nonce: u16) -> Self {
639 self.nonce = nonce;
640 self
641 }
642
643 pub fn header_string(&self) -> String {
650 let mut out = String::with_capacity(DEFAULT_STRING_CAPACITY);
651
652 let _ = write!(
653 &mut out,
654 "Type: {}, Data Size: {}, Sender: {}, Endpoints: [",
655 get_message_name(self.ty),
656 self.data_size,
657 self.sender.as_ref(),
658 );
659 for (i, ep) in self.endpoints.iter().enumerate() {
660 if i != 0 {
661 out.push_str(", ");
662 }
663 #[cfg(feature = "std")]
664 {
665 let endpoint_name = ep.as_str();
666 out.push_str(endpoint_name.as_ref());
667 }
668 #[cfg(not(feature = "std"))]
669 out.push_str(ep.as_str());
670 }
671 out.push_str("], Timestamp: ");
672 let _ = write!(&mut out, "{}", self.timestamp);
673
674 out.push_str(" (");
675 append_human_time(&mut out, self.timestamp);
676 out.push(')');
677 out
678 }
679
680 #[inline]
688 pub fn data_as_utf8_ref(&self) -> Option<&str> {
689 if self.effective_data_type() != MessageDataType::String {
690 return None;
691 }
692 let bytes = &self.payload;
693 let end = bytes.iter().rposition(|&b| b != 0).map(|i| i + 1)?;
694 core::str::from_utf8(&bytes[..end]).ok()
695 }
696
697 #[inline]
703 fn data_to_string<T>(&self, s: &mut String)
704 where
705 T: LeBytes + core::fmt::Display + 'static,
706 {
707 let mut first = true;
708
709 for chunk in self.payload.chunks(T::WIDTH) {
710 if chunk.len() != T::WIDTH {
711 break;
712 }
713 if !first {
714 s.push_str(", ");
715 }
716 first = false;
717
718 let v = T::from_le_slice(chunk);
719
720 if TypeId::of::<T>() == TypeId::of::<f32>() || TypeId::of::<T>() == TypeId::of::<f64>()
722 {
723 let _ = write!(s, "{:.*}", runtime_string_precision(), v);
725 } else {
726 let _ = write!(s, "{v}");
727 }
728 }
729 }
730
731 pub fn as_string(&self) -> String {
739 let mut s = String::from("{");
740 s.push_str(&self.header_string());
741
742 if self.payload.is_empty() {
743 s.push_str(", Data: (<NoData>)}");
744 return s;
745 }
746
747 match self.effective_message_class() {
748 MessageClass::Data => {
749 s.push_str(", Data: (");
750 }
751 MessageClass::Error => {
752 s.push_str(", Error: (");
753 }
754 MessageClass::Warning => {
755 s.push_str(", Warning: (");
756 }
757 }
758
759 if let Some(msg) = self.data_as_utf8_ref() {
760 s.push('"');
761 s.push_str(msg);
762 s.push_str("\")}");
763 return s;
764 }
765
766 match self.effective_data_type() {
767 MessageDataType::Float64 => {
768 self.data_to_string::<f64>(&mut s);
769 }
770 MessageDataType::Float32 => {
771 self.data_to_string::<f32>(&mut s);
772 }
773 MessageDataType::UInt128 => {
774 self.data_to_string::<u128>(&mut s);
775 }
776 MessageDataType::UInt64 => {
777 self.data_to_string::<u64>(&mut s);
778 }
779 MessageDataType::UInt32 => {
780 self.data_to_string::<u32>(&mut s);
781 }
782 MessageDataType::UInt16 => {
783 self.data_to_string::<u16>(&mut s);
784 }
785 MessageDataType::UInt8 => {
786 self.data_to_string::<u8>(&mut s);
787 }
788 MessageDataType::Int128 => {
789 self.data_to_string::<i128>(&mut s);
790 }
791 MessageDataType::Int64 => {
792 self.data_to_string::<i64>(&mut s);
793 }
794 MessageDataType::Int32 => {
795 self.data_to_string::<i32>(&mut s);
796 }
797 MessageDataType::Int16 => {
798 self.data_to_string::<i16>(&mut s);
799 }
800 MessageDataType::Int8 => {
801 self.data_to_string::<i8>(&mut s);
802 }
803 MessageDataType::Bool => {
804 let mut it = self.payload.iter().peekable();
806 while let Some(b) = it.next() {
807 let _ = write!(s, "{}", *b != 0);
808 if it.peek().is_some() {
809 s.push_str(", ");
810 }
811 }
812 }
813 MessageDataType::String => {
814 }
816 MessageDataType::Binary => return self.to_hex_string(),
817 MessageDataType::NoData => {
818 s.push_str("<no data>");
819 }
820 }
821
822 s.push_str(")}");
823 s
824 }
825
826 pub fn to_hex_string(&self) -> String {
834 let mut s = self.header_string();
836 s.push_str(", Data (hex):");
837
838 if !self.payload.is_empty() {
839 s.reserve(self.payload.len().saturating_mul(5));
841 for &b in self.payload.iter() {
842 let _ = write!(&mut s, " 0x{:02x}", b);
843 }
844 }
845 s
846 }
847
848 #[inline]
854 fn ensure_kind(&self, expected: MessageDataType) -> TelemetryResult<()> {
855 let dt = self.effective_data_type();
856 if dt != expected {
857 return Err(TelemetryError::TypeMismatch {
858 expected: data_type_size(expected),
859 got: data_type_size(dt),
860 });
861 }
862 Ok(())
863 }
864
865 impl_data_as_prim! {
866 data_as_f32, f32, MessageDataType::Float32;
867 data_as_f64, f64, MessageDataType::Float64;
868
869 data_as_u8, u8, MessageDataType::UInt8;
870 data_as_u16, u16, MessageDataType::UInt16;
871 data_as_u32, u32, MessageDataType::UInt32;
872 data_as_u64, u64, MessageDataType::UInt64;
873 data_as_u128, u128, MessageDataType::UInt128;
874
875 data_as_i8, i8, MessageDataType::Int8;
876 data_as_i16, i16, MessageDataType::Int16;
877 data_as_i32, i32, MessageDataType::Int32;
878 data_as_i64, i64, MessageDataType::Int64;
879 data_as_i128, i128, MessageDataType::Int128;
880 }
881
882 #[inline]
884 pub fn data_as_bool(&self) -> TelemetryResult<Vec<bool>> {
885 self.ensure_kind(MessageDataType::Bool)?;
886 Ok(self.payload.iter().map(|&b| b != 0).collect())
887 }
888
889 #[inline]
891 pub fn data_as_string(&self) -> TelemetryResult<String> {
892 self.ensure_kind(MessageDataType::String)?;
893
894 let bytes = &self.payload;
895 let end = bytes
896 .iter()
897 .rposition(|&b| b != 0)
898 .map(|i| i + 1)
899 .unwrap_or(0);
900
901 if end == 0 {
902 return Ok(String::new());
903 }
904
905 let s = core::str::from_utf8(&bytes[..end])
906 .map_err(|_| TelemetryError::InvalidUtf8)?
907 .to_string();
908 Ok(s)
909 }
910
911 #[inline]
913 pub fn data_as_binary(&self) -> TelemetryResult<Vec<u8>> {
914 self.ensure_kind(MessageDataType::Binary)?;
915 Ok(self.payload.to_vec())
916 }
917
918 fn from_prim_le_slice_with_sender<T>(
924 ty: DataType,
925 values: &[T],
926 endpoints: &[DataEndpoint],
927 sender: &str,
928 timestamp: u64,
929 ) -> TelemetryResult<Self>
930 where
931 T: Copy + 'static,
932 {
933 let dt = get_data_type(ty);
934
935 if dt == MessageDataType::Bool
938 || dt == MessageDataType::String
939 || dt == MessageDataType::Binary
940 {
941 return Err(TelemetryError::BadArg);
943 }
944 let element_size = data_type_size(dt);
945
946 if element_size != size_of::<T>() {
948 return Err(TelemetryError::TypeMismatch {
949 expected: element_size,
950 got: size_of::<T>(),
951 });
952 }
953
954 let total_bytes = values.len() * element_size;
955
956 if let MessageElement::Static(exact, _, _) = message_meta(ty).element {
958 let exact_bytes = exact * element_size;
959 if total_bytes != exact_bytes {
960 return Err(TelemetryError::SizeMismatch {
961 expected: exact_bytes,
962 got: total_bytes,
963 });
964 }
965 }
966
967 let mut bytes = vec![0u8; total_bytes];
968 unsafe { bytes.set_len(total_bytes) };
969
970 for (i, v) in values.iter().copied().enumerate() {
971 let offset = i * element_size;
972 let dst = &mut bytes[offset..offset + element_size];
973
974 unsafe {
976 core::ptr::copy_nonoverlapping(
977 &v as *const T as *const u8,
978 dst.as_mut_ptr(),
979 element_size,
980 );
981 }
982
983 #[cfg(target_endian = "big")]
986 {
987 dst.reverse();
988 }
989 }
990
991 Self::new(ty, endpoints, sender, timestamp, Arc::<[u8]>::from(bytes))
992 }
993
994 #[inline]
997 pub fn from_prim_le_slice<T>(
998 ty: DataType,
999 values: &[T],
1000 endpoints: &[DataEndpoint],
1001 timestamp: u64,
1002 ) -> TelemetryResult<Self>
1003 where
1004 T: Copy + 'static,
1005 {
1006 let sender = runtime_device_identifier();
1007 Self::from_prim_le_slice_with_sender(ty, values, endpoints, &sender, timestamp)
1008 }
1009
1010 impl_from_prim_slices! {
1014 from_u8_slice, u8;
1015 from_u16_slice, u16;
1016 from_i8_slice, i8;
1017 from_i16_slice, i16;
1018 from_u32_slice, u32;
1019 from_i32_slice, i32;
1020 from_u64_slice, u64;
1021 from_i64_slice, i64;
1022 from_u128_slice, u128;
1023 from_i128_slice, i128;
1024 from_f32_slice, f32;
1025 from_f64_slice, f64;
1026 }
1027
1028 #[inline]
1030 pub fn from_no_data(
1031 ty: DataType,
1032 endpoints: &[DataEndpoint],
1033 timestamp: u64,
1034 ) -> TelemetryResult<Self> {
1035 let meta = message_meta(ty);
1036 match meta.element {
1037 MessageElement::Static(need, _, _) => {
1038 if need != 0 {
1039 return Err(TelemetryError::SizeMismatch {
1040 expected: need,
1041 got: 0,
1042 });
1043 }
1044 }
1045 MessageElement::Dynamic(_, _) => {
1046 }
1048 }
1049
1050 let sender = runtime_device_identifier();
1051 Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from([]))
1052 }
1053
1054 #[inline]
1056 pub fn from_bool_slice(
1057 ty: DataType,
1058 values: &[bool],
1059 endpoints: &[DataEndpoint],
1060 timestamp: u64,
1061 ) -> TelemetryResult<Self> {
1062 if get_data_type(ty) != MessageDataType::Bool {
1063 return Err(TelemetryError::TypeMismatch {
1064 expected: data_type_size(get_data_type(ty)),
1065 got: size_of::<bool>(),
1066 });
1067 }
1068
1069 let total_bytes = values.len();
1070 if let MessageElement::Static(exact, _, _) = message_meta(ty).element
1071 && total_bytes != exact
1072 {
1073 return Err(TelemetryError::SizeMismatch {
1074 expected: exact,
1075 got: total_bytes,
1076 });
1077 }
1078
1079 let mut bytes = Vec::with_capacity(total_bytes);
1080 bytes.extend(values.iter().map(|b| if *b { 1u8 } else { 0u8 }));
1081
1082 let sender = runtime_device_identifier();
1083 Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from(bytes))
1084 }
1085
1086 #[inline]
1089 pub fn from_str_slice(
1090 ty: DataType,
1091 s: &str,
1092 endpoints: &[DataEndpoint],
1093 timestamp: u64,
1094 ) -> TelemetryResult<Self> {
1095 if get_data_type(ty) != MessageDataType::String {
1096 return Err(TelemetryError::TypeMismatch {
1097 expected: data_type_size(get_data_type(ty)),
1098 got: 1,
1099 });
1100 }
1101
1102 let bytes: Arc<[u8]> = Arc::from(s.as_bytes());
1103 let sender = runtime_device_identifier();
1104 Self::new(ty, endpoints, &sender, timestamp, bytes)
1105 }
1106}
1107
1108#[inline]
1113fn div_mod_u64(n: u64, d: u64) -> (u64, u64) {
1114 (n / d, n % d)
1115}
1116
1117fn civil_from_days(mut z: i64) -> (i32, u32, u32) {
1119 z += 719_468; let era = (if z >= 0 { z } else { z - 146_096 }) / 146_097;
1122 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;
1125 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)
1131}
1132
1133fn append_human_time(out: &mut String, total_ms: u64) {
1136 if total_ms >= EPOCH_MS_THRESHOLD {
1137 let (secs, sub_ms) = div_mod_u64(total_ms, 1_000);
1139 let days = (secs / 86_400) as i64;
1140 let sod = (secs % 86_400) as u32; let (year, month, day) = civil_from_days(days);
1142 let hour = sod / 3_600;
1143 let min = (sod % 3_600) / 60;
1144 let sec = sod % 60;
1145 let _ = Write::write_fmt(
1146 out,
1147 format_args!(
1148 "{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}Z",
1149 year, month, day, hour, min, sec, sub_ms as u32
1150 ),
1151 );
1152 } else {
1153 let hours = total_ms / 3_600_000;
1155 let minutes = (total_ms % 3_600_000) / 60_000;
1156 let seconds = (total_ms % 60_000) / 1_000;
1157 let milliseconds = total_ms % 1_000;
1158 if hours > 0 {
1159 let _ = Write::write_fmt(
1160 out,
1161 format_args!("{hours}h {minutes:02}m {seconds:02}s {milliseconds:03}ms"),
1162 );
1163 } else if minutes > 0 {
1164 let _ = Write::write_fmt(
1165 out,
1166 format_args!("{minutes}m {seconds:02}s {milliseconds:03}ms"),
1167 );
1168 } else {
1169 let _ = Write::write_fmt(out, format_args!("{seconds}s {milliseconds:03}ms"));
1170 }
1171 }
1172}
1173
1174impl core::fmt::Display for Packet {
1179 #[inline]
1180 fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
1181 f.write_str(&Packet::as_string(self))
1182 }
1183}
1184
1185impl ByteCost for Packet {
1186 #[cfg(feature = "compact-packet-store")]
1187 fn park_for_queue(&mut self) -> TelemetryResult<()> {
1188 self.payload.park_for_queue()
1189 }
1190 #[inline]
1191 fn byte_cost(&self) -> usize {
1192 size_of::<Self>()
1193 + self.sender.len()
1194 + self.endpoints.len() * size_of::<DataEndpoint>()
1195 + self.payload.byte_cost()
1196 }
1197}
1198
1199#[cfg(feature = "compact-packet-store")]
1202pub struct StoredPacket {
1203 header: Packet,
1204 payload: crate::packet_store::StoredBytes,
1205}
1206#[cfg(feature = "compact-packet-store")]
1207impl StoredPacket {
1208 pub fn data_type(&self) -> DataType {
1209 self.header.data_type()
1210 }
1211 pub fn sender(&self) -> &str {
1212 self.header.sender()
1213 }
1214 pub fn data_size(&self) -> usize {
1215 self.payload.len()
1216 }
1217 pub fn stored_payload_size(&self) -> usize {
1218 self.payload.stored_len()
1219 }
1220 pub fn payload_is_compressed(&self) -> bool {
1221 self.payload.is_compressed()
1222 }
1223 pub fn pin_payload(&self) -> TelemetryResult<crate::packet_store::PinnedBytes> {
1224 self.payload.pin()
1225 }
1226 pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
1227 self.payload.read_range(offset, out)
1228 }
1229 #[cfg(feature = "compact-packet-compression")]
1230 pub fn read_payload_range_bounded(
1231 &self,
1232 codec: &mut crate::packet_store::BoundedPacketCodec,
1233 offset: usize,
1234 out: &mut [u8],
1235 ) -> TelemetryResult<()> {
1236 codec.read_range(&self.payload, offset, out)
1237 }
1238 #[cfg(feature = "compact-packet-compression")]
1239 pub fn read_payload_range_with_codec(
1240 &self,
1241 codec: &mut crate::packet_store::PacketCodec,
1242 offset: usize,
1243 out: &mut [u8],
1244 scratch: &mut [u8],
1245 ) -> TelemetryResult<()> {
1246 codec.read_range(&self.payload, offset, out, scratch)
1247 }
1248}
1249#[cfg(feature = "compact-packet-store")]
1250impl Packet {
1251 pub fn into_stored(
1252 mut self,
1253 store: &Arc<crate::packet_store::PacketStore>,
1254 ) -> TelemetryResult<StoredPacket> {
1255 let payload = store.store(self.payload())?;
1256 self.payload = StandardSmallPayload::new(&[]);
1257 Ok(StoredPacket {
1258 header: self,
1259 payload,
1260 })
1261 }
1262 #[cfg(feature = "compact-packet-compression")]
1263 pub fn into_stored_bounded(
1264 mut self,
1265 store: &Arc<crate::packet_store::PacketStore>,
1266 codec: &mut crate::packet_store::BoundedPacketCodec,
1267 ) -> TelemetryResult<StoredPacket> {
1268 let payload = codec.store(store, self.payload())?;
1269 self.payload = StandardSmallPayload::new(&[]);
1270 Ok(StoredPacket {
1271 header: self,
1272 payload,
1273 })
1274 }
1275 #[cfg(feature = "compact-packet-compression")]
1276 pub fn into_stored_compressed(
1277 mut self,
1278 store: &Arc<crate::packet_store::PacketStore>,
1279 codec: &mut crate::packet_store::PacketCodec,
1280 scratch: &mut [u8],
1281 ) -> TelemetryResult<StoredPacket> {
1282 let payload = codec.store(store, self.payload(), scratch)?;
1283 self.payload = StandardSmallPayload::new(&[]);
1284 Ok(StoredPacket {
1285 header: self,
1286 payload,
1287 })
1288 }
1289}