1use crate::config::ComponentConfig;
5use crate::context::CuContext;
6use crate::reflect::{GetTypeRegistration, Reflect, TypePath, TypeRegistry};
7#[cfg(feature = "reflect")]
8use bevy_reflect;
9use bincode::de::{Decode, Decoder};
10use bincode::enc::{Encode, Encoder};
11use bincode::error::{DecodeError, EncodeError};
12use compact_str::{CompactString, ToCompactString};
13use core::any::{TypeId, type_name};
14use cu29_clock::{PartialCuTimeRange, Tov};
15use cu29_traits::{
16 COMPACT_STRING_CAPACITY, CuCompactString, CuError, CuMsgMetadataTrait, CuMsgOrigin, CuResult,
17 ErasedCuStampedData, Metadata,
18};
19use serde::de::DeserializeOwned;
20use serde::{Deserialize, Serialize};
21
22use alloc::format;
23use core::fmt::{Debug, Display, Formatter, Result as FmtResult};
24
25#[cfg(feature = "reflect")]
28pub trait CuMsgPayload:
29 Default
30 + Debug
31 + Clone
32 + Encode
33 + Decode<()>
34 + Serialize
35 + DeserializeOwned
36 + Reflect
37 + TypePath
38 + GetTypeRegistration
39 + Sized
40{
41}
42
43#[cfg(not(feature = "reflect"))]
44pub trait CuMsgPayload:
45 Default + Debug + Clone + Encode + Decode<()> + Serialize + DeserializeOwned + Reflect + Sized
46{
47}
48
49pub trait CuMsgPack {}
50
51#[cfg(feature = "reflect")]
53impl<T> CuMsgPayload for T where
54 T: Default
55 + Debug
56 + Clone
57 + Encode
58 + Decode<()>
59 + Serialize
60 + DeserializeOwned
61 + Reflect
62 + TypePath
63 + GetTypeRegistration
64 + Sized
65{
66}
67
68#[cfg(not(feature = "reflect"))]
69impl<T> CuMsgPayload for T where
70 T: Default
71 + Debug
72 + Clone
73 + Encode
74 + Decode<()>
75 + Serialize
76 + DeserializeOwned
77 + Reflect
78 + Sized
79{
80}
81
82macro_rules! impl_cu_msg_pack {
83 ($($name:ident),+) => {
84 impl<'cl, $($name),+> CuMsgPack for ($(&CuMsg<$name>,)+)
85 where
86 $($name: CuMsgPayload),+
87 {}
88 };
89}
90
91macro_rules! impl_cu_msg_pack_up_to {
92 ($first:ident, $second:ident $(, $rest:ident)* $(,)?) => {
93 impl_cu_msg_pack!($first, $second);
94 impl_cu_msg_pack_up_to!(@accumulate ($first, $second); $($rest),*);
95 };
96 (@accumulate ($($acc:ident),+);) => {};
97 (@accumulate ($($acc:ident),+); $next:ident $(, $rest:ident)*) => {
98 impl_cu_msg_pack!($($acc),+, $next);
99 impl_cu_msg_pack_up_to!(@accumulate ($($acc),+, $next); $($rest),*);
100 };
101}
102
103impl<T: CuMsgPayload> CuMsgPack for CuMsg<T> {}
104impl<T: CuMsgPayload> CuMsgPack for &CuMsg<T> {}
105impl<T: CuMsgPayload> CuMsgPack for (&CuMsg<T>,) {}
106impl CuMsgPack for () {}
107
108impl_cu_msg_pack_up_to!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12);
110
111#[macro_export]
114macro_rules! input_msg {
115 ($lt:lifetime, $first:ty, $($rest:ty),+) => {
116 ( & $lt CuMsg<$first>, $( & $lt CuMsg<$rest> ),+ )
117 };
118 ($ty:ty) => {
119 CuMsg<$ty>
120 };
121}
122
123#[macro_export]
125macro_rules! output_msg {
126 ($lt:lifetime, $first:ty, $($rest:ty),+) => {
127 ( CuMsg<$first>, $( CuMsg<$rest> ),+ )
128 };
129 ($first:ty, $($rest:ty),+) => {
130 ( CuMsg<$first>, $( CuMsg<$rest> ),+ )
131 };
132 ($ty:ty) => {
133 CuMsg<$ty>
134 };
135}
136
137pub trait CuSingleOutputMsg {
140 type Payload: CuMsgPayload;
141}
142
143impl<T: CuMsgPayload> CuSingleOutputMsg for CuMsg<T> {
144 type Payload = T;
145}
146
147#[derive(Debug, Clone, bincode::Encode, bincode::Decode, Serialize, Deserialize, Reflect)]
149#[reflect(opaque, from_reflect = false, no_field_bounds)]
150pub struct CuMsgMetadata {
151 pub process_time: PartialCuTimeRange,
153 pub status_txt: CuCompactString,
156 pub origin: Option<CuMsgOrigin>,
158}
159
160impl Metadata for CuMsgMetadata {}
161
162impl CuMsgMetadata {
163 pub fn set_status(&mut self, status: impl ToCompactString) {
164 self.status_txt = CuCompactString(status.to_compact_string());
165 }
166
167 pub fn set_origin(&mut self, origin: CuMsgOrigin) {
168 self.origin = Some(origin);
169 }
170
171 pub fn clear_origin(&mut self) {
172 self.origin = None;
173 }
174}
175
176impl CuMsgMetadataTrait for CuMsgMetadata {
177 fn process_time(&self) -> PartialCuTimeRange {
178 self.process_time
179 }
180
181 fn status_txt(&self) -> &CuCompactString {
182 &self.status_txt
183 }
184
185 fn origin(&self) -> Option<&CuMsgOrigin> {
186 self.origin.as_ref()
187 }
188}
189
190impl Display for CuMsgMetadata {
191 fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
192 write!(
193 f,
194 "process_time start: {}, process_time end: {}",
195 self.process_time.start, self.process_time.end
196 )
197 }
198}
199
200#[derive(Default, Debug, Clone, bincode::Decode, Serialize, Deserialize, Reflect)]
202#[reflect(opaque, from_reflect = false, no_field_bounds)]
203#[serde(bound(
204 serialize = "T: Serialize, M: Serialize",
205 deserialize = "T: DeserializeOwned, M: DeserializeOwned"
206))]
207pub struct CuStampedData<T, M>
208where
209 T: CuMsgPayload,
210 M: Metadata,
211{
212 payload: Option<T>,
214
215 pub tov: Tov,
218
219 pub metadata: M,
221}
222
223impl<T, M> Encode for CuStampedData<T, M>
224where
225 T: CuMsgPayload,
226 M: Metadata,
227{
228 fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
229 match &self.payload {
237 None => {
238 0u8.encode(encoder)?;
239 }
240 Some(payload) => {
241 1u8.encode(encoder)?;
242 let encoded_start = cu29_traits::observed_encode_bytes();
243 let handle_start = crate::monitoring::current_payload_handle_bytes();
244 payload.encode(encoder)?;
245 let encoded_bytes =
246 cu29_traits::observed_encode_bytes().saturating_sub(encoded_start);
247 let handle_bytes =
248 crate::monitoring::current_payload_handle_bytes().saturating_sub(handle_start);
249 crate::monitoring::record_current_slot_payload_io_stats(
250 core::mem::size_of::<T>(),
251 encoded_bytes,
252 handle_bytes,
253 );
254 }
255 }
256 self.tov.encode(encoder)?;
257 self.metadata.encode(encoder)?;
258 Ok(())
259 }
260}
261
262pub fn encode_metadata_only<T, M, E>(
270 msg: &CuStampedData<T, M>,
271 encoder: &mut E,
272) -> Result<(), EncodeError>
273where
274 T: CuMsgPayload,
275 M: Metadata,
276 E: Encoder,
277{
278 0u8.encode(encoder)?;
279 msg.tov.encode(encoder)?;
280 msg.metadata.encode(encoder)?;
281 Ok(())
282}
283
284impl Default for CuMsgMetadata {
285 fn default() -> Self {
286 CuMsgMetadata {
287 process_time: PartialCuTimeRange::default(),
288 status_txt: CuCompactString(CompactString::with_capacity(COMPACT_STRING_CAPACITY)),
289 origin: None,
290 }
291 }
292}
293
294impl<T, M> CuStampedData<T, M>
295where
296 T: CuMsgPayload,
297 M: Metadata,
298{
299 #[doc(hidden)]
305 pub unsafe fn init_in_place(dst: *mut Self) {
306 unsafe {
309 core::ptr::copy_nonoverlapping(
312 const { &None::<T> },
313 core::ptr::addr_of_mut!((*dst).payload),
314 1,
315 );
316 core::ptr::addr_of_mut!((*dst).tov).write(Tov::default());
317 core::ptr::addr_of_mut!((*dst).metadata).write(M::default());
318 }
319 }
320
321 pub(crate) fn from_parts(payload: Option<T>, tov: Tov, metadata: M) -> Self {
322 CuStampedData {
323 payload,
324 tov,
325 metadata,
326 }
327 }
328
329 pub fn new(payload: Option<T>) -> Self {
330 Self::from_parts(payload, Tov::default(), M::default())
331 }
332 pub fn payload(&self) -> Option<&T> {
333 self.payload.as_ref()
334 }
335
336 pub fn set_payload(&mut self, payload: T) {
337 self.payload = Some(payload);
338 }
339
340 pub fn clear_payload(&mut self) {
341 self.payload = None;
342 }
343
344 pub fn payload_mut(&mut self) -> &mut Option<T> {
345 &mut self.payload
346 }
347}
348
349impl<T, M> ErasedCuStampedData for CuStampedData<T, M>
350where
351 T: CuMsgPayload,
352 M: CuMsgMetadataTrait + Metadata,
353{
354 fn payload(&self) -> Option<&dyn erased_serde::Serialize> {
355 self.payload
356 .as_ref()
357 .map(|p| p as &dyn erased_serde::Serialize)
358 }
359
360 #[cfg(feature = "reflect")]
361 fn payload_reflect(&self) -> Option<&dyn cu29_traits::Reflect> {
362 self.payload
363 .as_ref()
364 .map(|p| p as &dyn cu29_traits::Reflect)
365 }
366
367 fn tov(&self) -> Tov {
368 self.tov
369 }
370
371 fn metadata(&self) -> &dyn CuMsgMetadataTrait {
372 &self.metadata
373 }
374}
375
376pub type CuMsg<T> = CuStampedData<T, CuMsgMetadata>;
379
380impl<T: CuMsgPayload> CuStampedData<T, CuMsgMetadata> {
381 pub unsafe fn assume_payload<U: CuMsgPayload>(&self) -> &CuMsg<U> {
388 unsafe { &*(self as *const CuMsg<T> as *const CuMsg<U>) }
390 }
391
392 pub unsafe fn assume_payload_mut<U: CuMsgPayload>(&mut self) -> &mut CuMsg<U> {
399 unsafe { &mut *(self as *mut CuMsg<T> as *mut CuMsg<U>) }
401 }
402}
403
404impl<T: CuMsgPayload + 'static> CuStampedData<T, CuMsgMetadata> {
405 fn downcast_err<U: CuMsgPayload + 'static>() -> CuError {
406 CuError::from(format!(
407 "CuMsg payload mismatch: {} cannot be reinterpreted as {}",
408 type_name::<T>(),
409 type_name::<U>()
410 ))
411 }
412
413 pub fn downcast_ref<U: CuMsgPayload + 'static>(&self) -> CuResult<&CuMsg<U>> {
415 if TypeId::of::<T>() == TypeId::of::<U>() {
416 Ok(unsafe { self.assume_payload::<U>() })
418 } else {
419 Err(Self::downcast_err::<U>())
420 }
421 }
422
423 pub fn downcast_mut<U: CuMsgPayload + 'static>(&mut self) -> CuResult<&mut CuMsg<U>> {
425 if TypeId::of::<T>() == TypeId::of::<U>() {
426 Ok(unsafe { self.assume_payload_mut::<U>() })
428 } else {
429 Err(Self::downcast_err::<U>())
430 }
431 }
432}
433
434pub trait Freezable {
437 fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
441 Encode::encode(&(), encoder) }
443
444 fn thaw<D: Decoder>(&mut self, _decoder: &mut D) -> Result<(), DecodeError> {
447 Ok(())
448 }
449}
450
451pub struct BincodeAdapter<'a, T: Freezable + ?Sized>(pub &'a T);
454
455impl<'a, T: Freezable + ?Sized> Encode for BincodeAdapter<'a, T> {
456 fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
457 self.0.freeze(encoder)
458 }
459}
460
461pub trait CuSrcTask: Freezable + Reflect {
466 type Output<'m>: CuMsgPayload;
467 type Resources<'r>;
469
470 fn register_debug_state_types(registry: &mut TypeRegistry)
476 where
477 Self: GetTypeRegistration + Sized,
478 {
479 registry.register::<Self>();
480 }
481
482 fn debug_state_type_path() -> &'static str
484 where
485 Self: TypePath + Sized,
486 {
487 Self::type_path()
488 }
489
490 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
495 where
496 Self: Sized,
497 {
498 f(self)
499 }
500
501 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
504 where
505 Self: Sized;
506
507 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
509 Ok(())
510 }
511
512 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
516 Ok(())
517 }
518
519 fn process<'o>(&mut self, ctx: &CuContext, new_msg: &mut Self::Output<'o>) -> CuResult<()>;
523
524 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
528 Ok(())
529 }
530
531 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
533 Ok(())
534 }
535}
536
537pub trait CuTask: Freezable + Reflect {
539 type Input<'m>: CuMsgPack;
540 type Output<'m>: CuMsgPayload;
541 type Resources<'r>;
543
544 fn register_debug_state_types(registry: &mut TypeRegistry)
550 where
551 Self: GetTypeRegistration + Sized,
552 {
553 registry.register::<Self>();
554 }
555
556 fn debug_state_type_path() -> &'static str
558 where
559 Self: TypePath + Sized,
560 {
561 Self::type_path()
562 }
563
564 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
569 where
570 Self: Sized,
571 {
572 f(self)
573 }
574
575 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
578 where
579 Self: Sized;
580
581 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
583 Ok(())
584 }
585
586 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
590 Ok(())
591 }
592
593 fn process<'i, 'o>(
597 &mut self,
598 _ctx: &CuContext,
599 input: &Self::Input<'i>,
600 output: &mut Self::Output<'o>,
601 ) -> CuResult<()>;
602
603 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
607 Ok(())
608 }
609
610 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
612 Ok(())
613 }
614}
615
616pub trait CuSinkTask: Freezable + Reflect {
618 type Input<'m>: CuMsgPack;
619 type Resources<'r>;
621
622 fn register_debug_state_types(registry: &mut TypeRegistry)
628 where
629 Self: GetTypeRegistration + Sized,
630 {
631 registry.register::<Self>();
632 }
633
634 fn debug_state_type_path() -> &'static str
636 where
637 Self: TypePath + Sized,
638 {
639 Self::type_path()
640 }
641
642 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
647 where
648 Self: Sized,
649 {
650 f(self)
651 }
652
653 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
656 where
657 Self: Sized;
658
659 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
661 Ok(())
662 }
663
664 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
668 Ok(())
669 }
670
671 fn process<'i>(&mut self, _ctx: &CuContext, input: &Self::Input<'i>) -> CuResult<()>;
675
676 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
680 Ok(())
681 }
682
683 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
685 Ok(())
686 }
687}
688
689#[cfg(test)]
690mod tests {
691 use super::*;
692 use bincode::{config, decode_from_slice, encode_to_vec};
693
694 #[test]
695 fn test_cucompactstr_encode_decode() {
696 let cstr = CuCompactString(CompactString::from("hello"));
697 let config = config::standard();
698 let encoded = encode_to_vec(&cstr, config).expect("Encoding failed");
699 let (decoded, _): (CuCompactString, usize) =
700 decode_from_slice(&encoded, config).expect("Decoding failed");
701 assert_eq!(cstr.0, decoded.0);
702 }
703
704 #[cfg(not(feature = "reflect"))]
712 #[derive(Debug, Clone, bincode::Encode, bincode::Decode, Serialize, Deserialize)]
713 struct TestHandlePayload {
714 handle: crate::pool::CuHandle<Vec<u8>>,
715 }
716
717 #[cfg(not(feature = "reflect"))]
718 impl Default for TestHandlePayload {
719 fn default() -> Self {
720 Self {
721 handle: crate::pool::CuHandle::new_detached(Vec::new()),
722 }
723 }
724 }
725
726 #[cfg(not(feature = "reflect"))]
727 impl TestHandlePayload {
728 fn payload_should_log(&self) -> bool {
731 self.handle.payload_should_log()
732 }
733 }
734
735 #[cfg(not(feature = "reflect"))]
741 #[test]
742 fn test_encode_skips_payload_for_untouched_handle() {
743 use crate::pool::{CuHandle, HandleContent};
744 let cfg = config::standard();
745
746 let untouched = TestHandlePayload {
747 handle: CuHandle::new_detached_with_mode(
748 vec![0xAA, 0xBB, 0xCC, 0xDD],
749 HandleContent::TouchedOnly,
750 ),
751 };
752 let msg_skip: CuMsg<TestHandlePayload> = CuMsg::new(Some(untouched));
753 let skip_bytes = encode_to_vec(&msg_skip, cfg).expect("encode");
754
755 let touched_payload = TestHandlePayload {
756 handle: CuHandle::new_detached_with_mode(
757 vec![0xAA, 0xBB, 0xCC, 0xDD],
758 HandleContent::TouchedOnly,
759 ),
760 };
761 touched_payload.handle.mark_touched();
762 let msg_keep: CuMsg<TestHandlePayload> = CuMsg::new(Some(touched_payload));
763 let keep_bytes = encode_to_vec(&msg_keep, cfg).expect("encode");
764
765 assert_eq!(
766 skip_bytes[0], 0u8,
767 "first byte must be the no-payload presence tag for an untouched TouchedOnly handle"
768 );
769 assert_eq!(
770 keep_bytes[0], 1u8,
771 "first byte must be the payload-present tag once the handle was touched"
772 );
773 assert!(
774 keep_bytes.len() > skip_bytes.len(),
775 "touched encoding ({} bytes) must include payload content; skip is {} bytes",
776 keep_bytes.len(),
777 skip_bytes.len()
778 );
779 }
780
781 #[cfg(not(feature = "reflect"))]
784 #[test]
785 fn test_encode_keeps_payload_for_default_mode() {
786 use crate::pool::{CuHandle, HandleContent};
787 let cfg = config::standard();
788
789 let payload = TestHandlePayload {
790 handle: CuHandle::new_detached_with_mode(vec![1, 2, 3], HandleContent::All),
791 };
792 let msg: CuMsg<TestHandlePayload> = CuMsg::new(Some(payload));
793 let bytes = encode_to_vec(&msg, cfg).expect("encode");
794 assert_eq!(
795 bytes[0], 1u8,
796 "default (HandleContent::All) must keep emitting the payload"
797 );
798 }
799}