spvirit_server/
simple_store.rs

1//! A simple in-memory [`Source`] implementation backed by `RecordInstance`.
2//!
3//! Used by [`PvaServer`](crate::pva_server::PvaServer) to serve PVs without
4//! requiring an external database.
5
6use std::collections::{HashMap, HashSet};
7use std::sync::Arc;
8
9use tokio::sync::{RwLock, mpsc};
10use tracing::debug;
11
12use std::future::Future;
13use std::pin::Pin;
14
15use spvirit_codec::spvd_decode::{DecodedValue, FieldDesc, FieldType, StructureDesc, TypeCode};
16use spvirit_types::{NtPayload, ScalarArrayValue, ScalarValue};
17
18use crate::apply::{
19    apply_alarm_update, apply_control_update, apply_display_update, apply_scalar_array_put,
20    apply_value_update,
21};
22use crate::monitor::MonitorRegistry;
23use crate::pvstore::{PvInfo, Source};
24use crate::types::{RecordData, RecordInstance};
25
26/// Callback invoked after a PUT value is applied to a record.
27pub type OnPutCallback = Arc<dyn Fn(&str, &DecodedValue) + Send + Sync>;
28
29/// Callback invoked by the scan scheduler; returns the new value for the PV.
30pub type ScanCallback = Arc<dyn Fn(&str) -> ScalarValue + Send + Sync>;
31
32/// Callback that computes a derived PV value from its input values.
33pub type LinkCallback = Arc<dyn Fn(&[ScalarValue]) -> ScalarValue + Send + Sync>;
34
35/// A link from one or more input PVs to a computed output PV.
36pub(crate) struct LinkDef {
37    pub output: String,
38    pub inputs: Vec<String>,
39    pub compute: LinkCallback,
40}
41
42struct PvEntry {
43    record: RecordInstance,
44    subscribers: Vec<mpsc::Sender<NtPayload>>,
45    /// Value carried by the last update posted to subscribers/monitors —
46    /// the reference point for the MDEL monitor deadband. `None` until the
47    /// first post (so the first change always posts).
48    last_posted: Option<f64>,
49}
50
51/// A simple in-memory PV store.
52pub struct SimplePvStore {
53    pvs: RwLock<HashMap<String, PvEntry>>,
54    on_put: HashMap<String, OnPutCallback>,
55    links: Vec<LinkDef>,
56    compute_alarms: bool,
57    registry: RwLock<Option<Arc<MonitorRegistry>>>,
58}
59
60impl SimplePvStore {
61    pub(crate) fn new(
62        records: HashMap<String, RecordInstance>,
63        on_put: HashMap<String, OnPutCallback>,
64        links: Vec<LinkDef>,
65        compute_alarms: bool,
66    ) -> Self {
67        let pvs = records
68            .into_iter()
69            .map(|(name, record)| {
70                let last_posted = initial_posted(&record);
71                (
72                    name,
73                    PvEntry {
74                        record,
75                        subscribers: Vec::new(),
76                        last_posted,
77                    },
78                )
79            })
80            .collect();
81        Self {
82            pvs: RwLock::new(pvs),
83            on_put,
84            links,
85            compute_alarms,
86            registry: RwLock::new(None),
87        }
88    }
89
90    /// Attach the [`MonitorRegistry`] so that `set_value` can push updates
91    /// to PVAccess monitor clients.  Called automatically by [`PvaServer::run`].
92    pub async fn set_registry(&self, registry: Arc<MonitorRegistry>) {
93        *self.registry.write().await = Some(registry);
94    }
95
96    /// Insert or replace a PV record at runtime.
97    pub async fn insert(&self, name: String, record: RecordInstance) {
98        let mut pvs = self.pvs.write().await;
99        let last_posted = initial_posted(&record);
100        pvs.insert(
101            name,
102            PvEntry {
103                record,
104                subscribers: Vec::new(),
105                last_posted,
106            },
107        );
108    }
109
110    /// Read the current [`ScalarValue`] of a PV.
111    pub async fn get_value(&self, name: &str) -> Option<ScalarValue> {
112        let pvs = self.pvs.read().await;
113        pvs.get(name).map(|e| e.record.current_value())
114    }
115
116    /// Read a clone of the full [`RecordInstance`] backing a PV.
117    ///
118    /// Used by [`RecordFieldSource`](crate::record_fields::RecordFieldSource)
119    /// to serve `<name>.<FIELD>` channels.
120    pub async fn get_record(&self, name: &str) -> Option<RecordInstance> {
121        let pvs = self.pvs.read().await;
122        pvs.get(name).map(|e| e.record.clone())
123    }
124
125    /// Read the full [`NtPayload`] of a PV.
126    pub async fn get_nt(&self, name: &str) -> Option<NtPayload> {
127        let pvs = self.pvs.read().await;
128        pvs.get(name).map(|e| e.record.to_ntpayload())
129    }
130
131    /// Write a [`ScalarValue`] to a PV (bypasses on_put).
132    pub async fn set_value(&self, name: &str, value: ScalarValue) -> bool {
133        if self.set_value_inner(name, value).await {
134            self.evaluate_links(name).await;
135            true
136        } else {
137            false
138        }
139    }
140
141    /// Write a [`ScalarArrayValue`] to an array PV (bypasses on_put).
142    pub async fn set_array_value(&self, name: &str, value: ScalarArrayValue) -> bool {
143        if self.set_array_value_inner(name, value).await {
144            self.evaluate_links(name).await;
145            true
146        } else {
147            false
148        }
149    }
150
151    /// Write a full [`NtPayload`] to a PV (bypasses on_put).
152    pub async fn put_nt(&self, name: &str, payload: NtPayload) -> bool {
153        if self.put_nt_inner(name, payload).await {
154            self.evaluate_links(name).await;
155            true
156        } else {
157            false
158        }
159    }
160
161    /// Core write logic — updates the value, notifies subscribers and monitors,
162    /// but does **not** trigger link evaluation (to avoid recursion).
163    async fn set_value_inner(&self, name: &str, value: ScalarValue) -> bool {
164        let payload = {
165            let mut pvs = self.pvs.write().await;
166            if let Some(entry) = pvs.get_mut(name) {
167                let prev_severity = entry.record.to_ntscalar().alarm_severity;
168                let changed = entry.record.set_scalar_value(value, self.compute_alarms);
169                if changed {
170                    if !should_post_update(entry, prev_severity) {
171                        // Changed, but within the MDEL monitor deadband:
172                        // the record holds the new value (GETs see it), just
173                        // no update is posted to subscribers/monitors.
174                        return true;
175                    }
176                    let payload = entry.record.to_ntpayload();
177                    entry
178                        .subscribers
179                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
180                    Some(payload)
181                } else {
182                    None
183                }
184            } else {
185                return false;
186            }
187        };
188
189        if let Some(payload) = payload {
190            // Notify PVAccess monitor clients (if the registry is attached).
191            let reg = self.registry.read().await;
192            if let Some(registry) = reg.as_ref() {
193                registry.notify_monitors(name, &payload).await;
194            }
195            true
196        } else {
197            false
198        }
199    }
200
201    /// Core array write logic — updates the value, notifies subscribers and monitors,
202    /// but does **not** trigger link evaluation (to avoid recursion).
203    async fn set_array_value_inner(&self, name: &str, value: ScalarArrayValue) -> bool {
204        let payload = {
205            let mut pvs = self.pvs.write().await;
206            if let Some(entry) = pvs.get_mut(name) {
207                let changed = entry.record.set_array_value(value);
208                if changed {
209                    let payload = entry.record.to_ntpayload();
210                    entry
211                        .subscribers
212                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
213                    Some(payload)
214                } else {
215                    None
216                }
217            } else {
218                return false;
219            }
220        };
221
222        if let Some(payload) = payload {
223            // Notify PVAccess monitor clients (if the registry is attached).
224            let reg = self.registry.read().await;
225            if let Some(registry) = reg.as_ref() {
226                registry.notify_monitors(name, &payload).await;
227            }
228            true
229        } else {
230            false
231        }
232    }
233
234    /// Core NtPayload write logic — updates the payload, notifies subscribers
235    /// and monitors, but does **not** trigger link evaluation.
236    async fn put_nt_inner(&self, name: &str, payload: NtPayload) -> bool {
237        let payload = {
238            let mut pvs = self.pvs.write().await;
239            if let Some(entry) = pvs.get_mut(name) {
240                let changed = entry.record.set_nt_payload(payload);
241                if changed {
242                    let payload = entry.record.to_ntpayload();
243                    entry
244                        .subscribers
245                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
246                    Some(payload)
247                } else {
248                    None
249                }
250            } else {
251                return false;
252            }
253        };
254
255        if let Some(payload) = payload {
256            // Notify PVAccess monitor clients (if the registry is attached).
257            let reg = self.registry.read().await;
258            if let Some(registry) = reg.as_ref() {
259                registry.notify_monitors(name, &payload).await;
260            }
261            true
262        } else {
263            false
264        }
265    }
266
267    /// Walk every link whose inputs include `changed_pv`, compute the output,
268    /// and propagate (BFS with cycle detection).
269    async fn evaluate_links(&self, changed_pv: &str) {
270        if self.links.is_empty() {
271            return;
272        }
273        let mut queue = vec![changed_pv.to_string()];
274        let mut visited = HashSet::new();
275
276        while let Some(pv) = queue.pop() {
277            if !visited.insert(pv.clone()) {
278                debug!("Circular link detected for PV '{}', skipping", pv);
279                continue;
280            }
281            for link in &self.links {
282                if !link.inputs.iter().any(|i| i == &pv) {
283                    continue;
284                }
285                // Gather current values of all inputs.
286                let values = {
287                    let pvs = self.pvs.read().await;
288                    link.inputs
289                        .iter()
290                        .map(|n| {
291                            pvs.get(n)
292                                .map(|e| e.record.current_value())
293                                .unwrap_or(ScalarValue::F64(0.0))
294                        })
295                        .collect::<Vec<_>>()
296                };
297                let new_val = (link.compute)(&values);
298                if self.set_value_inner(&link.output, new_val).await {
299                    queue.push(link.output.clone());
300                }
301            }
302        }
303    }
304
305    /// List all PV names.
306    pub async fn pv_names(&self) -> Vec<String> {
307        let pvs = self.pvs.read().await;
308        pvs.keys().cloned().collect()
309    }
310}
311
312impl Source for SimplePvStore {
313    fn claim(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<PvInfo>> + Send + '_>> {
314        let name = name.to_string();
315        Box::pin(async move {
316            let pvs = self.pvs.read().await;
317            let entry = pvs.get(&name)?;
318            let descriptor = descriptor_for_payload(&entry.record.to_ntpayload());
319            Some(PvInfo {
320                descriptor,
321                writable: entry.record.writable(),
322            })
323        })
324    }
325
326    fn get(&self, name: &str) -> Pin<Box<dyn Future<Output = Option<NtPayload>> + Send + '_>> {
327        let name = name.to_string();
328        Box::pin(async move {
329            let pvs = self.pvs.read().await;
330            pvs.get(&name).map(|e| e.record.to_ntpayload())
331        })
332    }
333
334    fn put(
335        &self,
336        name: &str,
337        value: &DecodedValue,
338    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, NtPayload)>, String>> + Send + '_>> {
339        let name = name.to_string();
340        let value = value.clone();
341        Box::pin(async move {
342            let result = {
343                let mut pvs = self.pvs.write().await;
344                let entry = pvs
345                    .get_mut(&name)
346                    .ok_or_else(|| format!("PV '{}' not found", name))?;
347
348                if !entry.record.writable() {
349                    return Err(format!("PV '{}' is not writable", name));
350                }
351
352                let prev_severity = entry.record.to_ntscalar().alarm_severity;
353                let changed = apply_put_to_record(&mut entry.record, &value, self.compute_alarms);
354                if !changed {
355                    return Ok(vec![]);
356                }
357
358                if should_post_update(entry, prev_severity) {
359                    let payload = entry.record.to_ntpayload();
360                    entry
361                        .subscribers
362                        .retain(|tx| tx.try_send(payload.clone()).is_ok());
363                    Some((name.clone(), payload))
364                } else {
365                    // Within the MDEL monitor deadband — value applied, no post.
366                    None
367                }
368            }; // pvs lock dropped
369
370            // Fire on_put callback (non-blocking).
371            if let Some(cb) = self.on_put.get(&name) {
372                let cb = cb.clone();
373                let n = name.clone();
374                let v = value.clone();
375                tokio::spawn(async move { cb(&n, &v) });
376            }
377
378            // Propagate linked PV updates.
379            self.evaluate_links(&name).await;
380
381            Ok(result.into_iter().collect())
382        })
383    }
384
385    fn subscribe(
386        &self,
387        name: &str,
388    ) -> Pin<Box<dyn Future<Output = Option<mpsc::Receiver<NtPayload>>> + Send + '_>> {
389        let name = name.to_string();
390        Box::pin(async move {
391            let mut pvs = self.pvs.write().await;
392            let entry = pvs.get_mut(&name)?;
393            let (tx, rx) = mpsc::channel(64);
394            entry.subscribers.push(tx);
395            Some(rx)
396        })
397    }
398
399    fn names(&self) -> Pin<Box<dyn Future<Output = Vec<String>> + Send + '_>> {
400        Box::pin(async move {
401            let pvs = self.pvs.read().await;
402            pvs.keys().cloned().collect()
403        })
404    }
405}
406
407// ── Helpers ──────────────────────────────────────────────────────────────
408
409/// Numeric view of a scalar value, for deadband arithmetic.
410fn scalar_as_f64(v: &ScalarValue) -> Option<f64> {
411    Some(match v {
412        ScalarValue::I8(x) => *x as f64,
413        ScalarValue::I16(x) => *x as f64,
414        ScalarValue::I32(x) => *x as f64,
415        ScalarValue::I64(x) => *x as f64,
416        ScalarValue::U8(x) => *x as f64,
417        ScalarValue::U16(x) => *x as f64,
418        ScalarValue::U32(x) => *x as f64,
419        ScalarValue::U64(x) => *x as f64,
420        ScalarValue::F32(x) => *x as f64,
421        ScalarValue::F64(x) => *x,
422        ScalarValue::Bool(_) | ScalarValue::Str(_) => return None,
423    })
424}
425
426/// Initial MDEL deadband reference: the record's starting value (EPICS
427/// initialises MLST from the initial VAL, so the first small change is
428/// already subject to the deadband).
429fn initial_posted(record: &RecordInstance) -> Option<f64> {
430    match record.to_ntpayload() {
431        NtPayload::Scalar(nt) => scalar_as_f64(&nt.value),
432        _ => None,
433    }
434}
435
436/// MDEL monitor-deadband gate, called after a record changed.
437///
438/// Returns `true` when the update must be posted to subscribers/monitors
439/// (and records it as the new deadband reference point). An update is
440/// suppressed only when the record is a numeric scalar with MDEL > 0, the
441/// alarm severity did not change, and the value moved less than MDEL from
442/// the last *posted* value — EPICS monitor-deadband semantics.
443fn should_post_update(entry: &mut PvEntry, prev_severity: i32) -> bool {
444    let new_f = match entry.record.to_ntpayload() {
445        NtPayload::Scalar(nt) => match scalar_as_f64(&nt.value) {
446            Some(f) => f,
447            None => return true,
448        },
449        _ => return true,
450    };
451    let mdel = crate::record_fields::mdel_of(&entry.record);
452    let severity_changed = entry.record.to_ntscalar().alarm_severity != prev_severity;
453    let within_deadband = mdel > 0.0
454        && !severity_changed
455        && entry
456            .last_posted
457            .is_some_and(|last| (new_f - last).abs() < mdel);
458    if within_deadband {
459        return false;
460    }
461    entry.last_posted = Some(new_f);
462    true
463}
464
465/// Apply a decoded PUT value to a RecordInstance, returning whether it changed.
466fn apply_put_to_record(
467    record: &mut RecordInstance,
468    value: &DecodedValue,
469    compute_alarms: bool,
470) -> bool {
471    let fields = match value {
472        DecodedValue::Structure(f) => f,
473        other => {
474            // Bare scalar — wrap as value field.
475            return apply_put_to_record(
476                record,
477                &DecodedValue::Structure(vec![("value".to_string(), other.clone())]),
478                compute_alarms,
479            );
480        }
481    };
482
483    let mut changed = false;
484
485    match &mut record.data {
486        RecordData::Ai { nt, .. }
487        | RecordData::Ao { nt, .. }
488        | RecordData::Bi { nt, .. }
489        | RecordData::Bo { nt, .. }
490        | RecordData::StringIn { nt, .. }
491        | RecordData::StringOut { nt, .. } => {
492            for (name, val) in fields {
493                match name.as_str() {
494                    "value" => {
495                        changed |= apply_value_update(nt, val, compute_alarms);
496                    }
497                    "alarm" => {
498                        changed |= apply_alarm_update(nt, val);
499                    }
500                    "display" => {
501                        changed |= apply_display_update(nt, val);
502                    }
503                    "control" => {
504                        changed |= apply_control_update(nt, val);
505                    }
506                    _ => {}
507                }
508            }
509        }
510        RecordData::Waveform { nt, nord, .. }
511        | RecordData::Aai { nt, nord, .. }
512        | RecordData::Aao { nt, nord, .. }
513        | RecordData::SubArray { nt, nord, .. } => {
514            changed = apply_scalar_array_put(nt, nord, value);
515        }
516        RecordData::NtTable { .. } | RecordData::NtNdArray { .. } => {
517            // Table/NdArray PUT not supported via high-level API yet.
518            debug!("PUT to NtTable/NtNdArray not yet supported in SimplePvStore");
519        }
520        RecordData::NtEnum { nt, .. } => {
521            // Accept index updates for NtEnum PVs.
522            for (name, val) in fields {
523                if name == "value" {
524                    let idx = match val {
525                        DecodedValue::Int32(v) => Some(*v),
526                        DecodedValue::Int64(v) => Some(*v as i32),
527                        DecodedValue::Int16(v) => Some(*v as i32),
528                        DecodedValue::Int8(v) => Some(*v as i32),
529                        DecodedValue::Float64(v) => Some(*v as i32),
530                        _ => None,
531                    };
532                    if let Some(idx) = idx {
533                        if nt.index != idx {
534                            nt.index = idx;
535                            changed = true;
536                        }
537                    }
538                }
539            }
540        }
541        RecordData::Generic { .. } => {
542            debug!("PUT to Generic not yet supported in SimplePvStore");
543        }
544    }
545
546    changed
547}
548
549// ── NtPayload → StructureDesc ────────────────────────────────────────────
550
551pub fn descriptor_for_payload(payload: &NtPayload) -> StructureDesc {
552    match payload {
553        NtPayload::Scalar(nt) => nt_scalar_desc(&nt.value),
554        NtPayload::ScalarArray(arr) => nt_scalar_array_desc(&arr.value),
555        _ => StructureDesc::new(),
556    }
557}
558
559fn value_type_code(sv: &ScalarValue) -> TypeCode {
560    match sv {
561        ScalarValue::Bool(_) => TypeCode::Boolean,
562        ScalarValue::I8(_) => TypeCode::Int8,
563        ScalarValue::I16(_) => TypeCode::Int16,
564        ScalarValue::I32(_) => TypeCode::Int32,
565        ScalarValue::I64(_) => TypeCode::Int64,
566        ScalarValue::U8(_) => TypeCode::UInt8,
567        ScalarValue::U16(_) => TypeCode::UInt16,
568        ScalarValue::U32(_) => TypeCode::UInt32,
569        ScalarValue::U64(_) => TypeCode::UInt64,
570        ScalarValue::F32(_) => TypeCode::Float32,
571        ScalarValue::F64(_) => TypeCode::Float64,
572        ScalarValue::Str(_) => TypeCode::String,
573    }
574}
575
576fn array_type_code(sav: &ScalarArrayValue) -> TypeCode {
577    match sav {
578        ScalarArrayValue::Bool(_) => TypeCode::Boolean,
579        ScalarArrayValue::I8(_) => TypeCode::Int8,
580        ScalarArrayValue::I16(_) => TypeCode::Int16,
581        ScalarArrayValue::I32(_) => TypeCode::Int32,
582        ScalarArrayValue::I64(_) => TypeCode::Int64,
583        ScalarArrayValue::U8(_) => TypeCode::UInt8,
584        ScalarArrayValue::U16(_) => TypeCode::UInt16,
585        ScalarArrayValue::U32(_) => TypeCode::UInt32,
586        ScalarArrayValue::U64(_) => TypeCode::UInt64,
587        ScalarArrayValue::F32(_) => TypeCode::Float32,
588        ScalarArrayValue::F64(_) => TypeCode::Float64,
589        ScalarArrayValue::Str(_) => TypeCode::String,
590    }
591}
592
593fn nt_scalar_desc(sv: &ScalarValue) -> StructureDesc {
594    let tc = value_type_code(sv);
595    StructureDesc {
596        struct_id: Some("epics:nt/NTScalar:1.0".to_string()),
597        fields: vec![
598            FieldDesc {
599                name: "value".to_string(),
600                field_type: FieldType::Scalar(tc),
601            },
602            FieldDesc {
603                name: "alarm".to_string(),
604                field_type: FieldType::Structure(alarm_desc()),
605            },
606            FieldDesc {
607                name: "timeStamp".to_string(),
608                field_type: FieldType::Structure(timestamp_desc()),
609            },
610            FieldDesc {
611                name: "display".to_string(),
612                field_type: FieldType::Structure(display_desc()),
613            },
614            FieldDesc {
615                name: "control".to_string(),
616                field_type: FieldType::Structure(control_desc()),
617            },
618            FieldDesc {
619                name: "valueAlarm".to_string(),
620                field_type: FieldType::Structure(value_alarm_desc()),
621            },
622        ],
623    }
624}
625
626fn nt_scalar_array_desc(sav: &ScalarArrayValue) -> StructureDesc {
627    let tc = array_type_code(sav);
628    StructureDesc {
629        struct_id: Some("epics:nt/NTScalarArray:1.0".to_string()),
630        fields: vec![
631            FieldDesc {
632                name: "value".to_string(),
633                field_type: FieldType::ScalarArray(tc),
634            },
635            FieldDesc {
636                name: "alarm".to_string(),
637                field_type: FieldType::Structure(alarm_desc()),
638            },
639            FieldDesc {
640                name: "timeStamp".to_string(),
641                field_type: FieldType::Structure(timestamp_desc()),
642            },
643            FieldDesc {
644                name: "display".to_string(),
645                field_type: FieldType::Structure(display_desc()),
646            },
647            FieldDesc {
648                name: "control".to_string(),
649                field_type: FieldType::Structure(control_desc()),
650            },
651        ],
652    }
653}
654
655fn alarm_desc() -> StructureDesc {
656    StructureDesc {
657        struct_id: Some("alarm_t".to_string()),
658        fields: vec![
659            FieldDesc {
660                name: "severity".to_string(),
661                field_type: FieldType::Scalar(TypeCode::Int32),
662            },
663            FieldDesc {
664                name: "status".to_string(),
665                field_type: FieldType::Scalar(TypeCode::Int32),
666            },
667            FieldDesc {
668                name: "message".to_string(),
669                field_type: FieldType::String,
670            },
671        ],
672    }
673}
674
675fn timestamp_desc() -> StructureDesc {
676    StructureDesc {
677        struct_id: Some("time_t".to_string()),
678        fields: vec![
679            FieldDesc {
680                name: "secondsPastEpoch".to_string(),
681                field_type: FieldType::Scalar(TypeCode::Int64),
682            },
683            FieldDesc {
684                name: "nanoseconds".to_string(),
685                field_type: FieldType::Scalar(TypeCode::Int32),
686            },
687            FieldDesc {
688                name: "userTag".to_string(),
689                field_type: FieldType::Scalar(TypeCode::Int32),
690            },
691        ],
692    }
693}
694
695fn display_desc() -> StructureDesc {
696    StructureDesc {
697        struct_id: Some("display_t".to_string()),
698        fields: vec![
699            FieldDesc {
700                name: "limitLow".to_string(),
701                field_type: FieldType::Scalar(TypeCode::Float64),
702            },
703            FieldDesc {
704                name: "limitHigh".to_string(),
705                field_type: FieldType::Scalar(TypeCode::Float64),
706            },
707            FieldDesc {
708                name: "description".to_string(),
709                field_type: FieldType::String,
710            },
711            FieldDesc {
712                name: "units".to_string(),
713                field_type: FieldType::String,
714            },
715            FieldDesc {
716                name: "precision".to_string(),
717                field_type: FieldType::Scalar(TypeCode::Int32),
718            },
719            FieldDesc {
720                name: "form".to_string(),
721                field_type: FieldType::Structure(StructureDesc {
722                    struct_id: Some("enum_t".to_string()),
723                    fields: vec![
724                        FieldDesc {
725                            name: "index".to_string(),
726                            field_type: FieldType::Scalar(TypeCode::Int32),
727                        },
728                        FieldDesc {
729                            name: "choices".to_string(),
730                            field_type: FieldType::StringArray,
731                        },
732                    ],
733                }),
734            },
735        ],
736    }
737}
738
739fn control_desc() -> StructureDesc {
740    StructureDesc {
741        struct_id: Some("control_t".to_string()),
742        fields: vec![
743            FieldDesc {
744                name: "limitLow".to_string(),
745                field_type: FieldType::Scalar(TypeCode::Float64),
746            },
747            FieldDesc {
748                name: "limitHigh".to_string(),
749                field_type: FieldType::Scalar(TypeCode::Float64),
750            },
751            FieldDesc {
752                name: "minStep".to_string(),
753                field_type: FieldType::Scalar(TypeCode::Float64),
754            },
755        ],
756    }
757}
758
759fn value_alarm_desc() -> StructureDesc {
760    StructureDesc {
761        struct_id: Some("valueAlarm_t".to_string()),
762        fields: vec![
763            FieldDesc {
764                name: "active".to_string(),
765                field_type: FieldType::Scalar(TypeCode::Boolean),
766            },
767            FieldDesc {
768                name: "lowAlarmLimit".to_string(),
769                field_type: FieldType::Scalar(TypeCode::Float64),
770            },
771            FieldDesc {
772                name: "lowWarningLimit".to_string(),
773                field_type: FieldType::Scalar(TypeCode::Float64),
774            },
775            FieldDesc {
776                name: "highWarningLimit".to_string(),
777                field_type: FieldType::Scalar(TypeCode::Float64),
778            },
779            FieldDesc {
780                name: "highAlarmLimit".to_string(),
781                field_type: FieldType::Scalar(TypeCode::Float64),
782            },
783            FieldDesc {
784                name: "lowAlarmSeverity".to_string(),
785                field_type: FieldType::Scalar(TypeCode::Int32),
786            },
787            FieldDesc {
788                name: "lowWarningSeverity".to_string(),
789                field_type: FieldType::Scalar(TypeCode::Int32),
790            },
791            FieldDesc {
792                name: "highWarningSeverity".to_string(),
793                field_type: FieldType::Scalar(TypeCode::Int32),
794            },
795            FieldDesc {
796                name: "highAlarmSeverity".to_string(),
797                field_type: FieldType::Scalar(TypeCode::Int32),
798            },
799            FieldDesc {
800                name: "hysteresis".to_string(),
801                field_type: FieldType::Scalar(TypeCode::UInt8),
802            },
803        ],
804    }
805}
806
807#[cfg(test)]
808mod tests {
809    use super::*;
810    use crate::types::{DbCommonState, RecordType};
811    use spvirit_types::{
812        NdCodec, NdDimension, NtNdArray, NtPayload, NtScalar, NtScalarArray, NtTable,
813        NtTableColumn, ScalarArrayValue, ScalarValue,
814    };
815
816    fn make_ai(name: &str, val: f64) -> RecordInstance {
817        RecordInstance {
818            name: name.to_string(),
819            record_type: RecordType::Ai,
820            common: DbCommonState::default(),
821            data: RecordData::Ai {
822                nt: NtScalar::from_value(ScalarValue::F64(val)),
823                inp: None,
824                siml: None,
825                siol: None,
826                simm: false,
827            },
828            raw_fields: HashMap::new(),
829        }
830    }
831
832    fn make_ao(name: &str, val: f64) -> RecordInstance {
833        RecordInstance {
834            name: name.to_string(),
835            record_type: RecordType::Ao,
836            common: DbCommonState::default(),
837            data: RecordData::Ao {
838                nt: NtScalar::from_value(ScalarValue::F64(val)),
839                out: None,
840                dol: None,
841                omsl: crate::types::OutputMode::Supervisory,
842                drvl: None,
843                drvh: None,
844                oroc: None,
845                siml: None,
846                siol: None,
847                simm: false,
848            },
849            raw_fields: HashMap::new(),
850        }
851    }
852
853    #[tokio::test]
854    async fn mdel_deadband_suppresses_small_monitor_updates() {
855        let recs = crate::db::parse_db(
856            r#"
857record(ao, "DB:AO") {
858    field(VAL, "0.0")
859    field(MDEL, "0.5")
860}"#,
861        )
862        .expect("parse");
863        let store = SimplePvStore::new(recs, HashMap::new(), Vec::new(), false);
864        let mut rx = Source::subscribe(&store, "DB:AO")
865            .await
866            .expect("subscribed");
867
868        // |Δ| = 0.2 < MDEL 0.5 → value updates but no monitor post.
869        assert!(store.set_value("DB:AO", ScalarValue::F64(0.2)).await);
870        // |Δ| = 0.9 ≥ MDEL 0.5 → posted.
871        assert!(store.set_value("DB:AO", ScalarValue::F64(0.9)).await);
872
873        match rx.recv().await.expect("posted update") {
874            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(0.9)),
875            other => panic!("expected scalar, got {other:?}"),
876        }
877        // The suppressed 0.2 update must not be queued behind it.
878        assert!(rx.try_recv().is_err());
879        // GETs always see the latest value regardless of the deadband.
880        assert_eq!(store.get_value("DB:AO").await, Some(ScalarValue::F64(0.9)));
881    }
882
883    fn make_waveform(name: &str, value: ScalarArrayValue) -> RecordInstance {
884        let nelm = value.len();
885        RecordInstance {
886            name: name.to_string(),
887            record_type: RecordType::Waveform,
888            common: DbCommonState::default(),
889            data: RecordData::Waveform {
890                nt: NtScalarArray::from_value(value),
891                inp: None,
892                ftvl: "DOUBLE".to_string(),
893                nelm,
894                nord: nelm,
895            },
896            raw_fields: HashMap::new(),
897        }
898    }
899
900    fn make_nt_table(name: &str) -> RecordInstance {
901        RecordInstance {
902            name: name.to_string(),
903            record_type: RecordType::NtTable,
904            common: DbCommonState::default(),
905            data: RecordData::NtTable {
906                nt: NtTable {
907                    labels: vec!["X".to_string(), "Y".to_string()],
908                    columns: vec![
909                        NtTableColumn {
910                            name: "x".to_string(),
911                            values: ScalarArrayValue::F64(vec![1.0, 2.0]),
912                        },
913                        NtTableColumn {
914                            name: "y".to_string(),
915                            values: ScalarArrayValue::F64(vec![10.0, 20.0]),
916                        },
917                    ],
918                    descriptor: Some("table".to_string()),
919                    alarm: None,
920                    time_stamp: None,
921                },
922                inp: None,
923                out: None,
924                omsl: crate::types::OutputMode::Supervisory,
925            },
926            raw_fields: HashMap::new(),
927        }
928    }
929
930    fn make_nt_ndarray(name: &str) -> RecordInstance {
931        RecordInstance {
932            name: name.to_string(),
933            record_type: RecordType::NtNdArray,
934            common: DbCommonState::default(),
935            data: RecordData::NtNdArray {
936                nt: NtNdArray {
937                    value: ScalarArrayValue::U8(vec![0; 4]),
938                    codec: NdCodec {
939                        name: "none".to_string(),
940                        parameters: HashMap::new(),
941                    },
942                    compressed_size: 4,
943                    uncompressed_size: 4,
944                    dimension: vec![NdDimension {
945                        size: 2,
946                        offset: 0,
947                        full_size: 2,
948                        binning: 1,
949                        reverse: false,
950                    }],
951                    unique_id: 1,
952                    data_time_stamp: Default::default(),
953                    attribute: vec![],
954                    descriptor: Some("ndarray".to_string()),
955                    alarm: None,
956                    time_stamp: None,
957                    display: None,
958                },
959                inp: None,
960                out: None,
961                omsl: crate::types::OutputMode::Supervisory,
962            },
963            raw_fields: HashMap::new(),
964        }
965    }
966
967    #[tokio::test]
968    async fn has_pv_returns_true_for_existing() {
969        let mut records = HashMap::new();
970        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
971        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
972        assert!(store.claim("TEST:AI").await.is_some());
973        assert!(store.claim("MISSING").await.is_none());
974    }
975
976    #[tokio::test]
977    async fn get_snapshot_returns_payload() {
978        let mut records = HashMap::new();
979        records.insert("TEST:AI".into(), make_ai("TEST:AI", 42.0));
980        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
981        let snap = store.get("TEST:AI").await.unwrap();
982        match snap {
983            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(42.0)),
984            _ => panic!("expected scalar"),
985        }
986    }
987
988    #[tokio::test]
989    async fn put_value_updates_writable_record() {
990        let mut records = HashMap::new();
991        records.insert("TEST:AO".into(), make_ao("TEST:AO", 0.0));
992        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
993
994        let val = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(99.5))]);
995        let result = store.put("TEST:AO", &val).await.unwrap();
996        assert_eq!(result.len(), 1);
997        assert_eq!(result[0].0, "TEST:AO");
998
999        let snap = store.get("TEST:AO").await.unwrap();
1000        match snap {
1001            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(99.5)),
1002            _ => panic!("expected scalar"),
1003        }
1004    }
1005
1006    #[tokio::test]
1007    async fn put_value_rejects_readonly() {
1008        let mut records = HashMap::new();
1009        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1010        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1011
1012        let val = DecodedValue::Float64(5.0);
1013        let err = store.put("TEST:AI", &val).await.unwrap_err();
1014        assert!(err.contains("not writable"));
1015    }
1016
1017    #[tokio::test]
1018    async fn set_value_bypasses_writable_check() {
1019        let mut records = HashMap::new();
1020        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1021        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1022        assert!(store.set_value("TEST:AI", ScalarValue::F64(10.0)).await);
1023        let val = store.get_value("TEST:AI").await.unwrap();
1024        assert_eq!(val, ScalarValue::F64(10.0));
1025    }
1026
1027    #[tokio::test]
1028    async fn set_array_value_updates_all_scalar_array_types() {
1029        let cases: Vec<ScalarArrayValue> = vec![
1030            ScalarArrayValue::Bool(vec![false, true]),
1031            ScalarArrayValue::I8(vec![1, 2]),
1032            ScalarArrayValue::I16(vec![1, 2]),
1033            ScalarArrayValue::I32(vec![1, 2]),
1034            ScalarArrayValue::I64(vec![1, 2]),
1035            ScalarArrayValue::U8(vec![1, 2]),
1036            ScalarArrayValue::U16(vec![1, 2]),
1037            ScalarArrayValue::U32(vec![1, 2]),
1038            ScalarArrayValue::U64(vec![1, 2]),
1039            ScalarArrayValue::F32(vec![1.0, 2.0]),
1040            ScalarArrayValue::F64(vec![1.0, 2.0]),
1041            ScalarArrayValue::Str(vec!["a".to_string(), "b".to_string()]),
1042        ];
1043
1044        for (idx, updated) in cases.into_iter().enumerate() {
1045            let pv = format!("TEST:WF:{idx}");
1046            let mut records = HashMap::new();
1047            records.insert(pv.clone(), make_waveform(&pv, updated.clone()));
1048            let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1049
1050            assert!(!store.set_array_value(&pv, updated.clone()).await);
1051
1052            let second = match updated {
1053                ScalarArrayValue::Bool(_) => ScalarArrayValue::Bool(vec![true, false]),
1054                ScalarArrayValue::I8(_) => ScalarArrayValue::I8(vec![3, 4]),
1055                ScalarArrayValue::I16(_) => ScalarArrayValue::I16(vec![3, 4]),
1056                ScalarArrayValue::I32(_) => ScalarArrayValue::I32(vec![3, 4]),
1057                ScalarArrayValue::I64(_) => ScalarArrayValue::I64(vec![3, 4]),
1058                ScalarArrayValue::U8(_) => ScalarArrayValue::U8(vec![3, 4]),
1059                ScalarArrayValue::U16(_) => ScalarArrayValue::U16(vec![3, 4]),
1060                ScalarArrayValue::U32(_) => ScalarArrayValue::U32(vec![3, 4]),
1061                ScalarArrayValue::U64(_) => ScalarArrayValue::U64(vec![3, 4]),
1062                ScalarArrayValue::F32(_) => ScalarArrayValue::F32(vec![3.0, 4.0]),
1063                ScalarArrayValue::F64(_) => ScalarArrayValue::F64(vec![3.0, 4.0]),
1064                ScalarArrayValue::Str(_) => {
1065                    ScalarArrayValue::Str(vec!["x".to_string(), "y".to_string()])
1066                }
1067            };
1068
1069            assert!(store.set_array_value(&pv, second.clone()).await);
1070            let snap = store.get(&pv).await.unwrap();
1071            match snap {
1072                NtPayload::ScalarArray(nt) => assert_eq!(nt.value, second),
1073                _ => panic!("expected scalar array"),
1074            }
1075        }
1076    }
1077
1078    #[tokio::test]
1079    async fn get_nt_returns_full_payload() {
1080        let mut records = HashMap::new();
1081        records.insert("TEST:AI".into(), make_ai("TEST:AI", 12.5));
1082        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1083
1084        let nt = store.get_nt("TEST:AI").await.unwrap();
1085        match nt {
1086            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(12.5)),
1087            _ => panic!("expected scalar payload"),
1088        }
1089    }
1090
1091    #[tokio::test]
1092    async fn put_nt_updates_scalar_array_table_and_ndarray() {
1093        let mut records = HashMap::new();
1094        records.insert("TEST:AI".into(), make_ai("TEST:AI", 1.0));
1095        records.insert(
1096            "TEST:WF".into(),
1097            make_waveform("TEST:WF", ScalarArrayValue::F64(vec![0.0, 0.0])),
1098        );
1099        records.insert("TEST:TBL".into(), make_nt_table("TEST:TBL"));
1100        records.insert("TEST:NDA".into(), make_nt_ndarray("TEST:NDA"));
1101        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1102
1103        assert!(
1104            store
1105                .put_nt(
1106                    "TEST:AI",
1107                    NtPayload::Scalar(NtScalar::from_value(ScalarValue::F64(5.0))),
1108                )
1109                .await
1110        );
1111        assert!(
1112            store
1113                .put_nt(
1114                    "TEST:WF",
1115                    NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::F64(vec![
1116                        3.0, 4.0
1117                    ],))),
1118                )
1119                .await
1120        );
1121
1122        let table = NtTable {
1123            labels: vec!["X".to_string(), "Y".to_string()],
1124            columns: vec![
1125                NtTableColumn {
1126                    name: "x".to_string(),
1127                    values: ScalarArrayValue::F64(vec![2.0, 3.0]),
1128                },
1129                NtTableColumn {
1130                    name: "y".to_string(),
1131                    values: ScalarArrayValue::F64(vec![20.0, 30.0]),
1132                },
1133            ],
1134            descriptor: Some("updated table".to_string()),
1135            alarm: None,
1136            time_stamp: None,
1137        };
1138        assert!(
1139            store
1140                .put_nt("TEST:TBL", NtPayload::Table(table.clone()))
1141                .await
1142        );
1143
1144        let ndarray = NtNdArray {
1145            value: ScalarArrayValue::U8(vec![1, 2, 3, 4]),
1146            codec: NdCodec {
1147                name: "none".to_string(),
1148                parameters: HashMap::new(),
1149            },
1150            compressed_size: 4,
1151            uncompressed_size: 4,
1152            dimension: vec![NdDimension {
1153                size: 4,
1154                offset: 0,
1155                full_size: 4,
1156                binning: 1,
1157                reverse: false,
1158            }],
1159            unique_id: 2,
1160            data_time_stamp: Default::default(),
1161            attribute: vec![],
1162            descriptor: Some("updated ndarray".to_string()),
1163            alarm: None,
1164            time_stamp: None,
1165            display: None,
1166        };
1167        assert!(
1168            store
1169                .put_nt("TEST:NDA", NtPayload::NdArray(ndarray.clone()))
1170                .await
1171        );
1172
1173        assert!(
1174            !store
1175                .put_nt(
1176                    "TEST:AI",
1177                    NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::F64(vec![
1178                        1.0
1179                    ]))),
1180                )
1181                .await
1182        );
1183
1184        match store.get_nt("TEST:TBL").await.unwrap() {
1185            NtPayload::Table(nt) => assert_eq!(nt, table),
1186            _ => panic!("expected table payload"),
1187        }
1188        match store.get_nt("TEST:NDA").await.unwrap() {
1189            NtPayload::NdArray(nt) => assert_eq!(nt, ndarray),
1190            _ => panic!("expected ndarray payload"),
1191        }
1192    }
1193
1194    #[tokio::test]
1195    async fn descriptor_matches_value_type() {
1196        let mut records = HashMap::new();
1197        records.insert("TEST:AI".into(), make_ai("TEST:AI", 0.0));
1198        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1199        let info = store.claim("TEST:AI").await.unwrap();
1200        assert_eq!(
1201            info.descriptor.struct_id.as_deref(),
1202            Some("epics:nt/NTScalar:1.0")
1203        );
1204        let desc = info.descriptor;
1205        let value_field = desc.field("value").unwrap();
1206        assert!(matches!(
1207            value_field.field_type,
1208            FieldType::Scalar(TypeCode::Float64)
1209        ));
1210    }
1211
1212    #[tokio::test]
1213    async fn subscribe_receives_updates() {
1214        let mut records = HashMap::new();
1215        records.insert("TEST:AO".into(), make_ao("TEST:AO", 0.0));
1216        let store = SimplePvStore::new(records, HashMap::new(), vec![], false);
1217
1218        let mut rx = Source::subscribe(&store, "TEST:AO").await.unwrap();
1219
1220        let val = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(7.7))]);
1221        store.put("TEST:AO", &val).await.unwrap();
1222
1223        let update = rx.recv().await.unwrap();
1224        match update {
1225            NtPayload::Scalar(nt) => assert_eq!(nt.value, ScalarValue::F64(7.7)),
1226            _ => panic!("expected scalar"),
1227        }
1228    }
1229
1230    #[tokio::test]
1231    async fn on_put_callback_is_invoked() {
1232        use std::sync::atomic::{AtomicBool, Ordering};
1233
1234        let called = Arc::new(AtomicBool::new(false));
1235        let called2 = called.clone();
1236
1237        let mut records = HashMap::new();
1238        records.insert("CB:AO".into(), make_ao("CB:AO", 0.0));
1239
1240        let mut on_put = HashMap::new();
1241        let cb: OnPutCallback = Arc::new(move |_name, _val| {
1242            called2.store(true, Ordering::SeqCst);
1243        });
1244        on_put.insert("CB:AO".into(), cb);
1245
1246        let store = SimplePvStore::new(records, on_put, vec![], false);
1247        let val = DecodedValue::Structure(vec![("value".to_string(), DecodedValue::Float64(1.0))]);
1248        store.put("CB:AO", &val).await.unwrap();
1249
1250        // Give the spawned task time to run.
1251        tokio::task::yield_now().await;
1252        tokio::task::yield_now().await;
1253
1254        assert!(called.load(Ordering::SeqCst));
1255    }
1256}
spvirit_server/simple_store.rs

spvirit_server/
simple_store.rs
