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_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 let mut h: u64 = 0x9E37_79B9_7F4A_7C15;
486
487 h = hash_bytes_u64(h, &sender_address_u32(self.sender.as_ref()).to_le_bytes());
490
491 h = hash_bytes_u64(h, &self.ty.0.to_le_bytes());
493
494 if self
498 .endpoints
499 .windows(2)
500 .all(|pair| pair[0].as_u32() < pair[1].as_u32())
501 {
502 for ep in self.endpoints.iter() {
503 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
504 }
505 } else {
506 for ep in crate::wire_format::unique_sorted_endpoints(&self.endpoints) {
507 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
508 }
509 }
510
511 h = hash_bytes_u64(h, &self.timestamp.to_le_bytes());
513 h = hash_bytes_u64(h, &self.nonce.to_le_bytes());
514 h = hash_bytes_u64(h, &(self.data_size as u64).to_le_bytes());
515
516 h = hash_bytes_u64(h, self.payload());
518
519 h
520 }
521
522 #[inline]
525 fn _as_le_bytes<T>(&self, expected_kind: MessageDataType) -> TelemetryResult<Vec<T>>
526 where
527 T: LeDecode,
528 {
529 self.ensure_kind(expected_kind)?;
530
531 let bytes: &[u8] = self.payload();
532 let width = T::WIDTH;
533
534 if !bytes.len().is_multiple_of(width) {
535 return Err(TelemetryError::SizeMismatch {
537 expected: (bytes.len() / width) * width,
538 got: bytes.len(),
539 });
540 }
541
542 let count = bytes.len() / width;
543 let mut out = Vec::with_capacity(count);
544
545 for chunk in bytes.chunks_exact(width) {
546 out.push(T::from_le(chunk));
547 }
548
549 Ok(out)
550 }
551
552 pub fn validate(&self) -> TelemetryResult<()> {
564 if self.endpoints.is_empty() {
565 return Err(TelemetryError::EmptyEndpoints);
566 }
567 if self.payload.len() != self.data_size {
568 return Err(TelemetryError::SizeMismatch {
569 expected: self.data_size,
570 got: self.payload.len(),
571 });
572 }
573
574 Self::validate_payload_against_element(self.effective_element(), &self.payload)
575 }
576
577 #[inline]
581 pub fn data_type(&self) -> DataType {
582 self.ty
583 }
584
585 #[inline]
588 pub fn sender(&self) -> &str {
589 &self.sender
590 }
591
592 #[inline]
594 pub fn endpoints(&self) -> &[DataEndpoint] {
595 &self.endpoints
596 }
597
598 #[inline]
600 pub fn timestamp(&self) -> u64 {
601 self.timestamp
602 }
603
604 #[inline]
606 pub fn nonce(&self) -> u16 {
607 self.nonce
608 }
609
610 #[inline]
612 pub fn data_size(&self) -> usize {
613 self.data_size
614 }
615
616 #[inline]
618 pub fn payload(&self) -> &[u8] {
619 &self.payload
620 }
621
622 pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
624 #[cfg(feature = "compact-packet-store")]
625 {
626 self.payload.read_range(offset, out)
627 }
628 #[cfg(not(feature = "compact-packet-store"))]
629 {
630 let end = offset
631 .checked_add(out.len())
632 .filter(|&n| n <= self.data_size)
633 .ok_or(TelemetryError::Unpack("payload range out of bounds"))?;
634 out.copy_from_slice(&self.payload()[offset..end]);
635 Ok(())
636 }
637 }
638
639 #[inline]
644 pub(crate) fn wire_shape(&self) -> Option<MessageElement> {
645 self.wire_shape
646 }
647
648 #[inline]
654 pub(crate) fn wire_target_senders(&self) -> &[u64] {
655 self.wire_target_senders.as_deref().unwrap_or(&[])
656 }
657
658 #[inline]
660 pub fn with_nonce(mut self, nonce: u16) -> Self {
661 self.nonce = nonce;
662 self
663 }
664
665 pub fn header_string(&self) -> String {
672 let mut out = String::with_capacity(DEFAULT_STRING_CAPACITY);
673
674 let _ = write!(
675 &mut out,
676 "Type: {}, Data Size: {}, Sender: {}, Endpoints: [",
677 get_message_name(self.ty),
678 self.data_size,
679 self.sender.as_ref(),
680 );
681 for (i, ep) in self.endpoints.iter().enumerate() {
682 if i != 0 {
683 out.push_str(", ");
684 }
685 #[cfg(feature = "std")]
686 {
687 let endpoint_name = ep.as_str();
688 out.push_str(endpoint_name.as_ref());
689 }
690 #[cfg(not(feature = "std"))]
691 out.push_str(ep.as_str());
692 }
693 out.push_str("], Timestamp: ");
694 let _ = write!(&mut out, "{}", self.timestamp);
695
696 out.push_str(" (");
697 append_human_time(&mut out, self.timestamp);
698 out.push(')');
699 out
700 }
701
702 #[inline]
710 pub fn data_as_utf8_ref(&self) -> Option<&str> {
711 if self.effective_data_type() != MessageDataType::String {
712 return None;
713 }
714 let bytes = &self.payload;
715 let end = bytes.iter().rposition(|&b| b != 0).map(|i| i + 1)?;
716 core::str::from_utf8(&bytes[..end]).ok()
717 }
718
719 #[inline]
725 fn data_to_string<T>(&self, s: &mut String)
726 where
727 T: LeBytes + core::fmt::Display + 'static,
728 {
729 let mut first = true;
730
731 for chunk in self.payload.chunks(T::WIDTH) {
732 if chunk.len() != T::WIDTH {
733 break;
734 }
735 if !first {
736 s.push_str(", ");
737 }
738 first = false;
739
740 let v = T::from_le_slice(chunk);
741
742 if TypeId::of::<T>() == TypeId::of::<f32>() || TypeId::of::<T>() == TypeId::of::<f64>()
744 {
745 let _ = write!(s, "{:.*}", runtime_string_precision(), v);
747 } else {
748 let _ = write!(s, "{v}");
749 }
750 }
751 }
752
753 pub fn as_string(&self) -> String {
761 let mut s = String::from("{");
762 s.push_str(&self.header_string());
763
764 if self.payload.is_empty() {
765 s.push_str(", Data: (<NoData>)}");
766 return s;
767 }
768
769 match self.effective_message_class() {
770 MessageClass::Data => {
771 s.push_str(", Data: (");
772 }
773 MessageClass::Error => {
774 s.push_str(", Error: (");
775 }
776 MessageClass::Warning => {
777 s.push_str(", Warning: (");
778 }
779 }
780
781 if let Some(msg) = self.data_as_utf8_ref() {
782 s.push('"');
783 s.push_str(msg);
784 s.push_str("\")}");
785 return s;
786 }
787
788 match self.effective_data_type() {
789 MessageDataType::Float64 => {
790 self.data_to_string::<f64>(&mut s);
791 }
792 MessageDataType::Float32 => {
793 self.data_to_string::<f32>(&mut s);
794 }
795 MessageDataType::UInt128 => {
796 self.data_to_string::<u128>(&mut s);
797 }
798 MessageDataType::UInt64 => {
799 self.data_to_string::<u64>(&mut s);
800 }
801 MessageDataType::UInt32 => {
802 self.data_to_string::<u32>(&mut s);
803 }
804 MessageDataType::UInt16 => {
805 self.data_to_string::<u16>(&mut s);
806 }
807 MessageDataType::UInt8 => {
808 self.data_to_string::<u8>(&mut s);
809 }
810 MessageDataType::Int128 => {
811 self.data_to_string::<i128>(&mut s);
812 }
813 MessageDataType::Int64 => {
814 self.data_to_string::<i64>(&mut s);
815 }
816 MessageDataType::Int32 => {
817 self.data_to_string::<i32>(&mut s);
818 }
819 MessageDataType::Int16 => {
820 self.data_to_string::<i16>(&mut s);
821 }
822 MessageDataType::Int8 => {
823 self.data_to_string::<i8>(&mut s);
824 }
825 MessageDataType::Bool => {
826 let mut it = self.payload.iter().peekable();
828 while let Some(b) = it.next() {
829 let _ = write!(s, "{}", *b != 0);
830 if it.peek().is_some() {
831 s.push_str(", ");
832 }
833 }
834 }
835 MessageDataType::String => {
836 }
838 MessageDataType::Binary => return self.to_hex_string(),
839 MessageDataType::NoData => {
840 s.push_str("<no data>");
841 }
842 }
843
844 s.push_str(")}");
845 s
846 }
847
848 pub fn to_hex_string(&self) -> String {
856 let mut s = self.header_string();
858 s.push_str(", Data (hex):");
859
860 if !self.payload.is_empty() {
861 s.reserve(self.payload.len().saturating_mul(5));
863 for &b in self.payload.iter() {
864 let _ = write!(&mut s, " 0x{:02x}", b);
865 }
866 }
867 s
868 }
869
870 #[inline]
876 fn ensure_kind(&self, expected: MessageDataType) -> TelemetryResult<()> {
877 let dt = self.effective_data_type();
878 if dt != expected {
879 return Err(TelemetryError::TypeMismatch {
880 expected: data_type_size(expected),
881 got: data_type_size(dt),
882 });
883 }
884 Ok(())
885 }
886
887 impl_data_as_prim! {
888 data_as_f32, f32, MessageDataType::Float32;
889 data_as_f64, f64, MessageDataType::Float64;
890
891 data_as_u8, u8, MessageDataType::UInt8;
892 data_as_u16, u16, MessageDataType::UInt16;
893 data_as_u32, u32, MessageDataType::UInt32;
894 data_as_u64, u64, MessageDataType::UInt64;
895 data_as_u128, u128, MessageDataType::UInt128;
896
897 data_as_i8, i8, MessageDataType::Int8;
898 data_as_i16, i16, MessageDataType::Int16;
899 data_as_i32, i32, MessageDataType::Int32;
900 data_as_i64, i64, MessageDataType::Int64;
901 data_as_i128, i128, MessageDataType::Int128;
902 }
903
904 #[inline]
906 pub fn data_as_bool(&self) -> TelemetryResult<Vec<bool>> {
907 self.ensure_kind(MessageDataType::Bool)?;
908 Ok(self.payload.iter().map(|&b| b != 0).collect())
909 }
910
911 #[inline]
913 pub fn data_as_string(&self) -> TelemetryResult<String> {
914 self.ensure_kind(MessageDataType::String)?;
915
916 let bytes = &self.payload;
917 let end = bytes
918 .iter()
919 .rposition(|&b| b != 0)
920 .map(|i| i + 1)
921 .unwrap_or(0);
922
923 if end == 0 {
924 return Ok(String::new());
925 }
926
927 let s = core::str::from_utf8(&bytes[..end])
928 .map_err(|_| TelemetryError::InvalidUtf8)?
929 .to_string();
930 Ok(s)
931 }
932
933 #[inline]
935 pub fn data_as_binary(&self) -> TelemetryResult<Vec<u8>> {
936 self.ensure_kind(MessageDataType::Binary)?;
937 Ok(self.payload.to_vec())
938 }
939
940 fn from_prim_le_slice_with_sender<T>(
946 ty: DataType,
947 values: &[T],
948 endpoints: &[DataEndpoint],
949 sender: &str,
950 timestamp: u64,
951 ) -> TelemetryResult<Self>
952 where
953 T: Copy + 'static,
954 {
955 let dt = get_data_type(ty);
956
957 if dt == MessageDataType::Bool
960 || dt == MessageDataType::String
961 || dt == MessageDataType::Binary
962 {
963 return Err(TelemetryError::BadArg);
965 }
966 let element_size = data_type_size(dt);
967
968 if element_size != size_of::<T>() {
970 return Err(TelemetryError::TypeMismatch {
971 expected: element_size,
972 got: size_of::<T>(),
973 });
974 }
975
976 let total_bytes = values.len() * element_size;
977
978 if let MessageElement::Static(exact, _, _) = message_meta(ty).element {
980 let exact_bytes = exact * element_size;
981 if total_bytes != exact_bytes {
982 return Err(TelemetryError::SizeMismatch {
983 expected: exact_bytes,
984 got: total_bytes,
985 });
986 }
987 }
988
989 let mut bytes = vec![0u8; total_bytes];
990 unsafe { bytes.set_len(total_bytes) };
991
992 for (i, v) in values.iter().copied().enumerate() {
993 let offset = i * element_size;
994 let dst = &mut bytes[offset..offset + element_size];
995
996 unsafe {
998 core::ptr::copy_nonoverlapping(
999 &v as *const T as *const u8,
1000 dst.as_mut_ptr(),
1001 element_size,
1002 );
1003 }
1004
1005 #[cfg(target_endian = "big")]
1008 {
1009 dst.reverse();
1010 }
1011 }
1012
1013 Self::new(ty, endpoints, sender, timestamp, Arc::<[u8]>::from(bytes))
1014 }
1015
1016 #[inline]
1019 pub fn from_prim_le_slice<T>(
1020 ty: DataType,
1021 values: &[T],
1022 endpoints: &[DataEndpoint],
1023 timestamp: u64,
1024 ) -> TelemetryResult<Self>
1025 where
1026 T: Copy + 'static,
1027 {
1028 let sender = runtime_device_identifier();
1029 Self::from_prim_le_slice_with_sender(ty, values, endpoints, &sender, timestamp)
1030 }
1031
1032 impl_from_prim_slices! {
1036 from_u8_slice, u8;
1037 from_u16_slice, u16;
1038 from_i8_slice, i8;
1039 from_i16_slice, i16;
1040 from_u32_slice, u32;
1041 from_i32_slice, i32;
1042 from_u64_slice, u64;
1043 from_i64_slice, i64;
1044 from_u128_slice, u128;
1045 from_i128_slice, i128;
1046 from_f32_slice, f32;
1047 from_f64_slice, f64;
1048 }
1049
1050 #[inline]
1052 pub fn from_no_data(
1053 ty: DataType,
1054 endpoints: &[DataEndpoint],
1055 timestamp: u64,
1056 ) -> TelemetryResult<Self> {
1057 let meta = message_meta(ty);
1058 match meta.element {
1059 MessageElement::Static(need, _, _) => {
1060 if need != 0 {
1061 return Err(TelemetryError::SizeMismatch {
1062 expected: need,
1063 got: 0,
1064 });
1065 }
1066 }
1067 MessageElement::Dynamic(_, _) => {
1068 }
1070 }
1071
1072 let sender = runtime_device_identifier();
1073 Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from([]))
1074 }
1075
1076 #[inline]
1078 pub fn from_bool_slice(
1079 ty: DataType,
1080 values: &[bool],
1081 endpoints: &[DataEndpoint],
1082 timestamp: u64,
1083 ) -> TelemetryResult<Self> {
1084 if get_data_type(ty) != MessageDataType::Bool {
1085 return Err(TelemetryError::TypeMismatch {
1086 expected: data_type_size(get_data_type(ty)),
1087 got: size_of::<bool>(),
1088 });
1089 }
1090
1091 let total_bytes = values.len();
1092 if let MessageElement::Static(exact, _, _) = message_meta(ty).element
1093 && total_bytes != exact
1094 {
1095 return Err(TelemetryError::SizeMismatch {
1096 expected: exact,
1097 got: total_bytes,
1098 });
1099 }
1100
1101 let mut bytes = Vec::with_capacity(total_bytes);
1102 bytes.extend(values.iter().map(|b| if *b { 1u8 } else { 0u8 }));
1103
1104 let sender = runtime_device_identifier();
1105 Self::new(ty, endpoints, &sender, timestamp, Arc::<[u8]>::from(bytes))
1106 }
1107
1108 #[inline]
1111 pub fn from_str_slice(
1112 ty: DataType,
1113 s: &str,
1114 endpoints: &[DataEndpoint],
1115 timestamp: u64,
1116 ) -> TelemetryResult<Self> {
1117 if get_data_type(ty) != MessageDataType::String {
1118 return Err(TelemetryError::TypeMismatch {
1119 expected: data_type_size(get_data_type(ty)),
1120 got: 1,
1121 });
1122 }
1123
1124 let bytes: Arc<[u8]> = Arc::from(s.as_bytes());
1125 let sender = runtime_device_identifier();
1126 Self::new(ty, endpoints, &sender, timestamp, bytes)
1127 }
1128}
1129
1130#[inline]
1135fn div_mod_u64(n: u64, d: u64) -> (u64, u64) {
1136 (n / d, n % d)
1137}
1138
1139fn civil_from_days(mut z: i64) -> (i32, u32, u32) {
1141 z += 719_468; let era = (if z >= 0 { z } else { z - 146_096 }) / 146_097;
1144 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;
1147 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)
1153}
1154
1155fn append_human_time(out: &mut String, total_ms: u64) {
1158 if total_ms >= EPOCH_MS_THRESHOLD {
1159 let (secs, sub_ms) = div_mod_u64(total_ms, 1_000);
1161 let days = (secs / 86_400) as i64;
1162 let sod = (secs % 86_400) as u32; let (year, month, day) = civil_from_days(days);
1164 let hour = sod / 3_600;
1165 let min = (sod % 3_600) / 60;
1166 let sec = sod % 60;
1167 let _ = Write::write_fmt(
1168 out,
1169 format_args!(
1170 "{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}Z",
1171 year, month, day, hour, min, sec, sub_ms as u32
1172 ),
1173 );
1174 } else {
1175 let hours = total_ms / 3_600_000;
1177 let minutes = (total_ms % 3_600_000) / 60_000;
1178 let seconds = (total_ms % 60_000) / 1_000;
1179 let milliseconds = total_ms % 1_000;
1180 if hours > 0 {
1181 let _ = Write::write_fmt(
1182 out,
1183 format_args!("{hours}h {minutes:02}m {seconds:02}s {milliseconds:03}ms"),
1184 );
1185 } else if minutes > 0 {
1186 let _ = Write::write_fmt(
1187 out,
1188 format_args!("{minutes}m {seconds:02}s {milliseconds:03}ms"),
1189 );
1190 } else {
1191 let _ = Write::write_fmt(out, format_args!("{seconds}s {milliseconds:03}ms"));
1192 }
1193 }
1194}
1195
1196impl core::fmt::Display for Packet {
1201 #[inline]
1202 fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
1203 f.write_str(&Packet::as_string(self))
1204 }
1205}
1206
1207impl ByteCost for Packet {
1208 #[cfg(feature = "compact-packet-store")]
1209 fn park_for_queue(&mut self) -> TelemetryResult<()> {
1210 self.payload.park_for_queue()
1211 }
1212 #[inline]
1213 fn byte_cost(&self) -> usize {
1214 size_of::<Self>()
1215 + self.sender.len()
1216 + self.endpoints.len() * size_of::<DataEndpoint>()
1217 + self.payload.byte_cost()
1218 }
1219}
1220
1221#[cfg(feature = "compact-packet-store")]
1224pub struct StoredPacket {
1225 header: Packet,
1226 payload: crate::packet_store::StoredBytes,
1227}
1228#[cfg(feature = "compact-packet-store")]
1229impl StoredPacket {
1230 pub fn data_type(&self) -> DataType {
1231 self.header.data_type()
1232 }
1233 pub fn sender(&self) -> &str {
1234 self.header.sender()
1235 }
1236 pub fn data_size(&self) -> usize {
1237 self.payload.len()
1238 }
1239 pub fn stored_payload_size(&self) -> usize {
1240 self.payload.stored_len()
1241 }
1242 pub fn payload_is_compressed(&self) -> bool {
1243 self.payload.is_compressed()
1244 }
1245 pub fn pin_payload(&self) -> TelemetryResult<crate::packet_store::PinnedBytes> {
1246 self.payload.pin()
1247 }
1248 pub fn read_payload_range(&self, offset: usize, out: &mut [u8]) -> TelemetryResult<()> {
1249 self.payload.read_range(offset, out)
1250 }
1251 #[cfg(feature = "compact-packet-compression")]
1252 pub fn read_payload_range_bounded(
1253 &self,
1254 codec: &mut crate::packet_store::BoundedPacketCodec,
1255 offset: usize,
1256 out: &mut [u8],
1257 ) -> TelemetryResult<()> {
1258 codec.read_range(&self.payload, offset, out)
1259 }
1260 #[cfg(feature = "compact-packet-compression")]
1261 pub fn read_payload_range_with_codec(
1262 &self,
1263 codec: &mut crate::packet_store::PacketCodec,
1264 offset: usize,
1265 out: &mut [u8],
1266 scratch: &mut [u8],
1267 ) -> TelemetryResult<()> {
1268 codec.read_range(&self.payload, offset, out, scratch)
1269 }
1270}
1271#[cfg(feature = "compact-packet-store")]
1272impl Packet {
1273 pub fn into_stored(
1274 mut self,
1275 store: &Arc<crate::packet_store::PacketStore>,
1276 ) -> TelemetryResult<StoredPacket> {
1277 let payload = store.store(self.payload())?;
1278 self.payload = StandardSmallPayload::new(&[]);
1279 Ok(StoredPacket {
1280 header: self,
1281 payload,
1282 })
1283 }
1284 #[cfg(feature = "compact-packet-compression")]
1285 pub fn into_stored_bounded(
1286 mut self,
1287 store: &Arc<crate::packet_store::PacketStore>,
1288 codec: &mut crate::packet_store::BoundedPacketCodec,
1289 ) -> TelemetryResult<StoredPacket> {
1290 let payload = codec.store(store, self.payload())?;
1291 self.payload = StandardSmallPayload::new(&[]);
1292 Ok(StoredPacket {
1293 header: self,
1294 payload,
1295 })
1296 }
1297 #[cfg(feature = "compact-packet-compression")]
1298 pub fn into_stored_compressed(
1299 mut self,
1300 store: &Arc<crate::packet_store::PacketStore>,
1301 codec: &mut crate::packet_store::PacketCodec,
1302 scratch: &mut [u8],
1303 ) -> TelemetryResult<StoredPacket> {
1304 let payload = codec.store(store, self.payload(), scratch)?;
1305 self.payload = StandardSmallPayload::new(&[]);
1306 Ok(StoredPacket {
1307 header: self,
1308 payload,
1309 })
1310 }
1311}