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 + Sized
39{
40}
41
42#[cfg(not(feature = "reflect"))]
43pub trait CuMsgPayload:
44 Default + Debug + Clone + Encode + Decode<()> + Serialize + DeserializeOwned + Reflect + Sized
45{
46}
47
48pub trait CuMsgPack {}
49
50#[cfg(feature = "reflect")]
52impl<T> CuMsgPayload for T where
53 T: Default
54 + Debug
55 + Clone
56 + Encode
57 + Decode<()>
58 + Serialize
59 + DeserializeOwned
60 + Reflect
61 + TypePath
62 + Sized
63{
64}
65
66#[cfg(not(feature = "reflect"))]
67impl<T> CuMsgPayload for T where
68 T: Default
69 + Debug
70 + Clone
71 + Encode
72 + Decode<()>
73 + Serialize
74 + DeserializeOwned
75 + Reflect
76 + Sized
77{
78}
79
80macro_rules! impl_cu_msg_pack {
81 ($($name:ident),+) => {
82 impl<'cl, $($name),+> CuMsgPack for ($(&CuMsg<$name>,)+)
83 where
84 $($name: CuMsgPayload),+
85 {}
86 };
87}
88
89macro_rules! impl_cu_msg_pack_up_to {
90 ($first:ident, $second:ident $(, $rest:ident)* $(,)?) => {
91 impl_cu_msg_pack!($first, $second);
92 impl_cu_msg_pack_up_to!(@accumulate ($first, $second); $($rest),*);
93 };
94 (@accumulate ($($acc:ident),+);) => {};
95 (@accumulate ($($acc:ident),+); $next:ident $(, $rest:ident)*) => {
96 impl_cu_msg_pack!($($acc),+, $next);
97 impl_cu_msg_pack_up_to!(@accumulate ($($acc),+, $next); $($rest),*);
98 };
99}
100
101impl<T: CuMsgPayload> CuMsgPack for CuMsg<T> {}
102impl<T: CuMsgPayload> CuMsgPack for &CuMsg<T> {}
103impl<T: CuMsgPayload> CuMsgPack for (&CuMsg<T>,) {}
104impl CuMsgPack for () {}
105
106impl_cu_msg_pack_up_to!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12);
108
109#[macro_export]
112macro_rules! input_msg {
113 ($lt:lifetime, $first:ty, $($rest:ty),+) => {
114 ( & $lt CuMsg<$first>, $( & $lt CuMsg<$rest> ),+ )
115 };
116 ($ty:ty) => {
117 CuMsg<$ty>
118 };
119}
120
121#[macro_export]
123macro_rules! output_msg {
124 ($lt:lifetime, $first:ty, $($rest:ty),+) => {
125 ( CuMsg<$first>, $( CuMsg<$rest> ),+ )
126 };
127 ($first:ty, $($rest:ty),+) => {
128 ( CuMsg<$first>, $( CuMsg<$rest> ),+ )
129 };
130 ($ty:ty) => {
131 CuMsg<$ty>
132 };
133}
134
135pub trait CuSingleOutputMsg {
138 type Payload: CuMsgPayload;
139}
140
141impl<T: CuMsgPayload> CuSingleOutputMsg for CuMsg<T> {
142 type Payload = T;
143}
144
145#[derive(Debug, Clone, bincode::Encode, bincode::Decode, Serialize, Deserialize, Reflect)]
147#[reflect(opaque, from_reflect = false, no_field_bounds)]
148pub struct CuMsgMetadata {
149 pub process_time: PartialCuTimeRange,
151 pub status_txt: CuCompactString,
154 pub origin: Option<CuMsgOrigin>,
156}
157
158impl Metadata for CuMsgMetadata {}
159
160impl CuMsgMetadata {
161 pub fn set_status(&mut self, status: impl ToCompactString) {
162 self.status_txt = CuCompactString(status.to_compact_string());
163 }
164
165 pub fn set_origin(&mut self, origin: CuMsgOrigin) {
166 self.origin = Some(origin);
167 }
168
169 pub fn clear_origin(&mut self) {
170 self.origin = None;
171 }
172}
173
174impl CuMsgMetadataTrait for CuMsgMetadata {
175 fn process_time(&self) -> PartialCuTimeRange {
176 self.process_time
177 }
178
179 fn status_txt(&self) -> &CuCompactString {
180 &self.status_txt
181 }
182
183 fn origin(&self) -> Option<&CuMsgOrigin> {
184 self.origin.as_ref()
185 }
186}
187
188impl Display for CuMsgMetadata {
189 fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
190 write!(
191 f,
192 "process_time start: {}, process_time end: {}",
193 self.process_time.start, self.process_time.end
194 )
195 }
196}
197
198#[derive(Default, Debug, Clone, bincode::Decode, Serialize, Deserialize, Reflect)]
200#[reflect(opaque, from_reflect = false, no_field_bounds)]
201#[serde(bound(
202 serialize = "T: Serialize, M: Serialize",
203 deserialize = "T: DeserializeOwned, M: DeserializeOwned"
204))]
205pub struct CuStampedData<T, M>
206where
207 T: CuMsgPayload,
208 M: Metadata,
209{
210 payload: Option<T>,
212
213 pub tov: Tov,
216
217 pub metadata: M,
219}
220
221impl<T, M> Encode for CuStampedData<T, M>
222where
223 T: CuMsgPayload,
224 M: Metadata,
225{
226 fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
227 match &self.payload {
235 None => {
236 0u8.encode(encoder)?;
237 }
238 Some(payload) => {
239 1u8.encode(encoder)?;
240 let encoded_start = cu29_traits::observed_encode_bytes();
241 let handle_start = crate::monitoring::current_payload_handle_bytes();
242 payload.encode(encoder)?;
243 let encoded_bytes =
244 cu29_traits::observed_encode_bytes().saturating_sub(encoded_start);
245 let handle_bytes =
246 crate::monitoring::current_payload_handle_bytes().saturating_sub(handle_start);
247 crate::monitoring::record_current_slot_payload_io_stats(
248 core::mem::size_of::<T>(),
249 encoded_bytes,
250 handle_bytes,
251 );
252 }
253 }
254 self.tov.encode(encoder)?;
255 self.metadata.encode(encoder)?;
256 Ok(())
257 }
258}
259
260pub fn encode_metadata_only<T, M, E>(
268 msg: &CuStampedData<T, M>,
269 encoder: &mut E,
270) -> Result<(), EncodeError>
271where
272 T: CuMsgPayload,
273 M: Metadata,
274 E: Encoder,
275{
276 0u8.encode(encoder)?;
277 msg.tov.encode(encoder)?;
278 msg.metadata.encode(encoder)?;
279 Ok(())
280}
281
282impl Default for CuMsgMetadata {
283 fn default() -> Self {
284 CuMsgMetadata {
285 process_time: PartialCuTimeRange::default(),
286 status_txt: CuCompactString(CompactString::with_capacity(COMPACT_STRING_CAPACITY)),
287 origin: None,
288 }
289 }
290}
291
292impl<T, M> CuStampedData<T, M>
293where
294 T: CuMsgPayload,
295 M: Metadata,
296{
297 #[doc(hidden)]
303 pub unsafe fn init_in_place(dst: *mut Self) {
304 unsafe {
307 core::ptr::copy_nonoverlapping(
310 const { &None::<T> },
311 core::ptr::addr_of_mut!((*dst).payload),
312 1,
313 );
314 core::ptr::addr_of_mut!((*dst).tov).write(Tov::default());
315 core::ptr::addr_of_mut!((*dst).metadata).write(M::default());
316 }
317 }
318
319 pub(crate) fn from_parts(payload: Option<T>, tov: Tov, metadata: M) -> Self {
320 CuStampedData {
321 payload,
322 tov,
323 metadata,
324 }
325 }
326
327 pub fn new(payload: Option<T>) -> Self {
328 Self::from_parts(payload, Tov::default(), M::default())
329 }
330 pub fn payload(&self) -> Option<&T> {
331 self.payload.as_ref()
332 }
333
334 pub fn set_payload(&mut self, payload: T) {
335 self.payload = Some(payload);
336 }
337
338 pub fn clear_payload(&mut self) {
339 self.payload = None;
340 }
341
342 pub fn payload_mut(&mut self) -> &mut Option<T> {
343 &mut self.payload
344 }
345}
346
347impl<T, M> ErasedCuStampedData for CuStampedData<T, M>
348where
349 T: CuMsgPayload,
350 M: CuMsgMetadataTrait + Metadata,
351{
352 fn payload(&self) -> Option<&dyn erased_serde::Serialize> {
353 self.payload
354 .as_ref()
355 .map(|p| p as &dyn erased_serde::Serialize)
356 }
357
358 #[cfg(feature = "reflect")]
359 fn payload_reflect(&self) -> Option<&dyn cu29_traits::Reflect> {
360 self.payload
361 .as_ref()
362 .map(|p| p as &dyn cu29_traits::Reflect)
363 }
364
365 fn tov(&self) -> Tov {
366 self.tov
367 }
368
369 fn metadata(&self) -> &dyn CuMsgMetadataTrait {
370 &self.metadata
371 }
372}
373
374pub type CuMsg<T> = CuStampedData<T, CuMsgMetadata>;
377
378impl<T: CuMsgPayload> CuStampedData<T, CuMsgMetadata> {
379 pub unsafe fn assume_payload<U: CuMsgPayload>(&self) -> &CuMsg<U> {
386 unsafe { &*(self as *const CuMsg<T> as *const CuMsg<U>) }
388 }
389
390 pub unsafe fn assume_payload_mut<U: CuMsgPayload>(&mut self) -> &mut CuMsg<U> {
397 unsafe { &mut *(self as *mut CuMsg<T> as *mut CuMsg<U>) }
399 }
400}
401
402impl<T: CuMsgPayload + 'static> CuStampedData<T, CuMsgMetadata> {
403 fn downcast_err<U: CuMsgPayload + 'static>() -> CuError {
404 CuError::from(format!(
405 "CuMsg payload mismatch: {} cannot be reinterpreted as {}",
406 type_name::<T>(),
407 type_name::<U>()
408 ))
409 }
410
411 pub fn downcast_ref<U: CuMsgPayload + 'static>(&self) -> CuResult<&CuMsg<U>> {
413 if TypeId::of::<T>() == TypeId::of::<U>() {
414 Ok(unsafe { self.assume_payload::<U>() })
416 } else {
417 Err(Self::downcast_err::<U>())
418 }
419 }
420
421 pub fn downcast_mut<U: CuMsgPayload + 'static>(&mut self) -> CuResult<&mut CuMsg<U>> {
423 if TypeId::of::<T>() == TypeId::of::<U>() {
424 Ok(unsafe { self.assume_payload_mut::<U>() })
426 } else {
427 Err(Self::downcast_err::<U>())
428 }
429 }
430}
431
432pub trait Freezable {
435 fn freeze<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
439 Encode::encode(&(), encoder) }
441
442 fn thaw<D: Decoder>(&mut self, _decoder: &mut D) -> Result<(), DecodeError> {
445 Ok(())
446 }
447}
448
449pub struct BincodeAdapter<'a, T: Freezable + ?Sized>(pub &'a T);
452
453impl<'a, T: Freezable + ?Sized> Encode for BincodeAdapter<'a, T> {
454 fn encode<E: Encoder>(&self, encoder: &mut E) -> Result<(), EncodeError> {
455 self.0.freeze(encoder)
456 }
457}
458
459pub trait CuSrcTask: Freezable + Reflect {
464 type Output<'m>: CuMsgPayload;
465 type Resources<'r>;
467
468 fn register_debug_state_types(registry: &mut TypeRegistry)
474 where
475 Self: GetTypeRegistration + Sized,
476 {
477 registry.register::<Self>();
478 }
479
480 fn debug_state_type_path() -> &'static str
482 where
483 Self: TypePath + Sized,
484 {
485 Self::type_path()
486 }
487
488 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
493 where
494 Self: Sized,
495 {
496 f(self)
497 }
498
499 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
502 where
503 Self: Sized;
504
505 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
507 Ok(())
508 }
509
510 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
514 Ok(())
515 }
516
517 fn process<'o>(&mut self, ctx: &CuContext, new_msg: &mut Self::Output<'o>) -> CuResult<()>;
521
522 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
526 Ok(())
527 }
528
529 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
531 Ok(())
532 }
533}
534
535pub trait CuTask: Freezable + Reflect {
537 type Input<'m>: CuMsgPack;
538 type Output<'m>: CuMsgPayload;
539 type Resources<'r>;
541
542 fn register_debug_state_types(registry: &mut TypeRegistry)
548 where
549 Self: GetTypeRegistration + Sized,
550 {
551 registry.register::<Self>();
552 }
553
554 fn debug_state_type_path() -> &'static str
556 where
557 Self: TypePath + Sized,
558 {
559 Self::type_path()
560 }
561
562 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
567 where
568 Self: Sized,
569 {
570 f(self)
571 }
572
573 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
576 where
577 Self: Sized;
578
579 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
581 Ok(())
582 }
583
584 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
588 Ok(())
589 }
590
591 fn process<'i, 'o>(
595 &mut self,
596 _ctx: &CuContext,
597 input: &Self::Input<'i>,
598 output: &mut Self::Output<'o>,
599 ) -> CuResult<()>;
600
601 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
605 Ok(())
606 }
607
608 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
610 Ok(())
611 }
612}
613
614pub trait CuSinkTask: Freezable + Reflect {
616 type Input<'m>: CuMsgPack;
617 type Resources<'r>;
619
620 fn register_debug_state_types(registry: &mut TypeRegistry)
626 where
627 Self: GetTypeRegistration + Sized,
628 {
629 registry.register::<Self>();
630 }
631
632 fn debug_state_type_path() -> &'static str
634 where
635 Self: TypePath + Sized,
636 {
637 Self::type_path()
638 }
639
640 fn with_debug_state<R>(&self, f: impl FnOnce(&dyn Reflect) -> R) -> R
645 where
646 Self: Sized,
647 {
648 f(self)
649 }
650
651 fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self>
654 where
655 Self: Sized;
656
657 fn start(&mut self, _ctx: &CuContext) -> CuResult<()> {
659 Ok(())
660 }
661
662 fn preprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
666 Ok(())
667 }
668
669 fn process<'i>(&mut self, _ctx: &CuContext, input: &Self::Input<'i>) -> CuResult<()>;
673
674 fn postprocess(&mut self, _ctx: &CuContext) -> CuResult<()> {
678 Ok(())
679 }
680
681 fn stop(&mut self, _ctx: &CuContext) -> CuResult<()> {
683 Ok(())
684 }
685}
686
687#[cfg(test)]
688mod tests {
689 use super::*;
690 use bincode::{config, decode_from_slice, encode_to_vec};
691
692 #[test]
693 fn test_cucompactstr_encode_decode() {
694 let cstr = CuCompactString(CompactString::from("hello"));
695 let config = config::standard();
696 let encoded = encode_to_vec(&cstr, config).expect("Encoding failed");
697 let (decoded, _): (CuCompactString, usize) =
698 decode_from_slice(&encoded, config).expect("Decoding failed");
699 assert_eq!(cstr.0, decoded.0);
700 }
701
702 #[cfg(not(feature = "reflect"))]
710 #[derive(Debug, Clone, bincode::Encode, bincode::Decode, Serialize, Deserialize)]
711 struct TestHandlePayload {
712 handle: crate::pool::CuHandle<Vec<u8>>,
713 }
714
715 #[cfg(not(feature = "reflect"))]
716 impl Default for TestHandlePayload {
717 fn default() -> Self {
718 Self {
719 handle: crate::pool::CuHandle::new_detached(Vec::new()),
720 }
721 }
722 }
723
724 #[cfg(not(feature = "reflect"))]
725 impl TestHandlePayload {
726 fn payload_should_log(&self) -> bool {
729 self.handle.payload_should_log()
730 }
731 }
732
733 #[cfg(not(feature = "reflect"))]
739 #[test]
740 fn test_encode_skips_payload_for_untouched_handle() {
741 use crate::pool::{CuHandle, HandleContent};
742 let cfg = config::standard();
743
744 let untouched = TestHandlePayload {
745 handle: CuHandle::new_detached_with_mode(
746 vec![0xAA, 0xBB, 0xCC, 0xDD],
747 HandleContent::TouchedOnly,
748 ),
749 };
750 let msg_skip: CuMsg<TestHandlePayload> = CuMsg::new(Some(untouched));
751 let skip_bytes = encode_to_vec(&msg_skip, cfg).expect("encode");
752
753 let touched_payload = TestHandlePayload {
754 handle: CuHandle::new_detached_with_mode(
755 vec![0xAA, 0xBB, 0xCC, 0xDD],
756 HandleContent::TouchedOnly,
757 ),
758 };
759 touched_payload.handle.mark_touched();
760 let msg_keep: CuMsg<TestHandlePayload> = CuMsg::new(Some(touched_payload));
761 let keep_bytes = encode_to_vec(&msg_keep, cfg).expect("encode");
762
763 assert_eq!(
764 skip_bytes[0], 0u8,
765 "first byte must be the no-payload presence tag for an untouched TouchedOnly handle"
766 );
767 assert_eq!(
768 keep_bytes[0], 1u8,
769 "first byte must be the payload-present tag once the handle was touched"
770 );
771 assert!(
772 keep_bytes.len() > skip_bytes.len(),
773 "touched encoding ({} bytes) must include payload content; skip is {} bytes",
774 keep_bytes.len(),
775 skip_bytes.len()
776 );
777 }
778
779 #[cfg(not(feature = "reflect"))]
782 #[test]
783 fn test_encode_keeps_payload_for_default_mode() {
784 use crate::pool::{CuHandle, HandleContent};
785 let cfg = config::standard();
786
787 let payload = TestHandlePayload {
788 handle: CuHandle::new_detached_with_mode(vec![1, 2, 3], HandleContent::All),
789 };
790 let msg: CuMsg<TestHandlePayload> = CuMsg::new(Some(payload));
791 let bytes = encode_to_vec(&msg, cfg).expect("encode");
792 assert_eq!(
793 bytes[0], 1u8,
794 "default (HandleContent::All) must keep emitting the payload"
795 );
796 }
797}