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
451#[doc(hidden)]
453pub struct CuStatelessTaskState;
454
455impl Freezable for CuStatelessTaskState {}
456
457pub struct BincodeAdapter<'a, T: Freezable + ?Sized>(pub &'a T);
460
461impl<'a, T: Freezable + ?Sized> Encode for BincodeAdapter<'a, T> {
462 fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
463 self.0.freeze(encoder)
464 }
465}
466
467pub trait CuSrcTask: Freezable + Reflect {
472 type Output<'m>: CuMsgPayload;
473 type Resources<'r>;
475
476 fn register_debug_state_types(registry: &mut TypeRegistry)
482 where
483 Self: GetTypeRegistration + Sized,
484 {
485 registry.register::<Self>();
486 }
487
488 fn debug_state_type_path() -> &'static str
490 where
491 Self: TypePath + Sized,
492 {
493 Self::type_path()
494 }
495
496 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
501 where
502 Self: Sized,
503 {
504 f(self)
505 }
506
507 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
510 where
511 Self: Sized;
512
513 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
515 Ok(())
516 }
517
518 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
522 Ok(())
523 }
524
525 fn process<'o>(&mut self, ctx: &CuContext, new_msg: &mut Self::Output<'o>) -> CuResult<()>;
529
530 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
534 Ok(())
535 }
536
537 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
539 Ok(())
540 }
541}
542
543pub trait CuTask: Freezable + Reflect {
545 type Input<'m>: CuMsgPack;
546 type Output<'m>: CuMsgPayload;
547 type Resources<'r>;
549
550 fn register_debug_state_types(registry: &mut TypeRegistry)
556 where
557 Self: GetTypeRegistration + Sized,
558 {
559 registry.register::<Self>();
560 }
561
562 fn debug_state_type_path() -> &'static str
564 where
565 Self: TypePath + Sized,
566 {
567 Self::type_path()
568 }
569
570 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
575 where
576 Self: Sized,
577 {
578 f(self)
579 }
580
581 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
584 where
585 Self: Sized;
586
587 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
589 Ok(())
590 }
591
592 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
596 Ok(())
597 }
598
599 fn process<'i, 'o>(
603 &mut self,
604 _ctx: &CuContext,
605 input: &Self::Input<'i>,
606 output: &mut Self::Output<'o>,
607 ) -> CuResult<()>;
608
609 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
613 Ok(())
614 }
615
616 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
618 Ok(())
619 }
620}
621
622pub trait CuStatelessTask: Reflect + Send + Sync {
635 type Input<'m>: CuMsgPack;
636 type Output<'m>: CuMsgPayload;
637 type Resources<'r>;
639
640 fn register_debug_state_types(registry: &mut TypeRegistry)
642 where
643 Self: GetTypeRegistration + Sized,
644 {
645 registry.register::<Self>();
646 }
647
648 fn debug_state_type_path() -> &'static str
650 where
651 Self: TypePath + Sized,
652 {
653 Self::type_path()
654 }
655
656 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
658 where
659 Self: Sized,
660 {
661 f(self)
662 }
663
664 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
666 where
667 Self: Sized;
668
669 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
671 Ok(())
672 }
673
674 fn preprocess(&self, _ctx: &CuContext) -> CuResult<()> {
676 Ok(())
677 }
678
679 fn process<'i, 'o>(
681 &self,
682 _ctx: &CuContext,
683 input: &Self::Input<'i>,
684 output: &mut Self::Output<'o>,
685 ) -> CuResult<()>;
686
687 fn postprocess(&self, _ctx: &CuContext) -> CuResult<()> {
689 Ok(())
690 }
691
692 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
694 Ok(())
695 }
696}
697
698pub trait CuSinkTask: Freezable + Reflect {
700 type Input<'m>: CuMsgPack;
701 type Resources<'r>;
703
704 fn register_debug_state_types(registry: &mut TypeRegistry)
710 where
711 Self: GetTypeRegistration + Sized,
712 {
713 registry.register::<Self>();
714 }
715
716 fn debug_state_type_path() -> &'static str
718 where
719 Self: TypePath + Sized,
720 {
721 Self::type_path()
722 }
723
724 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
729 where
730 Self: Sized,
731 {
732 f(self)
733 }
734
735 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
738 where
739 Self: Sized;
740
741 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
743 Ok(())
744 }
745
746 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
750 Ok(())
751 }
752
753 fn process<'i>(&mut self, _ctx: &CuContext, input: &Self::Input<'i>) -> CuResult<()>;
757
758 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
762 Ok(())
763 }
764
765 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
767 Ok(())
768 }
769}
770
771#[cfg(test)]
772mod tests {
773 use super::*;
774 use bincode::{config, decode_from_slice, encode_to_vec};
775
776 #[test]
777 fn test_cucompactstr_encode_decode() {
778 let cstr = CuCompactString(CompactString::from("hello"));
779 let config = config::standard();
780 let encoded = encode_to_vec(&cstr, config).expect("Encoding failed");
781 let (decoded, _): (CuCompactString, usize) =
782 decode_from_slice(&encoded, config).expect("Decoding failed");
783 assert_eq!(cstr.0, decoded.0);
784 }
785
786 #[cfg(not(feature = "reflect"))]
794 #[derive(Debug, Clone, bincode::Encode, bincode::Decode, Serialize, Deserialize)]
795 struct TestHandlePayload {
796 handle: crate::pool::CuHandle<Vec<u8>>,
797 }
798
799 #[cfg(not(feature = "reflect"))]
800 impl Default for TestHandlePayload {
801 fn default() -> Self {
802 Self {
803 handle: crate::pool::CuHandle::new_detached(Vec::new()),
804 }
805 }
806 }
807
808 #[cfg(not(feature = "reflect"))]
809 impl TestHandlePayload {
810 fn payload_should_log(&self) -> bool {
813 self.handle.payload_should_log()
814 }
815 }
816
817 #[cfg(not(feature = "reflect"))]
823 #[test]
824 fn test_encode_skips_payload_for_untouched_handle() {
825 use crate::pool::{CuHandle, HandleContent};
826 let cfg = config::standard();
827
828 let untouched = TestHandlePayload {
829 handle: CuHandle::new_detached_with_mode(
830 vec![0xAA, 0xBB, 0xCC, 0xDD],
831 HandleContent::TouchedOnly,
832 ),
833 };
834 let msg_skip: CuMsg<TestHandlePayload> = CuMsg::new(Some(untouched));
835 let skip_bytes = encode_to_vec(&msg_skip, cfg).expect("encode");
836
837 let touched_payload = TestHandlePayload {
838 handle: CuHandle::new_detached_with_mode(
839 vec![0xAA, 0xBB, 0xCC, 0xDD],
840 HandleContent::TouchedOnly,
841 ),
842 };
843 touched_payload.handle.mark_touched();
844 let msg_keep: CuMsg<TestHandlePayload> = CuMsg::new(Some(touched_payload));
845 let keep_bytes = encode_to_vec(&msg_keep, cfg).expect("encode");
846
847 assert_eq!(
848 skip_bytes[0], 0u8,
849 "first byte must be the no-payload presence tag for an untouched TouchedOnly handle"
850 );
851 assert_eq!(
852 keep_bytes[0], 1u8,
853 "first byte must be the payload-present tag once the handle was touched"
854 );
855 assert!(
856 keep_bytes.len() > skip_bytes.len(),
857 "touched encoding ({} bytes) must include payload content; skip is {} bytes",
858 keep_bytes.len(),
859 skip_bytes.len()
860 );
861 }
862
863 #[cfg(not(feature = "reflect"))]
866 #[test]
867 fn test_encode_keeps_payload_for_default_mode() {
868 use crate::pool::{CuHandle, HandleContent};
869 let cfg = config::standard();
870
871 let payload = TestHandlePayload {
872 handle: CuHandle::new_detached_with_mode(vec![1, 2, 3], HandleContent::All),
873 };
874 let msg: CuMsg<TestHandlePayload> = CuMsg::new(Some(payload));
875 let bytes = encode_to_vec(&msg, cfg).expect("encode");
876 assert_eq!(
877 bytes[0], 1u8,
878 "default (HandleContent::All) must keep emitting the payload"
879 );
880 }
881}