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

ad_plugins_rs/
scatter.rs
