ad_plugins_rs/
scatter.rs

1use std::sync::Arc;
2
3use ad_core_rs::ndarray::NDArray;
4use ad_core_rs::ndarray_pool::NDArrayPool;
5use ad_core_rs::plugin::runtime::{NDPluginProcess, ProcessResult};
6
7/// Scatter method.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum ScatterMethod {
10    RoundRobin = 0,
11}
12
13impl ScatterMethod {
14    pub fn from_i32(_v: i32) -> Self {
15        Self::RoundRobin
16    }
17}
18
19/// Scatter processor: marks each frame to be distributed to a single
20/// downstream consumer in round-robin order.
21///
22/// The target consumer — and the reroute-past-a-full-queue / drop-only-on-the-
23/// last-node decisions — are owned by the plugin runtime delivery path, which
24/// holds the persistent cursor (C++ `NDPluginScatter::nextClient_`) and is the
25/// only layer that can see per-consumer queue state. The processor therefore
26/// only flags the frame as a scatter frame, matching C++ `NDPluginScatter`,
27/// which keeps no per-frame index of its own (the cursor lives entirely in
28/// `nextClient_`).
29pub struct ScatterProcessor {
30    method: ScatterMethod,
31    method_idx: Option<usize>,
32}
33
34impl ScatterProcessor {
35    pub fn new() -> Self {
36        Self {
37            method: ScatterMethod::RoundRobin,
38            method_idx: None,
39        }
40    }
41}
42
43impl Default for ScatterProcessor {
44    fn default() -> Self {
45        Self::new()
46    }
47}
48
49impl NDPluginProcess for ScatterProcessor {
50    fn process_array(&mut self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
51        ProcessResult::scatter(vec![Arc::new(array.clone())])
52    }
53
54    fn plugin_type(&self) -> &str {
55        "NDPluginScatter"
56    }
57
58    fn register_params(
59        &mut self,
60        base: &mut asyn_rs::port::PortDriverBase,
61    ) -> asyn_rs::error::AsynResult<()> {
62        use asyn_rs::param::ParamType;
63        base.create_param("SCATTER_METHOD", ParamType::Int32)?;
64        self.method_idx = base.find_param("SCATTER_METHOD");
65        Ok(())
66    }
67
68    fn on_param_change(
69        &mut self,
70        reason: usize,
71        params: &ad_core_rs::plugin::runtime::PluginParamSnapshot,
72    ) -> ad_core_rs::plugin::runtime::ParamChangeResult {
73        if Some(reason) == self.method_idx {
74            self.method = ScatterMethod::from_i32(params.value.as_i32());
75        }
76        ad_core_rs::plugin::runtime::ParamChangeResult::updates(vec![])
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use super::*;
83    use ad_core_rs::ndarray::{NDDataType, NDDimension};
84
85    #[test]
86    fn test_scatter_processor_marks_scatter_frame() {
87        // The processor flags every frame as a scatter frame and passes the
88        // array through unchanged; the runtime owns the round-robin cursor and
89        // the reroute/drop decisions (see ad_core_rs runtime scatter_publish
90        // tests for the routing behavior itself).
91        let mut proc = ScatterProcessor::new();
92        let pool = NDArrayPool::new(1_000_000);
93
94        let mut arr = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
95        arr.unique_id = 42;
96
97        for _ in 0..4 {
98            let r = proc.process_array(&arr, &pool);
99            assert!(
100                r.scatter,
101                "scatter processor must mark the frame as scatter"
102            );
103            assert_eq!(r.output_arrays.len(), 1);
104            assert_eq!(r.output_arrays[0].unique_id, 42);
105        }
106    }
107}
ad_plugins_rs/scatter.rs

ad_plugins_rs/
scatter.rs
