async_nodes/
value.rs

1use std::{
2    any::{Any, type_name},
3    fmt,
4    sync::Arc,
5};
6
7use indexmap::IndexMap;
8
9use super::{
10    error::BspError,
11    graph::{NodeId, OutputPin, PortIndex},
12    node::NodeOutcome,
13};
14
15/// A marker type used to propagate bypassed execution paths in dynamic control-flow topologies.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub struct BspSkip;
18
19/// A thread-safe, reference-counted container wrapping arbitrary Rust types.
20#[derive(Clone)]
21pub struct BspValue {
22    inner: Arc<dyn Any + Send + Sync>,
23    type_name: &'static str,
24}
25
26impl fmt::Debug for BspValue {
27    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
28        f.debug_struct("BspValue")
29            .field("type_name", &self.type_name)
30            .finish_non_exhaustive()
31    }
32}
33
34impl BspValue {
35    pub fn new<T: Any + Send + Sync>(val: T) -> Self {
36        Self {
37            inner: Arc::new(val) as Arc<dyn Any + Send + Sync>,
38            type_name: type_name::<T>(),
39        }
40    }
41
42    #[must_use]
43    pub const fn type_name(&self) -> &'static str {
44        self.type_name
45    }
46
47    #[must_use]
48    pub fn is<T: Any>(&self) -> bool {
49        self.inner.as_ref().is::<T>()
50    }
51
52    pub fn downcast<T: Any + Send + Sync>(self) -> Result<Arc<T>, Self> {
53        if self.is::<T>() {
54            let type_name = self.type_name;
55            match self.inner.downcast::<T>() {
56                Ok(arc_t) => Ok(arc_t),
57                Err(arc_any) => Err(Self {
58                    inner: arc_any,
59                    type_name,
60                }),
61            }
62        } else {
63            Err(self)
64        }
65    }
66
67    #[must_use]
68    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
69        self.inner.as_ref().downcast_ref::<T>()
70    }
71
72    pub fn downcast_mut<T: Any>(&mut self) -> Option<&mut T> {
73        Arc::get_mut(&mut self.inner)?.downcast_mut::<T>()
74    }
75}
76
77// =========================================================================
78// TRAITS FOR MACRO INTEGRATION & STATIC DISPATCH
79// =========================================================================
80
81/// Translates a raw `BspValue` back into a typed parameter.
82pub trait BspFromValue: Sized {
83    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError>;
84}
85
86/// Serializes a typed output back into a type-erased `BspValue` container.
87pub trait BspToValue {
88    fn to_value(self) -> BspValue;
89}
90
91/// Standardized boundary compiler that converts functional return types into `NodeOutcome`.
92pub trait BspOutputPacker {
93    fn output_ports() -> Vec<PortIndex>;
94    fn pack(self, node_id: NodeId, node_name: &str) -> Result<NodeOutcome, BspError>;
95
96    /// Packs typed outputs into the raw port map — the form
97    /// [`crate::node::DataflowNode`] implementors return.
98    ///
99    /// The default routes through [`BspOutputPacker::pack`]; a `Suspend` outcome has no
100    /// port map and is rejected as a validation error (dataflow nodes never suspend —
101    /// suspend-capable nodes implement [`crate::node::BspNode`] directly).
102    fn pack_outputs(
103        self,
104        node_id: NodeId,
105        node_name: &str,
106    ) -> Result<IndexMap<PortIndex, BspValue>, BspError>
107    where
108        Self: Sized,
109    {
110        match self.pack(node_id, node_name)? {
111            NodeOutcome::Success(outputs) => Ok(outputs),
112            NodeOutcome::Suspend { .. } => Err(BspError::ValidationError(format!(
113                "node '{node_name}' suspended: suspend outcomes have no port map; \
114                 implement BspNode directly for suspend-capable nodes"
115            ))),
116        }
117    }
118
119    /// Returns whether this output compiles dynamic outputs that bypass static validation checks.
120    #[must_use]
121    fn is_dynamic() -> bool {
122        false
123    }
124}
125
126// Support wrapping directly into Arc references
127impl<T: Any + Send + Sync> BspFromValue for Arc<T> {
128    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError> {
129        let found_type = value.type_name();
130        value.downcast::<T>().map_err(|_| BspError::TypeMismatch {
131            node_id,
132            port_index: port,
133            expected: type_name::<T>(),
134            found: found_type,
135        })
136    }
137}
138
139impl<T: Any + Send + Sync> BspToValue for Arc<T> {
140    fn to_value(self) -> BspValue {
141        BspValue {
142            inner: self as Arc<dyn Any + Send + Sync>,
143            type_name: type_name::<T>(),
144        }
145    }
146}
147
148impl<T: BspFromValue> BspFromValue for Option<T> {
149    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError> {
150        if value.is::<BspSkip>() {
151            Ok(None)
152        } else {
153            T::from_value(value, node_id, port).map(Some)
154        }
155    }
156}
157
158impl<T: BspToValue + 'static> BspToValue for Option<T> {
159    fn to_value(self) -> BspValue {
160        self.map_or_else(|| BspValue::new(BspSkip), BspToValue::to_value)
161    }
162}
163
164impl<T: BspToValue + 'static> BspOutputPacker for Option<T> {
165    fn output_ports() -> Vec<PortIndex> {
166        vec![0]
167    }
168    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
169        let mut outputs = IndexMap::new();
170        outputs.insert(0, self.to_value());
171        Ok(NodeOutcome::Success(outputs))
172    }
173}
174
175// Core suspension pass-through implementations
176impl BspOutputPacker for NodeOutcome {
177    fn output_ports() -> Vec<PortIndex> {
178        vec![0]
179    }
180    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
181        Ok(self)
182    }
183    fn is_dynamic() -> bool {
184        true
185    }
186}
187
188// Support returning Arc directly
189impl<T: Any + Send + Sync> BspOutputPacker for Arc<T> {
190    fn output_ports() -> Vec<PortIndex> {
191        vec![0]
192    }
193    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
194        let mut outputs = IndexMap::new();
195        outputs.insert(0, self.to_value());
196        Ok(NodeOutcome::Success(outputs))
197    }
198}
199
200// Void return support
201impl BspOutputPacker for () {
202    fn output_ports() -> Vec<PortIndex> {
203        Vec::new()
204    }
205    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
206        Ok(NodeOutcome::Success(IndexMap::new()))
207    }
208}
209
210// Single generic output compiler routing standard results beautifully
211impl<T: BspOutputPacker, E: std::fmt::Display> BspOutputPacker for Result<T, E> {
212    fn output_ports() -> Vec<PortIndex> {
213        T::output_ports()
214    }
215    fn pack(self, node_id: NodeId, node_name: &str) -> Result<NodeOutcome, BspError> {
216        match self {
217            Ok(val) => val.pack(node_id, node_name),
218            Err(e) => Err(BspError::NodeExecutionFailed {
219                node_id,
220                node_name: node_name.to_string(),
221                message: e.to_string(),
222            }),
223        }
224    }
225    fn is_dynamic() -> bool {
226        T::is_dynamic()
227    }
228}
229
230// =========================================================================
231// MULTIPLE-PORT TUPLE OUTPUT PACKING
232// =========================================================================
233
234impl<T1, T2> BspOutputPacker for (T1, T2)
235where
236    T1: BspToValue + 'static,
237    T2: BspToValue + 'static,
238{
239    fn output_ports() -> Vec<PortIndex> {
240        vec![0, 1]
241    }
242    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
243        let mut outputs = IndexMap::new();
244        outputs.insert(0, self.0.to_value());
245        outputs.insert(1, self.1.to_value());
246        Ok(NodeOutcome::Success(outputs))
247    }
248}
249
250impl<T1, T2, T3> BspOutputPacker for (T1, T2, T3)
251where
252    T1: BspToValue + 'static,
253    T2: BspToValue + 'static,
254    T3: BspToValue + 'static,
255{
256    fn output_ports() -> Vec<PortIndex> {
257        vec![0, 1, 2]
258    }
259    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
260        let mut outputs = IndexMap::new();
261        outputs.insert(0, self.0.to_value());
262        outputs.insert(1, self.1.to_value());
263        outputs.insert(2, self.2.to_value());
264        Ok(NodeOutcome::Success(outputs))
265    }
266}
267
268// =========================================================================
269// BLANKET AGGREGATION & VECTOR PACKING
270// =========================================================================
271
272impl<T: BspFromValue + Clone + Send + Sync + 'static> BspFromValue for Vec<T> {
273    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError> {
274        if value.is::<BspSkip>() {
275            return Ok(Self::new());
276        }
277
278        // Resolves recursive vector wrapping bugs by validating direct type downcasts
279        if value.is::<Self>() {
280            let arc_vec = value
281                .downcast::<Self>()
282                .map_err(|err| BspError::TypeMismatch {
283                    node_id,
284                    port_index: port,
285                    expected: std::any::type_name::<Self>(),
286                    found: err.type_name(), // Uses the returned BspValue ownership to extract the type name
287                })?;
288            return Ok((*arc_vec).clone());
289        }
290
291        if let Some(vec_bsp) = value.downcast_ref::<Vec<BspValue>>() {
292            let mut result = Self::with_capacity(vec_bsp.len());
293            for val in vec_bsp {
294                if !val.is::<BspSkip>() {
295                    result.push(T::from_value(val.clone(), node_id, port)?);
296                }
297            }
298            Ok(result)
299        } else {
300            Ok(vec![T::from_value(value, node_id, port)?])
301        }
302    }
303}
304
305impl<T: BspToValue + Send + Sync + 'static> BspToValue for Vec<T> {
306    fn to_value(self) -> BspValue {
307        BspValue::new(self)
308    }
309}
310
311impl<T: BspToValue + Send + Sync + 'static> BspOutputPacker for Vec<T> {
312    fn output_ports() -> Vec<PortIndex> {
313        vec![0]
314    }
315    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
316        let mut outputs = IndexMap::new();
317        outputs.insert(0, self.to_value());
318        Ok(NodeOutcome::Success(outputs))
319    }
320}
321
322// =========================================================================
323// ERGONOMIC MAP DOWNCASTING EXTENSIONS
324// =========================================================================
325
326pub trait BspOutputMapExt {
327    fn get_typed<T: std::any::Any + Send + Sync + Clone>(
328        &self,
329        pin: OutputPin,
330    ) -> Result<T, BspError>;
331}
332
333impl BspOutputMapExt for IndexMap<OutputPin, BspValue> {
334    fn get_typed<T: std::any::Any + Send + Sync + Clone>(
335        &self,
336        pin: OutputPin,
337    ) -> Result<T, BspError> {
338        let value = self.get(&pin).ok_or(BspError::NodeNotFound {
339            node_id: pin.node_id,
340        })?;
341        let found_type = value.type_name();
342        let arc_t = value
343            .clone()
344            .downcast::<T>()
345            .map_err(|_| BspError::TypeMismatch {
346                node_id: pin.node_id,
347                port_index: pin.port,
348                expected: std::any::type_name::<T>(),
349                found: found_type,
350            })?;
351        Ok((*arc_t).clone())
352    }
353}
354
355// =========================================================================
356// PRIMITIVE TYPE REGISTRATION
357// =========================================================================
358
359#[macro_export]
360macro_rules! register_bsp_type {
361    ($t:ty) => {
362        impl $crate::value::BspFromValue for $t {
363            fn from_value(
364                value: $crate::value::BspValue,
365                node_id: $crate::graph::NodeId,
366                port: $crate::graph::PortIndex,
367            ) -> Result<Self, $crate::error::BspError> {
368                let found_type = value.type_name();
369                let arc_t =
370                    value
371                        .downcast::<$t>()
372                        .map_err(|_| $crate::error::BspError::TypeMismatch {
373                            node_id,
374                            port_index: port,
375                            expected: std::any::type_name::<$t>(),
376                            found: found_type,
377                        })?;
378                Ok((*arc_t).clone())
379            }
380        }
381
382        impl $crate::value::BspToValue for $t {
383            fn to_value(self) -> $crate::value::BspValue {
384                $crate::value::BspValue::new(self)
385            }
386        }
387
388        impl $crate::value::BspOutputPacker for $t {
389            fn output_ports() -> Vec<$crate::graph::PortIndex> {
390                vec![0]
391            }
392            fn pack(
393                self,
394                _node_id: $crate::graph::NodeId,
395                _node_name: &str,
396            ) -> Result<$crate::node::NodeOutcome, $crate::error::BspError> {
397                let mut outputs = ::indexmap::IndexMap::new();
398                outputs.insert(0, $crate::value::BspToValue::to_value(self));
399                Ok($crate::node::NodeOutcome::Success(outputs))
400            }
401        }
402    };
403}
404
405// Standard primitives registration
406register_bsp_type!(i64);
407register_bsp_type!(f64);
408register_bsp_type!(bool);
409register_bsp_type!(String);
410register_bsp_type!(serde_json::Value);
411register_bsp_type!(BspSkip);
async_nodes/value.rs

async_nodes/
value.rs
