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

spvirit_server/
pva_server.rs
