1use 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
39pub 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 pub fn port(mut self, port: u16) -> Self {
479 self.tcp_port = port;
480 self
481 }
482
483 pub fn udp_port(mut self, port: u16) -> Self {
485 self.udp_port = port;
486 self
487 }
488
489 pub fn listen_ip(mut self, ip: IpAddr) -> Self {
491 self.listen_ip = Some(ip);
492 self
493 }
494
495 pub fn advertise_ip(mut self, ip: IpAddr) -> Self {
497 self.advertise_ip = Some(ip);
498 self
499 }
500
501 pub fn compute_alarms(mut self, enabled: bool) -> Self {
503 self.compute_alarms = enabled;
504 self
505 }
506
507 pub fn beacon_period(mut self, secs: u64) -> Self {
509 self.beacon_period_secs = secs;
510 self
511 }
512
513 pub fn conn_timeout(mut self, timeout: Duration) -> Self {
515 self.conn_timeout = timeout;
516 self
517 }
518
519 pub fn pvlist_mode(mut self, mode: PvListMode) -> Self {
521 self.pvlist_mode = mode;
522 self
523 }
524
525 pub fn pvlist_max(mut self, max: usize) -> Self {
527 self.pvlist_max = max;
528 self
529 }
530
531 pub fn pvlist_allow_pattern(mut self, pattern: Regex) -> Self {
533 self.pvlist_allow_pattern = Some(pattern);
534 self
535 }
536
537 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
570pub 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 monitor_registry: Option<Arc<MonitorRegistry>>,
598}
599
600impl PvaServer {
601 pub fn builder() -> PvaServerBuilder {
603 PvaServerBuilder::new()
604 }
605
606 pub fn store(&self) -> &Arc<SimplePvStore> {
608 &self.store
609 }
610
611 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 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 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 pub fn set_monitor_registry(&mut self, registry: Arc<MonitorRegistry>) {
647 self.monitor_registry = Some(registry);
648 }
649
650 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 pub async fn run(self) -> Result<(), Box<dyn std::error::Error>> {
664 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 let sources = Arc::new(SourceRegistry::new());
674 sources.add("builtin", 0, self.store.clone()).await;
675
676 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 for (label, order, source) in &self.extra_sources {
690 sources.add(label.clone(), *order, source.clone()).await;
691 }
692
693 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
719impl PvaServer {
722 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
733pub 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 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 pub async fn run(self) -> Result<(), Box<dyn std::error::Error>> {
823 self.build().await.run().await
824 }
825
826 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
839pub struct RunningServer {
841 store: Arc<SimplePvStore>,
842 handle: tokio::task::JoinHandle<()>,
843}
844
845impl RunningServer {
846 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 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 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 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 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 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
929pub(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 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 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
1043pub(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 let h = server.add_scalar("RT:U32", ScalarValue::U32(7), true).await;
1148 assert_eq!(h.get().await.unwrap(), ScalarValue::U32(7));
1149 assert!(matches!(
1151 server.store().get_value("RT:U32").await,
1152 Some(ScalarValue::U32(7))
1153 ));
1154
1155 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 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 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 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 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 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 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 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 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 temp.set(23.0).await.unwrap();
1392 assert_eq!(temp.get().await, Ok(23.0));
1393
1394 a.set(10.0).await.unwrap();
1396 assert_eq!(sum.get().await, Ok(12.0));
1397
1398 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 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}