reader_invalidation_recovery/
reader_invalidation_recovery.rs

1// A reader is invalidated when its signal's descriptor changes to a sample
2// type the reader cannot convert to its configured read type.  The reader
3// needs to be recovered by creating a new one via `StreamReader::from_existing`.
4
5use opendaq::{Channel, Instance, Sample, StreamReader};
6
7/// Read four averages with their domain values and print them (retrying while
8/// the stream warms up).
9fn show_averages<D: Sample + std::fmt::Display>(
10    reader: &StreamReader<f64, D>,
11) -> opendaq::Result<()> {
12    for _ in 0..50 {
13        let (averages, ticks) = reader.read_with_domain(4, 1000)?;
14        if !averages.is_empty() {
15            for (average, tick) in averages.iter().zip(&ticks) {
16                println!("  tick {tick}  ->  avg {average:.3}");
17            }
18            return Ok(());
19        }
20    }
21    panic!("The statistics stream produced no averages.");
22}
23
24fn main() -> opendaq::Result<()> {
25    let instance = Instance::new()?;
26    instance
27        .add_device("daqref://device0")?
28        .expect("reference device");
29    let channel = instance
30        .find_component("Dev/RefDev0/IO/AI/RefCh0")?
31        .expect("reference channel")
32        .cast::<Channel>()?;
33    channel.set_property_value("Waveform", 0)?;
34    channel.set_property_value("Amplitude", 0.0)?;
35    channel.set_property_value("DC", 2.0)?;
36    channel.set_property_value("NoiseAmplitude", 0.0)?;
37
38    // Average the channel in blocks of 10 samples.  DomainSignalType starts at
39    // 0 = Implicit: the output domain is int64 ticks under a linear rule.
40    let statistics_fb = instance
41        .add_function_block("RefFBModuleStatistics")?
42        .expect("statistics function block");
43    statistics_fb.input_ports()?[0].connect(&channel.signals()?[0])?;
44    let avg_signal = statistics_fb.signals()?[0].clone();
45
46    // Phase 1: read averages with the domain ticks as doubles -- fine while
47    // the domain is int64, which converts to float64.
48    let reader = StreamReader::<f64, f64>::new(&avg_signal)?;
49    println!("implicit int64 domain, read as float64:");
50    show_averages(&reader)?;
51
52    // Phase 2: switch the output domain to 2 = ExplicitRange -- each average's
53    // domain value becomes the RangeInt64 tick range of the block it covers,
54    // and RangeInt64 has no conversion to float64.  The change reaches the
55    // reader as a descriptor-changed event in its packet queue; `read` hands
56    // over the samples queued before the change, then hits the event and
57    // fails with a reader-invalidated error.
58    statistics_fb.set_property_value("DomainSignalType", 2)?;
59    let mut recovered = None;
60    for _ in 0..50 {
61        match reader.read_with_domain(100, 200) {
62            Ok(_) => continue,
63            Err(err) if err.is_reader_invalidated() => {
64                println!("{err}");
65                // Recover: a new reader with read types matching the new
66                // descriptor, inheriting the invalidated reader's connection
67                // and unread packets.
68                recovered = Some(StreamReader::<f64, i64>::from_existing(&reader)?);
69                break;
70            }
71            Err(err) => return Err(err),
72        }
73    }
74    let recovered = recovered.expect("expected the domain change to invalidate the reader");
75
76    println!("explicit RangeInt64 domain, read as int64 range starts:");
77    show_averages(&recovered)?;
78    Ok(())
79}
reader_invalidation_recovery/reader_invalidation_recovery.rs

reader_invalidation_recovery/
reader_invalidation_recovery.rs
