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

spvirit_server/
simple_store.rs
