spvirit_server/
pva_server.rs

1//! High-level PVAccess server — builder pattern for typed records.
2//!
3//! # Example
4//!
5//! ```rust,ignore
6//! use spvirit_server::PvaServer;
7//!
8//! let server = PvaServer::builder()
9//!     .ai("SIM:TEMPERATURE", 22.5)
10//!     .ao("SIM:SETPOINT", 25.0)
11//!     .bo("SIM:ENABLE", false)
12//!     .build();
13//!
14//! server.run().await?;
15//! ```
16
17use std::collections::HashMap;
18use std::net::IpAddr;
19use std::sync::Arc;
20use std::time::Duration;
21
22use regex::Regex;
23use tracing::info;
24
25use spvirit_types::{NtEnum, NtScalar, NtScalarArray, NtTable as NtTableType, NtTableColumn, NtNdArray as NtNdArrayType, NdCodec, NdDimension, NtTimeStamp, PvValue, ScalarArrayValue, ScalarValue};
26
27use crate::db::{load_db, parse_db};
28use crate::handler::PvListMode;
29use crate::monitor::MonitorRegistry;
30use crate::pvstore::{Source, SourceRegistry};
31use crate::server::{PvaServerConfig, run_pva_server_with_registry};
32use crate::simple_store::{LinkDef, OnPutCallback, ScanCallback, SimplePvStore};
33use crate::types::{DbCommonState, OutputMode, RecordData, RecordInstance, RecordType};
34
35// ─── PvaServerBuilder ────────────────────────────────────────────────────
36
37/// Builder for [`PvaServer`].
38///
39/// ```rust,ignore
40/// let server = PvaServer::builder()
41///     .ai("TEMP:READBACK", 22.5)
42///     .ao("TEMP:SETPOINT", 25.0)
43///     .bo("HEATER:ON", false)
44///     .port(5075)
45///     .build();
46/// ```
47pub struct PvaServerBuilder {
48    records: HashMap<String, RecordInstance>,
49    on_put: HashMap<String, OnPutCallback>,
50    scans: Vec<(String, Duration, ScanCallback)>,
51    links: Vec<LinkDef>,
52    extra_sources: Vec<(String, i32, Arc<dyn Source>)>,
53    tcp_port: u16,
54    udp_port: u16,
55    listen_ip: Option<IpAddr>,
56    advertise_ip: Option<IpAddr>,
57    compute_alarms: bool,
58    beacon_period_secs: u64,
59    conn_timeout: Duration,
60    pvlist_mode: PvListMode,
61    pvlist_max: usize,
62    pvlist_allow_pattern: Option<Regex>,
63}
64
65impl PvaServerBuilder {
66    fn new() -> Self {
67        Self {
68            records: HashMap::new(),
69            on_put: HashMap::new(),
70            scans: Vec::new(),
71            links: Vec::new(),
72            extra_sources: Vec::new(),
73            tcp_port: 5075,
74            udp_port: 5076,
75            listen_ip: None,
76            advertise_ip: None,
77            compute_alarms: false,
78            beacon_period_secs: 15,
79            conn_timeout: Duration::from_secs(64000),
80            pvlist_mode: PvListMode::List,
81            pvlist_max: 1024,
82            pvlist_allow_pattern: None,
83        }
84    }
85
86    // ─── Typed record constructors ───────────────────────────────────
87
88    /// Add an `ai` (analog input, read-only) record.
89    pub fn ai(mut self, name: impl Into<String>, initial: f64) -> Self {
90        let name = name.into();
91        self.records.insert(
92            name.clone(),
93            make_scalar_record(&name, RecordType::Ai, ScalarValue::F64(initial)),
94        );
95        self
96    }
97
98    /// Add an `ao` (analog output, writable) record.
99    pub fn ao(mut self, name: impl Into<String>, initial: f64) -> Self {
100        let name = name.into();
101        self.records.insert(
102            name.clone(),
103            make_output_record(&name, RecordType::Ao, ScalarValue::F64(initial)),
104        );
105        self
106    }
107
108    /// Add a `bi` (binary input, read-only) record.
109    pub fn bi(mut self, name: impl Into<String>, initial: bool) -> Self {
110        let name = name.into();
111        self.records.insert(
112            name.clone(),
113            make_scalar_record(&name, RecordType::Bi, ScalarValue::Bool(initial)),
114        );
115        self
116    }
117
118    /// Add a `bo` (binary output, writable) record.
119    pub fn bo(mut self, name: impl Into<String>, initial: bool) -> Self {
120        let name = name.into();
121        self.records.insert(
122            name.clone(),
123            make_output_record(&name, RecordType::Bo, ScalarValue::Bool(initial)),
124        );
125        self
126    }
127
128    /// Add a `stringin` (string input, read-only) record.
129    pub fn string_in(mut self, name: impl Into<String>, initial: impl Into<String>) -> Self {
130        let name = name.into();
131        self.records.insert(
132            name.clone(),
133            make_scalar_record(
134                &name,
135                RecordType::StringIn,
136                ScalarValue::Str(initial.into()),
137            ),
138        );
139        self
140    }
141
142    /// Add a `stringout` (string output, writable) record.
143    pub fn string_out(mut self, name: impl Into<String>, initial: impl Into<String>) -> Self {
144        let name = name.into();
145        self.records.insert(
146            name.clone(),
147            make_output_record(
148                &name,
149                RecordType::StringOut,
150                ScalarValue::Str(initial.into()),
151            ),
152        );
153        self
154    }
155
156    /// Add a `waveform` record (array) with the given initial data.
157    pub fn waveform(mut self, name: impl Into<String>, data: ScalarArrayValue) -> Self {
158        let name = name.into();
159        let ftvl = data.type_label().trim_end_matches("[]").to_string();
160        let nelm = data.len();
161        self.records.insert(
162            name.clone(),
163            RecordInstance {
164                name: name.clone(),
165                record_type: RecordType::Waveform,
166                common: DbCommonState::default(),
167                data: RecordData::Waveform {
168                    nt: NtScalarArray::from_value(data),
169                    inp: None,
170                    ftvl,
171                    nelm,
172                    nord: nelm,
173                },
174                raw_fields: HashMap::new(),
175            },
176        );
177        self
178    }
179
180    /// Add an `aai` (analog array input, read-only) record.
181    pub fn aai(mut self, name: impl Into<String>, data: ScalarArrayValue) -> Self {
182        let name = name.into();
183        let ftvl = data.type_label().trim_end_matches("[]").to_string();
184        let nelm = data.len();
185        self.records.insert(
186            name.clone(),
187            RecordInstance {
188                name: name.clone(),
189                record_type: RecordType::Aai,
190                common: DbCommonState::default(),
191                data: RecordData::Aai {
192                    nt: NtScalarArray::from_value(data),
193                    inp: None,
194                    ftvl,
195                    nelm,
196                    nord: nelm,
197                },
198                raw_fields: HashMap::new(),
199            },
200        );
201        self
202    }
203
204    /// Add an `aao` (analog array output, writable) record.
205    pub fn aao(mut self, name: impl Into<String>, data: ScalarArrayValue) -> Self {
206        let name = name.into();
207        let ftvl = data.type_label().trim_end_matches("[]").to_string();
208        let nelm = data.len();
209        self.records.insert(
210            name.clone(),
211            RecordInstance {
212                name: name.clone(),
213                record_type: RecordType::Aao,
214                common: DbCommonState::default(),
215                data: RecordData::Aao {
216                    nt: NtScalarArray::from_value(data),
217                    out: None,
218                    dol: None,
219                    omsl: OutputMode::Supervisory,
220                    ftvl,
221                    nelm,
222                    nord: nelm,
223                },
224                raw_fields: HashMap::new(),
225            },
226        );
227        self
228    }
229
230    /// Add a `subarray` record — a view into part of an array.
231    pub fn sub_array(
232        mut self,
233        name: impl Into<String>,
234        data: ScalarArrayValue,
235        indx: usize,
236        nelm: usize,
237    ) -> Self {
238        let name = name.into();
239        let ftvl = data.type_label().trim_end_matches("[]").to_string();
240        let malm = data.len();
241        let nord = nelm.min(malm.saturating_sub(indx));
242        self.records.insert(
243            name.clone(),
244            RecordInstance {
245                name: name.clone(),
246                record_type: RecordType::SubArray,
247                common: DbCommonState::default(),
248                data: RecordData::SubArray {
249                    nt: NtScalarArray::from_value(data),
250                    inp: None,
251                    ftvl,
252                    malm,
253                    nelm,
254                    nord,
255                    indx,
256                },
257                raw_fields: HashMap::new(),
258            },
259        );
260        self
261    }
262
263    /// Add an NTTable record.
264    pub fn nt_table(
265        mut self,
266        name: impl Into<String>,
267        columns: Vec<(String, ScalarArrayValue)>,
268    ) -> Self {
269        let name = name.into();
270        let labels: Vec<String> = columns.iter().map(|(n, _)| n.clone()).collect();
271        let cols: Vec<NtTableColumn> = columns
272            .into_iter()
273            .map(|(n, v)| NtTableColumn { name: n, values: v })
274            .collect();
275        self.records.insert(
276            name.clone(),
277            RecordInstance {
278                name: name.clone(),
279                record_type: RecordType::NtTable,
280                common: DbCommonState::default(),
281                data: RecordData::NtTable {
282                    nt: NtTableType {
283                        labels,
284                        columns: cols,
285                        descriptor: None,
286                        alarm: None,
287                        time_stamp: None,
288                    },
289                    inp: None,
290                    out: None,
291                    omsl: OutputMode::Supervisory,
292                },
293                raw_fields: HashMap::new(),
294            },
295        );
296        self
297    }
298
299    /// Add an NTNDArray record.
300    pub fn nt_ndarray(
301        mut self,
302        name: impl Into<String>,
303        data: ScalarArrayValue,
304        dims: Vec<(i32, i32)>,
305    ) -> Self {
306        let name = name.into();
307        let dimension: Vec<NdDimension> = dims
308            .into_iter()
309            .map(|(size, offset)| NdDimension {
310                size,
311                offset,
312                full_size: size,
313                binning: 1,
314                reverse: false,
315            })
316            .collect();
317        let uncompressed_size =
318            (data.len() * data.element_size_bytes().max(1)) as i64;
319        self.records.insert(
320            name.clone(),
321            RecordInstance {
322                name: name.clone(),
323                record_type: RecordType::NtNdArray,
324                common: DbCommonState::default(),
325                data: RecordData::NtNdArray {
326                    nt: NtNdArrayType {
327                        value: data,
328                        codec: NdCodec {
329                            name: String::new(),
330                            parameters: Default::default(),
331                        },
332                        compressed_size: uncompressed_size,
333                        uncompressed_size,
334                        dimension,
335                        unique_id: 0,
336                        data_time_stamp: NtTimeStamp {
337                            seconds_past_epoch: 0,
338                            nanoseconds: 0,
339                            user_tag: 0,
340                        },
341                        attribute: vec![],
342                        descriptor: None,
343                        alarm: None,
344                        time_stamp: None,
345                        display: None,
346                    },
347                    inp: None,
348                    out: None,
349                    omsl: OutputMode::Supervisory,
350                },
351                raw_fields: HashMap::new(),
352            },
353        );
354        self
355    }
356
357    /// Add an `mbbi` (multi-bit binary input, read-only) NTEnum record.
358    pub fn mbbi(
359        mut self,
360        name: impl Into<String>,
361        choices: Vec<String>,
362        initial: i32,
363    ) -> Self {
364        let name = name.into();
365        self.records.insert(
366            name.clone(),
367            RecordInstance {
368                name: name.clone(),
369                record_type: RecordType::Mbbi,
370                common: DbCommonState::default(),
371                data: RecordData::NtEnum {
372                    nt: NtEnum::new(initial, choices),
373                    inp: None,
374                    out: None,
375                    omsl: OutputMode::Supervisory,
376                },
377                raw_fields: HashMap::new(),
378            },
379        );
380        self
381    }
382
383    /// Add an `mbbo` (multi-bit binary output, writable) NTEnum record.
384    pub fn mbbo(
385        mut self,
386        name: impl Into<String>,
387        choices: Vec<String>,
388        initial: i32,
389    ) -> Self {
390        let name = name.into();
391        self.records.insert(
392            name.clone(),
393            RecordInstance {
394                name: name.clone(),
395                record_type: RecordType::Mbbo,
396                common: DbCommonState::default(),
397                data: RecordData::NtEnum {
398                    nt: NtEnum::new(initial, choices),
399                    inp: None,
400                    out: None,
401                    omsl: OutputMode::Supervisory,
402                },
403                raw_fields: HashMap::new(),
404            },
405        );
406        self
407    }
408
409    /// Add a generic structure record with a custom struct ID and fields.
410    pub fn generic(
411        mut self,
412        name: impl Into<String>,
413        struct_id: impl Into<String>,
414        fields: Vec<(String, PvValue)>,
415    ) -> Self {
416        let name = name.into();
417        self.records.insert(
418            name.clone(),
419            RecordInstance {
420                name: name.clone(),
421                record_type: RecordType::Generic,
422                common: DbCommonState::default(),
423                data: RecordData::Generic {
424                    struct_id: struct_id.into(),
425                    fields,
426                    inp: None,
427                    out: None,
428                    omsl: OutputMode::Supervisory,
429                },
430                raw_fields: HashMap::new(),
431            },
432        );
433        self
434    }
435
436    // ─── .db file loading ────────────────────────────────────────────
437
438    /// Load records from an EPICS `.db` file.
439    pub fn db_file(mut self, path: impl AsRef<str>) -> Self {
440        match load_db(path.as_ref()) {
441            Ok(records) => {
442                self.records.extend(records);
443            }
444            Err(e) => {
445                tracing::error!("Failed to load db file '{}': {}", path.as_ref(), e);
446            }
447        }
448        self
449    }
450
451    /// Parse records from an EPICS `.db` string.
452    pub fn db_string(mut self, content: &str) -> Self {
453        match parse_db(content) {
454            Ok(records) => {
455                self.records.extend(records);
456            }
457            Err(e) => {
458                tracing::error!("Failed to parse db string: {}", e);
459            }
460        }
461        self
462    }
463
464    // ─── Callbacks ───────────────────────────────────────────────────
465
466    /// Register a callback invoked when a PUT is applied to the named PV.
467    pub fn on_put<F>(mut self, name: impl Into<String>, callback: F) -> Self
468    where
469        F: Fn(&str, &spvirit_codec::spvd_decode::DecodedValue) + Send + Sync + 'static,
470    {
471        self.on_put.insert(name.into(), Arc::new(callback));
472        self
473    }
474
475    /// Register a periodic scan callback that produces a new value for a PV.
476    pub fn scan<F>(mut self, name: impl Into<String>, period: Duration, callback: F) -> Self
477    where
478        F: Fn(&str) -> ScalarValue + Send + Sync + 'static,
479    {
480        self.scans.push((name.into(), period, Arc::new(callback)));
481        self
482    }
483
484    /// Link an output PV to one or more input PVs.
485    ///
486    /// Whenever any input PV changes (via `set_value`, protocol PUT, or
487    /// another link), the `compute` callback is invoked with the current
488    /// values of **all** inputs (in order) and the result is written to
489    /// the output PV.
490    ///
491    /// ```rust,ignore
492    /// .link("CALC:SUM", &["INPUT:A", "INPUT:B"], |values| {
493    ///     let a = values[0].as_f64().unwrap_or(0.0);
494    ///     let b = values[1].as_f64().unwrap_or(0.0);
495    ///     ScalarValue::F64(a + b)
496    /// })
497    /// ```
498    pub fn link<F>(mut self, output: impl Into<String>, inputs: &[&str], compute: F) -> Self
499    where
500        F: Fn(&[ScalarValue]) -> ScalarValue + Send + Sync + 'static,
501    {
502        self.links.push(LinkDef {
503            output: output.into(),
504            inputs: inputs.iter().map(|s| s.to_string()).collect(),
505            compute: Arc::new(compute),
506        });
507        self
508    }
509
510    // ─── External sources ────────────────────────────────────────────
511
512    /// Register an additional [`Source`] at the given priority.
513    ///
514    /// Lower `order` values are checked first during PV name resolution.
515    /// The built-in `SimplePvStore` (records added via `.ai()`, `.ao()`, etc.)
516    /// is always registered at order 0.
517    ///
518    /// ```rust,ignore
519    /// .source("hardware", -10, Arc::new(HardwareSource::new()))
520    /// ```
521    pub fn source(
522        mut self,
523        label: impl Into<String>,
524        order: i32,
525        source: Arc<dyn Source>,
526    ) -> Self {
527        self.extra_sources.push((label.into(), order, source));
528        self
529    }
530
531    // ─── Configuration ───────────────────────────────────────────────
532
533    /// Set the TCP port (default 5075).
534    pub fn port(mut self, port: u16) -> Self {
535        self.tcp_port = port;
536        self
537    }
538
539    /// Set the UDP search port (default 5076).
540    pub fn udp_port(mut self, port: u16) -> Self {
541        self.udp_port = port;
542        self
543    }
544
545    /// Set the IP address to listen on.
546    pub fn listen_ip(mut self, ip: IpAddr) -> Self {
547        self.listen_ip = Some(ip);
548        self
549    }
550
551    /// Set the IP address to advertise in search responses.
552    pub fn advertise_ip(mut self, ip: IpAddr) -> Self {
553        self.advertise_ip = Some(ip);
554        self
555    }
556
557    /// Enable alarm computation from limits.
558    pub fn compute_alarms(mut self, enabled: bool) -> Self {
559        self.compute_alarms = enabled;
560        self
561    }
562
563    /// Set the beacon broadcast period in seconds (default 15).
564    pub fn beacon_period(mut self, secs: u64) -> Self {
565        self.beacon_period_secs = secs;
566        self
567    }
568
569    /// Set the idle connection timeout (default ~18 hours).
570    pub fn conn_timeout(mut self, timeout: Duration) -> Self {
571        self.conn_timeout = timeout;
572        self
573    }
574
575    /// Set the PV list mode (default [`PvListMode::List`]).
576    pub fn pvlist_mode(mut self, mode: PvListMode) -> Self {
577        self.pvlist_mode = mode;
578        self
579    }
580
581    /// Set the maximum number of PV names in pvlist responses (default 1024).
582    pub fn pvlist_max(mut self, max: usize) -> Self {
583        self.pvlist_max = max;
584        self
585    }
586
587    /// Set a regex filter for PV names exposed by pvlist.
588    pub fn pvlist_allow_pattern(mut self, pattern: Regex) -> Self {
589        self.pvlist_allow_pattern = Some(pattern);
590        self
591    }
592
593    /// Build the [`PvaServer`].
594    pub fn build(self) -> PvaServer {
595        let store = Arc::new(SimplePvStore::new(
596            self.records,
597            self.on_put,
598            self.links,
599            self.compute_alarms,
600        ));
601
602        let mut config = PvaServerConfig::default();
603        config.tcp_port = self.tcp_port;
604        config.udp_port = self.udp_port;
605        config.compute_alarms = self.compute_alarms;
606        if let Some(ip) = self.listen_ip {
607            config.listen_ip = ip;
608        }
609        config.advertise_ip = self.advertise_ip;
610        config.beacon_period_secs = self.beacon_period_secs;
611        config.conn_timeout = self.conn_timeout;
612        config.pvlist_mode = self.pvlist_mode;
613        config.pvlist_max = self.pvlist_max;
614        config.pvlist_allow_pattern = self.pvlist_allow_pattern;
615
616        PvaServer {
617            store,
618            extra_sources: self.extra_sources,
619            config,
620            scans: self.scans,
621            monitor_registry: None,
622        }
623    }
624}
625
626// ─── PvaServer ───────────────────────────────────────────────────────────
627
628/// High-level PVAccess server.
629///
630/// Built via [`PvaServer::builder()`] with typed record constructors,
631/// `.db_file()` loading, `.on_put()` / `.scan()` callbacks, and a
632/// simple `.run()` to start serving.
633///
634/// ```rust,ignore
635/// let server = PvaServer::builder()
636///     .ai("SIM:TEMP", 22.5)
637///     .ao("SIM:SP", 25.0)
638///     .build();
639///
640/// // Read/write PVs from another task:
641/// let store = server.store();
642/// store.set_value("SIM:TEMP", ScalarValue::F64(23.1)).await;
643///
644/// server.run().await?;
645/// ```
646pub struct PvaServer {
647    store: Arc<SimplePvStore>,
648    extra_sources: Vec<(String, i32, Arc<dyn Source>)>,
649    config: PvaServerConfig,
650    scans: Vec<(String, Duration, ScanCallback)>,
651    /// Optional pre-supplied monitor registry so external code (e.g. Python
652    /// bindings) can notify monitors from outside `run()`.
653    monitor_registry: Option<Arc<MonitorRegistry>>,
654}
655
656impl PvaServer {
657    /// Create a builder for configuring a [`PvaServer`].
658    pub fn builder() -> PvaServerBuilder {
659        PvaServerBuilder::new()
660    }
661
662    /// Get a reference to the underlying store for runtime get/put.
663    pub fn store(&self) -> &Arc<SimplePvStore> {
664        &self.store
665    }
666
667    /// Register an additional [`Source`] after building the server.
668    ///
669    /// This is useful when the source needs a reference to the store
670    /// (which is only available after `.build()`).
671    ///
672    /// ```rust,ignore
673    /// let server = PvaServer::builder().ai("X", 0.0).build();
674    /// let store = server.store().clone();
675    /// server.add_source("agg", 10, Arc::new(MyAggSource::new(store)));
676    /// server.run().await?;
677    /// ```
678    pub fn add_source(
679        &mut self,
680        label: impl Into<String>,
681        order: i32,
682        source: Arc<dyn Source>,
683    ) {
684        self.extra_sources.push((label.into(), order, source));
685    }
686
687    /// Pre-supply the [`MonitorRegistry`] that [`Self::run`] will use.
688    ///
689    /// This lets external code (for example Python `Source` adapters)
690    /// hold onto the registry and publish monitor updates to subscribed
691    /// PVAccess clients from outside `run()`.
692    pub fn set_monitor_registry(&mut self, registry: Arc<MonitorRegistry>) {
693        self.monitor_registry = Some(registry);
694    }
695
696    /// Get a shared handle to the [`MonitorRegistry`] that will be used
697    /// when [`Self::run`] starts.  Creates (and stores) a new registry
698    /// on first call so external code can register before run.
699    pub fn monitor_registry(&mut self) -> Arc<MonitorRegistry> {
700        if self.monitor_registry.is_none() {
701            self.monitor_registry = Some(Arc::new(MonitorRegistry::new()));
702        }
703        self.monitor_registry.as_ref().unwrap().clone()
704    }
705
706    /// Start the PVA server (UDP search + TCP handler + beacon + scan tasks).
707    ///
708    /// This blocks until the server is shut down or an error occurs.
709    pub async fn run(self) -> Result<(), Box<dyn std::error::Error>> {
710        // Create the monitor registry early so scan tasks can notify
711        // PVAccess monitor clients when values change.
712        let registry = self
713            .monitor_registry
714            .clone()
715            .unwrap_or_else(|| Arc::new(MonitorRegistry::new()));
716        self.store.set_registry(registry.clone()).await;
717
718        // Build the source registry with the built-in store at order 0.
719        let sources = Arc::new(SourceRegistry::new());
720        sources.add("builtin", 0, self.store.clone()).await;
721
722        // Register any extra sources provided via .source().
723        for (label, order, source) in &self.extra_sources {
724            sources.add(label.clone(), *order, source.clone()).await;
725        }
726
727        // Spawn scan tasks.
728        for (name, period, callback) in &self.scans {
729            let store = self.store.clone();
730            let name = name.clone();
731            let period = *period;
732            let callback = callback.clone();
733            tokio::spawn(async move {
734                let mut interval = tokio::time::interval(period);
735                loop {
736                    interval.tick().await;
737                    let new_val = callback(&name);
738                    store.set_value(&name, new_val).await;
739                }
740            });
741        }
742
743        let pv_count = self.store.pv_names().await.len();
744        info!(
745            "PvaServer starting: {} PVs on port {}",
746            pv_count, self.config.tcp_port
747        );
748
749        run_pva_server_with_registry(sources, self.config, registry).await
750    }
751}
752
753// ─── Record construction helpers ─────────────────────────────────────────
754
755fn make_scalar_record(name: &str, record_type: RecordType, value: ScalarValue) -> RecordInstance {
756    let nt = NtScalar::from_value(value);
757    let data = match record_type {
758        RecordType::Ai => RecordData::Ai {
759            nt,
760            inp: None,
761            siml: None,
762            siol: None,
763            simm: false,
764        },
765        RecordType::Bi => RecordData::Bi {
766            nt,
767            inp: None,
768            znam: "Off".to_string(),
769            onam: "On".to_string(),
770            siml: None,
771            siol: None,
772            simm: false,
773        },
774        RecordType::StringIn => RecordData::StringIn {
775            nt,
776            inp: None,
777            siml: None,
778            siol: None,
779            simm: false,
780        },
781        _ => panic!("make_scalar_record: unsupported type {record_type:?}"),
782    };
783    RecordInstance {
784        name: name.to_string(),
785        record_type,
786        common: DbCommonState::default(),
787        data,
788        raw_fields: HashMap::new(),
789    }
790}
791
792fn make_output_record(name: &str, record_type: RecordType, value: ScalarValue) -> RecordInstance {
793    let nt = NtScalar::from_value(value);
794    let data = match record_type {
795        RecordType::Ao => RecordData::Ao {
796            nt,
797            out: None,
798            dol: None,
799            omsl: OutputMode::Supervisory,
800            drvl: None,
801            drvh: None,
802            oroc: None,
803            siml: None,
804            siol: None,
805            simm: false,
806        },
807        RecordType::Bo => RecordData::Bo {
808            nt,
809            out: None,
810            dol: None,
811            omsl: OutputMode::Supervisory,
812            znam: "Off".to_string(),
813            onam: "On".to_string(),
814            siml: None,
815            siol: None,
816            simm: false,
817        },
818        RecordType::StringOut => RecordData::StringOut {
819            nt,
820            out: None,
821            dol: None,
822            omsl: OutputMode::Supervisory,
823            siml: None,
824            siol: None,
825            simm: false,
826        },
827        _ => panic!("make_output_record: unsupported type {record_type:?}"),
828    };
829    RecordInstance {
830        name: name.to_string(),
831        record_type,
832        common: DbCommonState::default(),
833        data,
834        raw_fields: HashMap::new(),
835    }
836}
837
838#[cfg(test)]
839mod tests {
840    use super::*;
841
842    #[test]
843    fn builder_creates_records() {
844        let server = PvaServer::builder()
845            .ai("T:AI", 1.0)
846            .ao("T:AO", 2.0)
847            .bi("T:BI", true)
848            .bo("T:BO", false)
849            .string_in("T:SI", "hello")
850            .string_out("T:SO", "world")
851            .build();
852
853        let rt = tokio::runtime::Builder::new_current_thread()
854            .enable_all()
855            .build()
856            .unwrap();
857        let names = rt.block_on(server.store.pv_names());
858        assert_eq!(names.len(), 6);
859    }
860
861    #[test]
862    fn builder_defaults() {
863        let server = PvaServer::builder().build();
864        assert_eq!(server.config.tcp_port, 5075);
865        assert_eq!(server.config.udp_port, 5076);
866        assert!(!server.config.compute_alarms);
867    }
868
869    #[test]
870    fn builder_port_override() {
871        let server = PvaServer::builder().port(9075).udp_port(9076).build();
872        assert_eq!(server.config.tcp_port, 9075);
873        assert_eq!(server.config.udp_port, 9076);
874    }
875
876    #[test]
877    fn builder_db_string() {
878        let db = r#"
879            record(ai, "TEST:VAL") {
880                field(VAL, "3.14")
881            }
882        "#;
883        let server = PvaServer::builder().db_string(db).build();
884        let rt = tokio::runtime::Builder::new_current_thread()
885            .enable_all()
886            .build()
887            .unwrap();
888        assert!(rt.block_on(server.store.get_value("TEST:VAL")).is_some());
889    }
890
891    #[test]
892    fn builder_waveform() {
893        let data = ScalarArrayValue::F64(vec![1.0, 2.0, 3.0]);
894        let server = PvaServer::builder().waveform("T:WF", data).build();
895        let rt = tokio::runtime::Builder::new_current_thread()
896            .enable_all()
897            .build()
898            .unwrap();
899        let names = rt.block_on(server.store.pv_names());
900        assert!(names.contains(&"T:WF".to_string()));
901    }
902
903    #[test]
904    fn builder_scan_callback() {
905        let server = PvaServer::builder()
906            .ai("SCAN:V", 0.0)
907            .scan("SCAN:V", Duration::from_secs(1), |_name| {
908                ScalarValue::F64(42.0)
909            })
910            .build();
911        assert_eq!(server.scans.len(), 1);
912    }
913
914    #[test]
915    fn builder_on_put_callback() {
916        let server = PvaServer::builder()
917            .ao("PUT:V", 0.0)
918            .on_put("PUT:V", |_name, _val| {})
919            .build();
920        // on_put is stored in the SimplePvStore, not directly inspectable,
921        // but the server built without panic.
922        let rt = tokio::runtime::Builder::new_current_thread()
923            .enable_all()
924            .build()
925            .unwrap();
926        assert!(rt.block_on(server.store.get_value("PUT:V")).is_some());
927    }
928
929    #[test]
930    fn store_runtime_get_set() {
931        let server = PvaServer::builder().ao("RT:V", 0.0).build();
932        let rt = tokio::runtime::Builder::new_current_thread()
933            .enable_all()
934            .build()
935            .unwrap();
936        let store = server.store().clone();
937        rt.block_on(async {
938            assert_eq!(store.get_value("RT:V").await, Some(ScalarValue::F64(0.0)));
939            store.set_value("RT:V", ScalarValue::F64(99.0)).await;
940            assert_eq!(store.get_value("RT:V").await, Some(ScalarValue::F64(99.0)));
941        });
942    }
943
944    #[test]
945    fn link_propagates_on_set_value() {
946        let server = PvaServer::builder()
947            .ao("INPUT:A", 1.0)
948            .ao("INPUT:B", 2.0)
949            .ai("CALC:SUM", 0.0)
950            .link("CALC:SUM", &["INPUT:A", "INPUT:B"], |values| {
951                let a = match &values[0] {
952                    ScalarValue::F64(v) => *v,
953                    _ => 0.0,
954                };
955                let b = match &values[1] {
956                    ScalarValue::F64(v) => *v,
957                    _ => 0.0,
958                };
959                ScalarValue::F64(a + b)
960            })
961            .build();
962
963        let rt = tokio::runtime::Builder::new_current_thread()
964            .enable_all()
965            .build()
966            .unwrap();
967        let store = server.store().clone();
968        rt.block_on(async {
969            // Writing INPUT:A should recompute CALC:SUM = 10 + 2.
970            store.set_value("INPUT:A", ScalarValue::F64(10.0)).await;
971            assert_eq!(
972                store.get_value("CALC:SUM").await,
973                Some(ScalarValue::F64(12.0))
974            );
975
976            // Writing INPUT:B should recompute CALC:SUM = 10 + 5.
977            store.set_value("INPUT:B", ScalarValue::F64(5.0)).await;
978            assert_eq!(
979                store.get_value("CALC:SUM").await,
980                Some(ScalarValue::F64(15.0))
981            );
982        });
983    }
984}
spvirit_server/pva_server.rs

spvirit_server/
pva_server.rs
