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        self.records.insert(
163            name.clone(),
164            make_array_record(&name, RecordType::Waveform, data),
165        );
166        self
167    }
168
169    /// Add an `aai` (analog array input, read-only) record.
170    pub fn aai(mut self, name: impl Into<String>, data: ScalarArrayValue) -> Self {
171        let name = name.into();
172        self.records.insert(
173            name.clone(),
174            make_array_record(&name, RecordType::Aai, data),
175        );
176        self
177    }
178
179    /// Add an `aao` (analog array output, writable) record.
180    pub fn aao(mut self, name: impl Into<String>, data: ScalarArrayValue) -> Self {
181        let name = name.into();
182        self.records.insert(
183            name.clone(),
184            make_array_record(&name, RecordType::Aao, data),
185        );
186        self
187    }
188
189    /// Add a `subarray` record — a view into part of an array.
190    pub fn sub_array(
191        mut self,
192        name: impl Into<String>,
193        data: ScalarArrayValue,
194        indx: usize,
195        nelm: usize,
196    ) -> Self {
197        let name = name.into();
198        let ftvl = data.type_label().trim_end_matches("[]").to_string();
199        let malm = data.len();
200        let nord = nelm.min(malm.saturating_sub(indx));
201        self.records.insert(
202            name.clone(),
203            RecordInstance {
204                name: name.clone(),
205                record_type: RecordType::SubArray,
206                common: DbCommonState::default(),
207                data: RecordData::SubArray {
208                    nt: NtScalarArray::from_value(data),
209                    inp: None,
210                    ftvl,
211                    malm,
212                    nelm,
213                    nord,
214                    indx,
215                },
216                raw_fields: HashMap::new(),
217            },
218        );
219        self
220    }
221
222    /// Add an NTTable record.
223    pub fn nt_table(
224        mut self,
225        name: impl Into<String>,
226        columns: Vec<(String, ScalarArrayValue)>,
227    ) -> Self {
228        let name = name.into();
229        let labels: Vec<String> = columns.iter().map(|(n, _)| n.clone()).collect();
230        let cols: Vec<NtTableColumn> = columns
231            .into_iter()
232            .map(|(n, v)| NtTableColumn { name: n, values: v })
233            .collect();
234        self.records.insert(
235            name.clone(),
236            RecordInstance {
237                name: name.clone(),
238                record_type: RecordType::NtTable,
239                common: DbCommonState::default(),
240                data: RecordData::NtTable {
241                    nt: NtTableType {
242                        labels,
243                        columns: cols,
244                        descriptor: None,
245                        alarm: None,
246                        time_stamp: None,
247                    },
248                    inp: None,
249                    out: None,
250                    omsl: OutputMode::Supervisory,
251                },
252                raw_fields: HashMap::new(),
253            },
254        );
255        self
256    }
257
258    /// Add an NTNDArray record.
259    pub fn nt_ndarray(
260        mut self,
261        name: impl Into<String>,
262        data: ScalarArrayValue,
263        dims: Vec<(i32, i32)>,
264    ) -> Self {
265        let name = name.into();
266        let dimension: Vec<NdDimension> = dims
267            .into_iter()
268            .map(|(size, offset)| NdDimension {
269                size,
270                offset,
271                full_size: size,
272                binning: 1,
273                reverse: false,
274            })
275            .collect();
276        let uncompressed_size = (data.len() * data.element_size_bytes().max(1)) as i64;
277        self.records.insert(
278            name.clone(),
279            RecordInstance {
280                name: name.clone(),
281                record_type: RecordType::NtNdArray,
282                common: DbCommonState::default(),
283                data: RecordData::NtNdArray {
284                    nt: NtNdArrayType {
285                        value: data,
286                        codec: NdCodec {
287                            name: String::new(),
288                            parameters: Default::default(),
289                        },
290                        compressed_size: uncompressed_size,
291                        uncompressed_size,
292                        dimension,
293                        unique_id: 0,
294                        data_time_stamp: NtTimeStamp {
295                            seconds_past_epoch: 0,
296                            nanoseconds: 0,
297                            user_tag: 0,
298                        },
299                        attribute: vec![],
300                        descriptor: None,
301                        alarm: None,
302                        time_stamp: None,
303                        display: None,
304                    },
305                    inp: None,
306                    out: None,
307                    omsl: OutputMode::Supervisory,
308                },
309                raw_fields: HashMap::new(),
310            },
311        );
312        self
313    }
314
315    /// Add an `mbbi` (multi-bit binary input, read-only) NTEnum record.
316    pub fn mbbi(mut self, name: impl Into<String>, choices: Vec<String>, initial: i32) -> Self {
317        let name = name.into();
318        self.records.insert(
319            name.clone(),
320            RecordInstance {
321                name: name.clone(),
322                record_type: RecordType::Mbbi,
323                common: DbCommonState::default(),
324                data: RecordData::NtEnum {
325                    nt: NtEnum::new(initial, choices),
326                    inp: None,
327                    out: None,
328                    omsl: OutputMode::Supervisory,
329                },
330                raw_fields: HashMap::new(),
331            },
332        );
333        self
334    }
335
336    /// Add an `mbbo` (multi-bit binary output, writable) NTEnum record.
337    pub fn mbbo(mut self, name: impl Into<String>, choices: Vec<String>, initial: i32) -> Self {
338        let name = name.into();
339        self.records.insert(
340            name.clone(),
341            RecordInstance {
342                name: name.clone(),
343                record_type: RecordType::Mbbo,
344                common: DbCommonState::default(),
345                data: RecordData::NtEnum {
346                    nt: NtEnum::new(initial, choices),
347                    inp: None,
348                    out: None,
349                    omsl: OutputMode::Supervisory,
350                },
351                raw_fields: HashMap::new(),
352            },
353        );
354        self
355    }
356
357    /// Add a generic structure record with a custom struct ID and fields.
358    pub fn generic(
359        mut self,
360        name: impl Into<String>,
361        struct_id: impl Into<String>,
362        fields: Vec<(String, PvValue)>,
363    ) -> Self {
364        let name = name.into();
365        self.records.insert(
366            name.clone(),
367            RecordInstance {
368                name: name.clone(),
369                record_type: RecordType::Generic,
370                common: DbCommonState::default(),
371                data: RecordData::Generic {
372                    struct_id: struct_id.into(),
373                    fields,
374                    inp: None,
375                    out: None,
376                    omsl: OutputMode::Supervisory,
377                },
378                raw_fields: HashMap::new(),
379            },
380        );
381        self
382    }
383
384    // ─── .db file loading ────────────────────────────────────────────
385
386    /// Load records from an EPICS `.db` file.
387    pub fn db_file(mut self, path: impl AsRef<str>) -> Self {
388        match load_db(path.as_ref()) {
389            Ok(records) => {
390                self.records.extend(records);
391            }
392            Err(e) => {
393                tracing::error!("Failed to load db file '{}': {}", path.as_ref(), e);
394            }
395        }
396        self
397    }
398
399    /// Parse records from an EPICS `.db` string.
400    pub fn db_string(mut self, content: &str) -> Self {
401        match parse_db(content) {
402            Ok(records) => {
403                self.records.extend(records);
404            }
405            Err(e) => {
406                tracing::error!("Failed to parse db string: {}", e);
407            }
408        }
409        self
410    }
411
412    // ─── Callbacks ───────────────────────────────────────────────────
413
414    /// Register a callback invoked when a PUT is applied to the named PV.
415    pub fn on_put<F>(mut self, name: impl Into<String>, callback: F) -> Self
416    where
417        F: Fn(&str, &spvirit_codec::spvd_decode::DecodedValue) + Send + Sync + 'static,
418    {
419        self.on_put.insert(name.into(), Arc::new(callback));
420        self
421    }
422
423    /// Register a periodic scan callback that produces a new value for a PV.
424    pub fn scan<F>(mut self, name: impl Into<String>, period: Duration, callback: F) -> Self
425    where
426        F: Fn(&str) -> ScalarValue + Send + Sync + 'static,
427    {
428        self.scans.push((name.into(), period, Arc::new(callback)));
429        self
430    }
431
432    /// Link an output PV to one or more input PVs.
433    ///
434    /// Whenever any input PV changes (via `set_value`, protocol PUT, or
435    /// another link), the `compute` callback is invoked with the current
436    /// values of **all** inputs (in order) and the result is written to
437    /// the output PV.
438    ///
439    /// ```rust,ignore
440    /// .link("CALC:SUM", &["INPUT:A", "INPUT:B"], |values| {
441    ///     let a = values[0].as_f64().unwrap_or(0.0);
442    ///     let b = values[1].as_f64().unwrap_or(0.0);
443    ///     ScalarValue::F64(a + b)
444    /// })
445    /// ```
446    pub fn link<F>(mut self, output: impl Into<String>, inputs: &[&str], compute: F) -> Self
447    where
448        F: Fn(&[ScalarValue]) -> ScalarValue + Send + Sync + 'static,
449    {
450        self.links.push(LinkDef {
451            output: output.into(),
452            inputs: inputs.iter().map(|s| s.to_string()).collect(),
453            compute: Arc::new(compute),
454        });
455        self
456    }
457
458    // ─── External sources ────────────────────────────────────────────
459
460    /// Register an additional [`Source`] at the given priority.
461    ///
462    /// Lower `order` values are checked first during PV name resolution.
463    /// The built-in `SimplePvStore` (records added via `.ai()`, `.ao()`, etc.)
464    /// is always registered at order 0.
465    ///
466    /// ```rust,ignore
467    /// .source("hardware", -10, Arc::new(HardwareSource::new()))
468    /// ```
469    pub fn source(mut self, label: impl Into<String>, order: i32, source: Arc<dyn Source>) -> Self {
470        self.extra_sources.push((label.into(), order, source));
471        self
472    }
473
474    // ─── Configuration ───────────────────────────────────────────────
475
476    /// Set the TCP port (default 5075).
477    pub fn port(mut self, port: u16) -> Self {
478        self.tcp_port = port;
479        self
480    }
481
482    /// Set the UDP search port (default 5076).
483    pub fn udp_port(mut self, port: u16) -> Self {
484        self.udp_port = port;
485        self
486    }
487
488    /// Set the IP address to listen on.
489    pub fn listen_ip(mut self, ip: IpAddr) -> Self {
490        self.listen_ip = Some(ip);
491        self
492    }
493
494    /// Set the IP address to advertise in search responses.
495    pub fn advertise_ip(mut self, ip: IpAddr) -> Self {
496        self.advertise_ip = Some(ip);
497        self
498    }
499
500    /// Enable alarm computation from limits.
501    pub fn compute_alarms(mut self, enabled: bool) -> Self {
502        self.compute_alarms = enabled;
503        self
504    }
505
506    /// Set the beacon broadcast period in seconds (default 15).
507    pub fn beacon_period(mut self, secs: u64) -> Self {
508        self.beacon_period_secs = secs;
509        self
510    }
511
512    /// Set the idle connection timeout (default ~18 hours).
513    pub fn conn_timeout(mut self, timeout: Duration) -> Self {
514        self.conn_timeout = timeout;
515        self
516    }
517
518    /// Set the PV list mode (default [`PvListMode::List`]).
519    pub fn pvlist_mode(mut self, mode: PvListMode) -> Self {
520        self.pvlist_mode = mode;
521        self
522    }
523
524    /// Set the maximum number of PV names in pvlist responses (default 1024).
525    pub fn pvlist_max(mut self, max: usize) -> Self {
526        self.pvlist_max = max;
527        self
528    }
529
530    /// Set a regex filter for PV names exposed by pvlist.
531    pub fn pvlist_allow_pattern(mut self, pattern: Regex) -> Self {
532        self.pvlist_allow_pattern = Some(pattern);
533        self
534    }
535
536    /// Build the [`PvaServer`].
537    pub fn build(self) -> PvaServer {
538        let store = Arc::new(SimplePvStore::new(
539            self.records,
540            self.on_put,
541            self.links,
542            self.compute_alarms,
543        ));
544
545        let mut config = PvaServerConfig::default();
546        config.tcp_port = self.tcp_port;
547        config.udp_port = self.udp_port;
548        config.compute_alarms = self.compute_alarms;
549        if let Some(ip) = self.listen_ip {
550            config.listen_ip = ip;
551        }
552        config.advertise_ip = self.advertise_ip;
553        config.beacon_period_secs = self.beacon_period_secs;
554        config.conn_timeout = self.conn_timeout;
555        config.pvlist_mode = self.pvlist_mode;
556        config.pvlist_max = self.pvlist_max;
557        config.pvlist_allow_pattern = self.pvlist_allow_pattern;
558
559        PvaServer {
560            store,
561            extra_sources: self.extra_sources,
562            config,
563            scans: self.scans,
564            monitor_registry: None,
565        }
566    }
567}
568
569// ─── PvaServer ───────────────────────────────────────────────────────────
570
571/// High-level PVAccess server.
572///
573/// Built via [`PvaServer::builder()`] with typed record constructors,
574/// `.db_file()` loading, `.on_put()` / `.scan()` callbacks, and a
575/// simple `.run()` to start serving.
576///
577/// ```rust,ignore
578/// let server = PvaServer::builder()
579///     .ai("SIM:TEMP", 22.5)
580///     .ao("SIM:SP", 25.0)
581///     .build();
582///
583/// // Read/write PVs from another task:
584/// let store = server.store();
585/// store.set_value("SIM:TEMP", ScalarValue::F64(23.1)).await;
586///
587/// server.run().await?;
588/// ```
589pub struct PvaServer {
590    store: Arc<SimplePvStore>,
591    extra_sources: Vec<(String, i32, Arc<dyn Source>)>,
592    config: PvaServerConfig,
593    scans: Vec<(String, Duration, ScanCallback)>,
594    /// Optional pre-supplied monitor registry so external code (e.g. Python
595    /// bindings) can notify monitors from outside `run()`.
596    monitor_registry: Option<Arc<MonitorRegistry>>,
597}
598
599impl PvaServer {
600    /// Create a builder for configuring a [`PvaServer`].
601    pub fn builder() -> PvaServerBuilder {
602        PvaServerBuilder::new()
603    }
604
605    /// Get a reference to the underlying store for runtime get/put.
606    pub fn store(&self) -> &Arc<SimplePvStore> {
607        &self.store
608    }
609
610    /// Mint a typed handle to any record in this server's store — the
611    /// pre-`run()` counterpart of [`RunningServer::pv`].
612    pub async fn pv<T: crate::pv::PvScalar>(
613        &self,
614        name: &str,
615    ) -> Result<crate::pv::Pv<T>, crate::pv::PvError> {
616        crate::pv::Pv::attach(&self.store, name).await
617    }
618
619    /// Mint an array handle to any record in this server's store — the
620    /// pre-`run()` counterpart of [`RunningServer::array_pv`].
621    pub async fn array_pv(&self, name: &str) -> Result<crate::pv::PvArray, crate::pv::PvError> {
622        crate::pv::PvArray::attach(&self.store, name).await
623    }
624
625    /// Register an additional [`Source`] after building the server.
626    ///
627    /// This is useful when the source needs a reference to the store
628    /// (which is only available after `.build()`).
629    ///
630    /// ```rust,ignore
631    /// let server = PvaServer::builder().ai("X", 0.0).build();
632    /// let store = server.store().clone();
633    /// server.add_source("agg", 10, Arc::new(MyAggSource::new(store)));
634    /// server.run().await?;
635    /// ```
636    pub fn add_source(&mut self, label: impl Into<String>, order: i32, source: Arc<dyn Source>) {
637        self.extra_sources.push((label.into(), order, source));
638    }
639
640    /// Pre-supply the [`MonitorRegistry`] that [`Self::run`] will use.
641    ///
642    /// This lets external code (for example Python `Source` adapters)
643    /// hold onto the registry and publish monitor updates to subscribed
644    /// PVAccess clients from outside `run()`.
645    pub fn set_monitor_registry(&mut self, registry: Arc<MonitorRegistry>) {
646        self.monitor_registry = Some(registry);
647    }
648
649    /// Get a shared handle to the [`MonitorRegistry`] that will be used
650    /// when [`Self::run`] starts.  Creates (and stores) a new registry
651    /// on first call so external code can register before run.
652    pub fn monitor_registry(&mut self) -> Arc<MonitorRegistry> {
653        if self.monitor_registry.is_none() {
654            self.monitor_registry = Some(Arc::new(MonitorRegistry::new()));
655        }
656        self.monitor_registry.as_ref().unwrap().clone()
657    }
658
659    /// Start the PVA server (UDP search + TCP handler + beacon + scan tasks).
660    ///
661    /// This blocks until the server is shut down or an error occurs.
662    pub async fn run(self) -> Result<(), Box<dyn std::error::Error>> {
663        // Create the monitor registry early so scan tasks can notify
664        // PVAccess monitor clients when values change.
665        let registry = self
666            .monitor_registry
667            .clone()
668            .unwrap_or_else(|| Arc::new(MonitorRegistry::new()));
669        self.store.set_registry(registry.clone()).await;
670
671        // Build the source registry with the built-in store at order 0.
672        let sources = Arc::new(SourceRegistry::new());
673        sources.add("builtin", 0, self.store.clone()).await;
674
675        // IOC/QSRV-style record field access (<name>.<FIELD>, <FIELD>$) so
676        // tools like the EPICS Archiver Appliance can fetch record metadata.
677        sources
678            .add(
679                "record-fields",
680                10,
681                Arc::new(crate::record_fields::RecordFieldSource::new(
682                    self.store.clone(),
683                )),
684            )
685            .await;
686
687        // Register any extra sources provided via .source().
688        for (label, order, source) in &self.extra_sources {
689            sources.add(label.clone(), *order, source.clone()).await;
690        }
691
692        // Spawn scan tasks.
693        for (name, period, callback) in &self.scans {
694            let store = self.store.clone();
695            let name = name.clone();
696            let period = *period;
697            let callback = callback.clone();
698            tokio::spawn(async move {
699                let mut interval = tokio::time::interval(period);
700                loop {
701                    interval.tick().await;
702                    let new_val = callback(&name);
703                    store.set_value(&name, new_val).await;
704                }
705            });
706        }
707
708        let pv_count = self.store.pv_names().await.len();
709        info!(
710            "PvaServer starting: {} PVs on port {}",
711            pv_count, self.config.tcp_port
712        );
713
714        run_pva_server_with_registry(sources, self.config, registry).await
715    }
716}
717
718// ─── Handle-based (`Pv<T>`) entry point ──────────────────────────────────
719
720impl PvaServer {
721    /// Serve a collection of typed PV handles. Shorthand entry point for the
722    /// handle-based API; combine with `.db_file()`, `.source()`, etc.
723    pub fn serve(pvs: impl IntoIterator<Item = impl Into<crate::pv::AnyPv>>) -> ServeBuilder {
724        ServeBuilder {
725            inner: PvaServerBuilder::new(),
726            handles: Vec::new(),
727        }
728        .pvs(pvs)
729    }
730}
731
732/// Builder for the PV-handle API. Wraps [`PvaServerBuilder`] and adds handle
733/// binding at `build()` time.
734pub struct ServeBuilder {
735    inner: PvaServerBuilder,
736    handles: Vec<crate::pv::AnyPv>,
737}
738
739impl ServeBuilder {
740    pub fn pvs(mut self, pvs: impl IntoIterator<Item = impl Into<crate::pv::AnyPv>>) -> Self {
741        self.handles.extend(pvs.into_iter().map(Into::into));
742        self
743    }
744    pub fn db_file(mut self, path: impl AsRef<str>) -> Self {
745        self.inner = self.inner.db_file(path);
746        self
747    }
748    pub fn db_string(mut self, content: &str) -> Self {
749        self.inner = self.inner.db_string(content);
750        self
751    }
752    pub fn source(mut self, label: impl Into<String>, order: i32, source: Arc<dyn Source>) -> Self {
753        self.inner = self.inner.source(label, order, source);
754        self
755    }
756    pub fn port(mut self, port: u16) -> Self {
757        self.inner = self.inner.port(port);
758        self
759    }
760    pub fn udp_port(mut self, port: u16) -> Self {
761        self.inner = self.inner.udp_port(port);
762        self
763    }
764    pub fn listen_ip(mut self, ip: IpAddr) -> Self {
765        self.inner = self.inner.listen_ip(ip);
766        self
767    }
768    pub fn advertise_ip(mut self, ip: IpAddr) -> Self {
769        self.inner = self.inner.advertise_ip(ip);
770        self
771    }
772    pub fn compute_alarms(mut self, enabled: bool) -> Self {
773        self.inner = self.inner.compute_alarms(enabled);
774        self
775    }
776    pub fn beacon_period(mut self, secs: u64) -> Self {
777        self.inner = self.inner.beacon_period(secs);
778        self
779    }
780
781    /// Materialise records, links and scans from the handles, build the
782    /// server, then bind every handle to the store.
783    ///
784    /// Async because registering PUT validators post-build goes through
785    /// `SimplePvStore::set_validator`, which is async (an `RwLock` write);
786    /// there is no synchronous alternative and `spvirit-server` does not
787    /// depend on `futures`, so this awaits inline rather than blocking.
788    pub async fn build(mut self) -> PvaServer {
789        let mut validators: Vec<(String, crate::simple_store::PutValidator)> = Vec::new();
790        for h in &self.handles {
791            let name = h.name().to_string();
792            if let Some(rec) = h.take_record() {
793                self.inner.records.insert(name.clone(), rec);
794            }
795            if let Some(v) = h.take_validator() {
796                validators.push((name.clone(), v));
797            }
798            if let Some((period, cb)) = h.take_scan() {
799                self.inner.scans.push((name.clone(), period, cb));
800            }
801            if let Some((inputs, compute)) = h.take_calc() {
802                self.inner.links.push(LinkDef {
803                    output: name.clone(),
804                    inputs,
805                    compute,
806                });
807            }
808        }
809        let server = self.inner.build();
810        let store = server.store().clone();
811        for h in &self.handles {
812            h.bind(&store);
813        }
814        for (name, v) in validators {
815            store.set_validator(name, v).await;
816        }
817        server
818    }
819
820    /// Build and run (blocks until shutdown).
821    pub async fn run(self) -> Result<(), Box<dyn std::error::Error>> {
822        self.build().await.run().await
823    }
824
825    /// Build and spawn; returns a handle for typed access and shutdown.
826    pub async fn start(self) -> RunningServer {
827        let server = self.build().await;
828        let store = server.store().clone();
829        let handle = tokio::spawn(async move {
830            if let Err(e) = server.run().await {
831                tracing::error!("PvaServer exited with error: {e}");
832            }
833        });
834        RunningServer { store, handle }
835    }
836}
837
838/// A started server: mint typed handles, then `abort()` to stop.
839pub struct RunningServer {
840    store: Arc<SimplePvStore>,
841    handle: tokio::task::JoinHandle<()>,
842}
843
844impl RunningServer {
845    /// Mint a typed handle to any served record (handle-built or `.db`-loaded).
846    pub async fn pv<T: crate::pv::PvScalar>(
847        &self,
848        name: &str,
849    ) -> Result<crate::pv::Pv<T>, crate::pv::PvError> {
850        crate::pv::Pv::attach(&self.store, name).await
851    }
852
853    /// Mint an array handle to any served record (handle-built or `.db`-loaded).
854    pub async fn array_pv(&self, name: &str) -> Result<crate::pv::PvArray, crate::pv::PvError> {
855        crate::pv::PvArray::attach(&self.store, name).await
856    }
857
858    pub fn store(&self) -> &Arc<SimplePvStore> {
859        &self.store
860    }
861
862    pub fn abort(&self) {
863        self.handle.abort();
864    }
865}
866
867// ─── Record construction helpers ─────────────────────────────────────────
868
869pub(crate) fn make_scalar_record(
870    name: &str,
871    record_type: RecordType,
872    value: ScalarValue,
873) -> RecordInstance {
874    let nt = NtScalar::from_value(value);
875    let data = match record_type {
876        RecordType::Ai => RecordData::Ai {
877            nt,
878            inp: None,
879            siml: None,
880            siol: None,
881            simm: false,
882        },
883        RecordType::Bi => RecordData::Bi {
884            nt,
885            inp: None,
886            znam: "Off".to_string(),
887            onam: "On".to_string(),
888            siml: None,
889            siol: None,
890            simm: false,
891        },
892        RecordType::StringIn => RecordData::StringIn {
893            nt,
894            inp: None,
895            siml: None,
896            siol: None,
897            simm: false,
898        },
899        // longin reuses the Ai data shape (NtScalar input record)
900        RecordType::LongIn => RecordData::Ai {
901            nt,
902            inp: None,
903            siml: None,
904            siol: None,
905            simm: false,
906        },
907        _ => panic!("make_scalar_record: unsupported type {record_type:?}"),
908    };
909    RecordInstance {
910        name: name.to_string(),
911        record_type,
912        common: DbCommonState::default(),
913        data,
914        raw_fields: HashMap::new(),
915    }
916}
917
918pub(crate) fn make_output_record(
919    name: &str,
920    record_type: RecordType,
921    value: ScalarValue,
922) -> RecordInstance {
923    let nt = NtScalar::from_value(value);
924    let data = match record_type {
925        RecordType::Ao => RecordData::Ao {
926            nt,
927            out: None,
928            dol: None,
929            omsl: OutputMode::Supervisory,
930            drvl: None,
931            drvh: None,
932            oroc: None,
933            siml: None,
934            siol: None,
935            simm: false,
936        },
937        RecordType::Bo => RecordData::Bo {
938            nt,
939            out: None,
940            dol: None,
941            omsl: OutputMode::Supervisory,
942            znam: "Off".to_string(),
943            onam: "On".to_string(),
944            siml: None,
945            siol: None,
946            simm: false,
947        },
948        RecordType::StringOut => RecordData::StringOut {
949            nt,
950            out: None,
951            dol: None,
952            omsl: OutputMode::Supervisory,
953            siml: None,
954            siol: None,
955            simm: false,
956        },
957        // longout reuses the Ao data shape (NtScalar output record)
958        RecordType::LongOut => RecordData::Ao {
959            nt,
960            out: None,
961            dol: None,
962            omsl: OutputMode::Supervisory,
963            drvl: None,
964            drvh: None,
965            oroc: None,
966            siml: None,
967            siol: None,
968            simm: false,
969        },
970        _ => panic!("make_output_record: unsupported type {record_type:?}"),
971    };
972    RecordInstance {
973        name: name.to_string(),
974        record_type,
975        common: DbCommonState::default(),
976        data,
977        raw_fields: HashMap::new(),
978    }
979}
980
981/// Build an array-backed record (`waveform`/`aai`/`aao`) with ftvl/nelm/nord
982/// inferred from `data`. Shared by the classic builder (`.waveform`/`.aai`/
983/// `.aao`) and `PvArray`'s constructors so the inference lives in one place.
984pub(crate) fn make_array_record(
985    name: &str,
986    record_type: RecordType,
987    data: ScalarArrayValue,
988) -> RecordInstance {
989    let ftvl = data.type_label().trim_end_matches("[]").to_string();
990    let nelm = data.len();
991    let nt = NtScalarArray::from_value(data);
992    let record_data = match record_type {
993        RecordType::Waveform => RecordData::Waveform {
994            nt,
995            inp: None,
996            ftvl,
997            nelm,
998            nord: nelm,
999        },
1000        RecordType::Aai => RecordData::Aai {
1001            nt,
1002            inp: None,
1003            ftvl,
1004            nelm,
1005            nord: nelm,
1006        },
1007        RecordType::Aao => RecordData::Aao {
1008            nt,
1009            out: None,
1010            dol: None,
1011            omsl: OutputMode::Supervisory,
1012            ftvl,
1013            nelm,
1014            nord: nelm,
1015        },
1016        _ => panic!("make_array_record: unsupported type {record_type:?}"),
1017    };
1018    RecordInstance {
1019        name: name.to_string(),
1020        record_type,
1021        common: DbCommonState::default(),
1022        data: record_data,
1023        raw_fields: HashMap::new(),
1024    }
1025}
1026
1027#[cfg(test)]
1028mod tests {
1029    use super::*;
1030
1031    #[test]
1032    fn builder_creates_records() {
1033        let server = PvaServer::builder()
1034            .ai("T:AI", 1.0)
1035            .ao("T:AO", 2.0)
1036            .bi("T:BI", true)
1037            .bo("T:BO", false)
1038            .string_in("T:SI", "hello")
1039            .string_out("T:SO", "world")
1040            .build();
1041
1042        let rt = tokio::runtime::Builder::new_current_thread()
1043            .enable_all()
1044            .build()
1045            .unwrap();
1046        let names = rt.block_on(server.store.pv_names());
1047        assert_eq!(names.len(), 6);
1048    }
1049
1050    #[test]
1051    fn builder_defaults() {
1052        let server = PvaServer::builder().build();
1053        assert_eq!(server.config.tcp_port, 5075);
1054        assert_eq!(server.config.udp_port, 5076);
1055        assert!(!server.config.compute_alarms);
1056    }
1057
1058    #[test]
1059    fn builder_port_override() {
1060        let server = PvaServer::builder().port(9075).udp_port(9076).build();
1061        assert_eq!(server.config.tcp_port, 9075);
1062        assert_eq!(server.config.udp_port, 9076);
1063    }
1064
1065    #[test]
1066    fn builder_db_string() {
1067        let db = r#"
1068            record(ai, "TEST:VAL") {
1069                field(VAL, "3.14")
1070            }
1071        "#;
1072        let server = PvaServer::builder().db_string(db).build();
1073        let rt = tokio::runtime::Builder::new_current_thread()
1074            .enable_all()
1075            .build()
1076            .unwrap();
1077        assert!(rt.block_on(server.store.get_value("TEST:VAL")).is_some());
1078    }
1079
1080    #[test]
1081    fn builder_waveform() {
1082        let data = ScalarArrayValue::F64(vec![1.0, 2.0, 3.0]);
1083        let server = PvaServer::builder().waveform("T:WF", data).build();
1084        let rt = tokio::runtime::Builder::new_current_thread()
1085            .enable_all()
1086            .build()
1087            .unwrap();
1088        let names = rt.block_on(server.store.pv_names());
1089        assert!(names.contains(&"T:WF".to_string()));
1090    }
1091
1092    #[test]
1093    fn builder_scan_callback() {
1094        let server = PvaServer::builder()
1095            .ai("SCAN:V", 0.0)
1096            .scan("SCAN:V", Duration::from_secs(1), |_name| {
1097                ScalarValue::F64(42.0)
1098            })
1099            .build();
1100        assert_eq!(server.scans.len(), 1);
1101    }
1102
1103    #[test]
1104    fn builder_on_put_callback() {
1105        let server = PvaServer::builder()
1106            .ao("PUT:V", 0.0)
1107            .on_put("PUT:V", |_name, _val| {})
1108            .build();
1109        // on_put is stored in the SimplePvStore, not directly inspectable,
1110        // but the server built without panic.
1111        let rt = tokio::runtime::Builder::new_current_thread()
1112            .enable_all()
1113            .build()
1114            .unwrap();
1115        assert!(rt.block_on(server.store.get_value("PUT:V")).is_some());
1116    }
1117
1118    #[test]
1119    fn store_runtime_get_set() {
1120        let server = PvaServer::builder().ao("RT:V", 0.0).build();
1121        let rt = tokio::runtime::Builder::new_current_thread()
1122            .enable_all()
1123            .build()
1124            .unwrap();
1125        let store = server.store().clone();
1126        rt.block_on(async {
1127            assert_eq!(store.get_value("RT:V").await, Some(ScalarValue::F64(0.0)));
1128            store.set_value("RT:V", ScalarValue::F64(99.0)).await;
1129            assert_eq!(store.get_value("RT:V").await, Some(ScalarValue::F64(99.0)));
1130        });
1131    }
1132
1133    #[test]
1134    fn link_propagates_on_set_value() {
1135        let server = PvaServer::builder()
1136            .ao("INPUT:A", 1.0)
1137            .ao("INPUT:B", 2.0)
1138            .ai("CALC:SUM", 0.0)
1139            .link("CALC:SUM", &["INPUT:A", "INPUT:B"], |values| {
1140                let a = match &values[0] {
1141                    ScalarValue::F64(v) => *v,
1142                    _ => 0.0,
1143                };
1144                let b = match &values[1] {
1145                    ScalarValue::F64(v) => *v,
1146                    _ => 0.0,
1147                };
1148                ScalarValue::F64(a + b)
1149            })
1150            .build();
1151
1152        let rt = tokio::runtime::Builder::new_current_thread()
1153            .enable_all()
1154            .build()
1155            .unwrap();
1156        let store = server.store().clone();
1157        rt.block_on(async {
1158            // Writing INPUT:A should recompute CALC:SUM = 10 + 2.
1159            store.set_value("INPUT:A", ScalarValue::F64(10.0)).await;
1160            assert_eq!(
1161                store.get_value("CALC:SUM").await,
1162                Some(ScalarValue::F64(12.0))
1163            );
1164
1165            // Writing INPUT:B should recompute CALC:SUM = 10 + 5.
1166            store.set_value("INPUT:B", ScalarValue::F64(5.0)).await;
1167            assert_eq!(
1168                store.get_value("CALC:SUM").await,
1169                Some(ScalarValue::F64(15.0))
1170            );
1171        });
1172    }
1173
1174    use crate::pv::{AnyPv, Pv};
1175
1176    #[tokio::test]
1177    async fn serve_builder_binds_handles_and_registers_everything() {
1178        let temp = Pv::ai("S:T", 22.5).mdel(0.1);
1179        let sp = Pv::ao("S:SP", 25.0).on_put(|_pv, _v: f64| Ok(()));
1180        let a = Pv::ai("S:A", 1.0);
1181        let b = Pv::ai("S:B", 2.0);
1182        let sum = Pv::calc("S:SUM", &[&a, &b], |vals| vals.iter().sum());
1183
1184        // Rust can't infer the `impl Into<AnyPv>` target type per-element
1185        // inside an array literal (E0283), so each handle is converted
1186        // explicitly here rather than via a bare `.into()`.
1187        let server = PvaServer::serve([AnyPv::from(temp.clone()), AnyPv::from(sp)])
1188            .pvs([
1189                AnyPv::from(a.clone()),
1190                AnyPv::from(b),
1191                AnyPv::from(sum.clone()),
1192            ])
1193            .build()
1194            .await;
1195
1196        // handles are bound: typed set/get works against the built store
1197        temp.set(23.0).await.unwrap();
1198        assert_eq!(temp.get().await, Ok(23.0));
1199
1200        // calc evaluated on input change
1201        a.set(10.0).await.unwrap();
1202        assert_eq!(sum.get().await, Ok(12.0));
1203
1204        // record made it into the store with its raw fields
1205        let rec = server.store().get_record("S:T").await.unwrap();
1206        assert_eq!(rec.raw_fields.get("MDEL").map(String::as_str), Some("0.1"));
1207    }
1208
1209    #[tokio::test]
1210    async fn running_server_mints_handles_to_db_records() {
1211        // parse_db is line-oriented (one `record(...)`/`field(...)`
1212        // statement per line); a packed one-liner silently drops its
1213        // fields, so this uses the same multi-line shape as the other
1214        // db_string tests in this module.
1215        let server = PvaServer::serve(Vec::<AnyPv>::new())
1216            .db_string("record(ao, \"DB:X\") {\n    field(VAL, \"2.5\")\n}")
1217            .build()
1218            .await;
1219        let store = server.store().clone();
1220        let h: crate::pv::Pv<f64> = crate::pv::Pv::attach(&store, "DB:X").await.unwrap();
1221        assert_eq!(h.get().await, Ok(2.5));
1222    }
1223
1224    #[tokio::test]
1225    async fn homogeneous_iterator_feeds_serve_without_manual_erasure() {
1226        let bpms: Vec<Pv<f64>> = (0..100)
1227            .map(|i| Pv::ai(format!("BPM:{i:03}:X"), 0.0))
1228            .collect();
1229        let server = PvaServer::serve(bpms.iter().cloned()).build().await;
1230        assert_eq!(server.store().pv_names().await.len(), 100);
1231        bpms[42].set(1.23).await.unwrap();
1232        assert_eq!(bpms[42].get().await, Ok(1.23));
1233    }
1234
1235    #[tokio::test]
1236    async fn pva_server_mints_typed_handles_pre_run() {
1237        let server = PvaServer::serve([AnyPv::from(Pv::ai("PRE:X", 5.0))])
1238            .build()
1239            .await;
1240        let h: crate::pv::Pv<f64> = server.pv("PRE:X").await.unwrap();
1241        assert_eq!(h.get().await, Ok(5.0));
1242        assert!(matches!(
1243            server.pv::<bool>("PRE:X").await,
1244            Err(crate::pv::PvError::TypeMismatch { .. })
1245        ));
1246    }
1247}
spvirit_server/pva_server.rs

spvirit_server/
pva_server.rs
