mockarchiver/
lib.rs

1//! A mock EPICS Archiver Appliance for tests, CI and demos.
2//!
3//! It serves deterministic synthetic history for a fixed set of PVs (see
4//! [`signals`]) with the real appliance's wire format and quirks:
5//!
6//! - `GET /retrieval/data/getData.json?pv=..&from=..&to=..` — raw (with the
7//!   sample before `from`), `optimized_N` (raw when few samples, else
8//!   `[mean, std, min, max, count]` bins at bin centres, filled), `mean_N`,
9//!   `min_N`, `max_N`, `std_N`, `variance_N`, `median_N`, `count_N`,
10//!   `firstSample_N`, `lastSample_N`, `nth_N`, `ncount` and the
11//!   `<op>Sample_N` no-fill variants; `op(PV,N)` and `pp=` forms too;
12//!   bare `NaN` tokens for `MOCK:NAN`; 404 (HTML body) for unknown PVs;
13//!   400 for a bad time, `to < from` or unbalanced parentheses. `to`
14//!   defaults to now + 1h, `from` to `to` − 1 day. Summaries of strings and
15//!   waveforms return raw data.
16//! - `GET /retrieval/bpl/getMatchingPVs?pv=<glob>|regex=<re>&limit=N` —
17//!   the appliance's naive glob (`*`→`.*`, `?`→`.`), whole-name match,
18//!   sorted; 404 with neither parameter.
19//! - `GET /retrieval/bpl/getMetadata?pv=..` — `DBRType`, `elementCount`,
20//!   `units`, `precision`, ... as strings.
21//! - `GET /retrieval/bpl/getVersion` — `{"version": "Archiver Appliance
22//!   Version <mock version>"}`.
23//! - `GET /mgmt/bpl/getAllPVs?pv=<glob>[,<glob>]&regex=..&limit=N` — unsorted.
24//!
25//! Every endpoint is served on one port (like the single-container
26//! appliance), so `--archiver http://127.0.0.1:PORT` works for the client.
27//! Data begins [`signals::HISTORY_DAYS`] days before the server started
28//! and grows with the clock; large responses are streamed (chunked).
29
30use std::io::{self, Read};
31use std::net::SocketAddr;
32use std::sync::Arc;
33use std::sync::atomic::{AtomicBool, Ordering};
34use std::thread::JoinHandle;
35
36mod bpl;
37mod data;
38pub mod signals;
39
40/// The appliance's retrieval port, the default for the binary.
41pub const DEFAULT_PORT: u16 = 17668;
42
43/// Request-handling threads.
44const WORKERS: usize = 8;
45
46/// A running mock appliance. Serves on background threads until dropped.
47pub struct MockArchiver {
48    addr: SocketAddr,
49    server: Arc<tiny_http::Server>,
50    stop: Arc<AtomicBool>,
51    metadata: Arc<AtomicBool>,
52    threads: Vec<JoinHandle<()>>,
53}
54
55impl MockArchiver {
56    /// Bind `addr` (port 0 picks a free port: use [`MockArchiver::url`])
57    /// and start serving.
58    pub fn start(addr: SocketAddr) -> io::Result<Self> {
59        let server = Arc::new(tiny_http::Server::http(addr).map_err(io::Error::other)?);
60        let addr = server
61            .server_addr()
62            .to_ip()
63            .ok_or_else(|| io::Error::other("not an IP listener"))?;
64        let clock = Arc::new(Clock::started_now());
65        let stop = Arc::new(AtomicBool::new(false));
66        let metadata = Arc::new(AtomicBool::new(true));
67        let threads = (0..WORKERS)
68            .map(|i| {
69                let (server, clock, stop, metadata) = (
70                    server.clone(),
71                    clock.clone(),
72                    stop.clone(),
73                    metadata.clone(),
74                );
75                std::thread::Builder::new()
76                    .name(format!("mockarchiver-{i}"))
77                    .spawn(move || serve(&server, &clock, &stop, &metadata))
78            })
79            .collect::<io::Result<Vec<_>>>()?;
80        Ok(Self {
81            addr,
82            server,
83            stop,
84            metadata,
85            threads,
86        })
87    }
88
89    /// Start on 127.0.0.1 with a free port: the usual test setup.
90    pub fn start_local() -> io::Result<Self> {
91        Self::start(SocketAddr::from(([127, 0, 0, 1], 0)))
92    }
93
94    pub fn addr(&self) -> SocketAddr {
95        self.addr
96    }
97
98    /// `http://127.0.0.1:PORT`, what to pass as `--archiver`.
99    pub fn url(&self) -> String {
100        format!("http://{}", self.addr)
101    }
102
103    /// Whether `bpl/getMetadata` is served (the default) or answers 404,
104    /// like a proxy that passes on only data and search requests.
105    pub fn serve_metadata(&self, on: bool) {
106        self.metadata.store(on, Ordering::SeqCst);
107    }
108
109    /// Serve until the process is killed (the binary).
110    pub fn wait(mut self) {
111        for t in std::mem::take(&mut self.threads) {
112            let _ = t.join();
113        }
114    }
115}
116
117impl Drop for MockArchiver {
118    fn drop(&mut self) {
119        self.stop.store(true, Ordering::SeqCst);
120        for _ in 0..self.threads.len() {
121            self.server.unblock();
122        }
123        for t in std::mem::take(&mut self.threads) {
124            let _ = t.join();
125        }
126    }
127}
128
129fn serve(server: &tiny_http::Server, clock: &Clock, stop: &AtomicBool, metadata: &AtomicBool) {
130    loop {
131        match server.recv() {
132            Ok(rq) => {
133                let reply = if !metadata.load(Ordering::SeqCst)
134                    && rq.url().starts_with("/retrieval/bpl/getMetadata")
135                {
136                    Reply::error(404, "Not Found", "getMetadata is not served here")
137                } else {
138                    route(rq.url(), clock)
139                };
140                log::debug!("{} {} -> {}", rq.method(), rq.url(), reply.status);
141                let _ = rq.respond(reply.into_response());
142            }
143            Err(_) if stop.load(Ordering::SeqCst) => return,
144            Err(e) => log::warn!("mockarchiver: {e}"),
145        }
146    }
147}
148
149/// The server's notion of time: history starts at `origin_ns`, data ends
150/// at the current time.
151#[derive(Clone, Debug)]
152pub(crate) struct Clock {
153    pub(crate) origin_ns: i64,
154    /// A fixed "now" for tests; the wall clock when `None`.
155    fixed_now: Option<i64>,
156}
157
158impl Clock {
159    fn started_now() -> Self {
160        Self::starting_at(wall_ns())
161    }
162
163    /// History from `HISTORY_DAYS` before `start_ns`, aligned to 10
164    /// minutes (a multiple of every PV's period).
165    fn starting_at(start_ns: i64) -> Self {
166        const ALIGN: i64 = 600_000_000_000;
167        let origin = start_ns - signals::HISTORY_DAYS as i64 * 86_400_000_000_000;
168        Self {
169            origin_ns: origin.div_euclid(ALIGN) * ALIGN,
170            fixed_now: None,
171        }
172    }
173
174    /// A clock whose history starts `HISTORY_DAYS` before `now_ns`, which
175    /// stays fixed (tests).
176    #[cfg(test)]
177    pub(crate) fn fixed(now_ns: i64) -> Self {
178        Self {
179            fixed_now: Some(now_ns),
180            ..Self::starting_at(now_ns)
181        }
182    }
183
184    pub(crate) fn now_ns(&self) -> i64 {
185        self.fixed_now.unwrap_or_else(wall_ns)
186    }
187}
188
189fn wall_ns() -> i64 {
190    chrono::Utc::now()
191        .timestamp_nanos_opt()
192        .unwrap_or(i64::MAX / 2)
193}
194
195/// Decoded query parameters, in order.
196pub(crate) struct Params(Vec<(String, String)>);
197
198impl Params {
199    fn parse(query: &str) -> Self {
200        Self(
201            query
202                .split('&')
203                .filter(|p| !p.is_empty())
204                .map(|p| {
205                    let (k, v) = p.split_once('=').unwrap_or((p, ""));
206                    (decode(k), decode(v))
207                })
208                .collect(),
209        )
210    }
211
212    /// The first value of `key`.
213    pub(crate) fn get(&self, key: &str) -> Option<&str> {
214        self.0
215            .iter()
216            .find(|(k, _)| k == key)
217            .map(|(_, v)| v.as_str())
218    }
219}
220
221/// `application/x-www-form-urlencoded` decoding, as the servlet does it.
222fn decode(s: &str) -> String {
223    let b = s.as_bytes();
224    let mut out = Vec::with_capacity(b.len());
225    let mut i = 0;
226    while i < b.len() {
227        match b[i] {
228            b'+' => out.push(b' '),
229            b'%' if i + 2 < b.len() => {
230                let hex = |c: u8| (c as char).to_digit(16);
231                match (hex(b[i + 1]), hex(b[i + 2])) {
232                    (Some(h), Some(l)) => {
233                        out.push((h * 16 + l) as u8);
234                        i += 2;
235                    }
236                    _ => out.push(b'%'),
237                }
238            }
239            c => out.push(c),
240        }
241        i += 1;
242    }
243    String::from_utf8_lossy(&out).into_owned()
244}
245
246/// A response body: small text, or streamed pieces.
247pub(crate) enum Body {
248    Text(String),
249    Stream(Box<dyn Iterator<Item = String> + Send>),
250}
251
252pub(crate) struct Reply {
253    pub(crate) status: u16,
254    content_type: &'static str,
255    body: Body,
256}
257
258impl Reply {
259    pub(crate) fn json(body: String) -> Self {
260        Self {
261            status: 200,
262            content_type: "application/json",
263            body: Body::Text(body),
264        }
265    }
266
267    pub(crate) fn json_stream(body: Box<dyn Iterator<Item = String> + Send>) -> Self {
268        Self {
269            status: 200,
270            content_type: "application/json",
271            body: Body::Stream(body),
272        }
273    }
274
275    /// A Tomcat-style HTML error page.
276    pub(crate) fn error(status: u16, reason: &str, message: &str) -> Self {
277        let message = message
278            .replace('&', "&amp;")
279            .replace('<', "&lt;")
280            .replace('>', "&gt;");
281        Self {
282            status,
283            content_type: "text/html;charset=utf-8",
284            body: Body::Text(format!(
285                "<!doctype html><html lang=\"en\"><head><title>HTTP Status {status} \u{2013} \
286                 {reason}</title><style type=\"text/css\">body {{font-family:Tahoma,Arial,\
287                 sans-serif;}}</style></head><body><h1>HTTP Status {status} \u{2013} {reason}\
288                 </h1><hr class=\"line\" /><p><b>Type</b> Status Report</p><p><b>Message</b> \
289                 {message}</p><hr class=\"line\" /><h3>Apache Tomcat (mock)</h3></body></html>"
290            )),
291        }
292    }
293
294    /// The whole body as text (tests).
295    #[cfg(test)]
296    pub(crate) fn text(self) -> String {
297        match self.body {
298            Body::Text(s) => s,
299            Body::Stream(it) => it.collect(),
300        }
301    }
302
303    fn into_response(self) -> tiny_http::Response<Box<dyn Read + Send>> {
304        let header =
305            tiny_http::Header::from_bytes("Content-Type", self.content_type).expect("valid header");
306        let (reader, len): (Box<dyn Read + Send>, Option<usize>) = match self.body {
307            Body::Text(s) => {
308                let len = s.len();
309                (Box::new(io::Cursor::new(s.into_bytes())), Some(len))
310            }
311            Body::Stream(it) => (Box::new(IterReader::new(it)), None),
312        };
313        tiny_http::Response::new(self.status.into(), vec![header], reader, len, None)
314    }
315}
316
317/// Reads the concatenation of an iterator's strings.
318struct IterReader {
319    iter: Box<dyn Iterator<Item = String> + Send>,
320    buf: Vec<u8>,
321    pos: usize,
322}
323
324impl IterReader {
325    fn new(iter: Box<dyn Iterator<Item = String> + Send>) -> Self {
326        Self {
327            iter,
328            buf: Vec::new(),
329            pos: 0,
330        }
331    }
332}
333
334impl Read for IterReader {
335    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
336        let mut n = 0;
337        while n < out.len() {
338            if self.pos == self.buf.len() {
339                match self.iter.next() {
340                    Some(s) => {
341                        self.buf = s.into_bytes();
342                        self.pos = 0;
343                        continue;
344                    }
345                    None => break,
346                }
347            }
348            let k = (self.buf.len() - self.pos).min(out.len() - n);
349            out[n..n + k].copy_from_slice(&self.buf[self.pos..self.pos + k]);
350            self.pos += k;
351            n += k;
352        }
353        Ok(n)
354    }
355}
356
357/// Dispatch one request URL (path and query).
358pub(crate) fn route(url: &str, clock: &Clock) -> Reply {
359    let (path, query) = url.split_once('?').unwrap_or((url, ""));
360    let params = Params::parse(query);
361    match path.trim_end_matches('/') {
362        "/retrieval/data/getData.json" => data::get_data(&params, clock),
363        "/retrieval/bpl/getMatchingPVs" => bpl::get_matching_pvs(&params),
364        "/retrieval/bpl/getMetadata" => bpl::get_metadata(&params),
365        "/retrieval/bpl/getVersion" => bpl::get_version(),
366        "/mgmt/bpl/getAllPVs" => bpl::get_all_pvs(&params),
367        _ => Reply::error(404, "Not Found", &format!("{path} is not served here")),
368    }
369}
370
371#[cfg(test)]
372mod tests;
mockarchiver/lib.rs

mockarchiver/
lib.rs
