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

spvirit_server/
pva_server.rs
