streamer/subjects.rs
1//! NATS subject builders for LabJack live data and archive exports.
2//!
3//! The streamer publishes each channel's scans on a live subject, the archiver
4//! consumes the same subjects from local JetStream and writes Parquet, and the exporter
5//! listens on an export request subject. All of them build their subject names here
6//! so the layouts stay in one place.
7//!
8//! Two layouts are supported:
9//!
10//! * Legacy, keyed by asset number: `avenabox.<asset>.data.chNN`, for example
11//! `avenabox.1456.data.ch11`. The root is used exactly as configured.
12//! * Structured, keyed by identity: `avenars.<site>.<box>.<source>.live.chNN`, for
13//! example `avenars.i69.i69-mu1.i69-lj2.live.ch11`. Every token is passed through
14//! [`sanitize_token`].
15//!
16//! The structured layout is chosen when the root is `avenars` or any of site, box or
17//! source ID is set (see [`uses_structured_namespace`]).
18
19#![allow(dead_code)]
20
21/// Converts arbitrary identity text into a single NATS subject token.
22///
23/// Leading and trailing whitespace is trimmed. ASCII letters and digits are
24/// lowercased, `-` and `_` are kept, and whitespace, `.` and `/` become `-`. Every
25/// other character is dropped. Leading and trailing `-` are then stripped. This keeps
26/// `.` (the subject separator) and the wildcards `*` and `>` out of the token.
27///
28/// # Arguments
29///
30/// * `raw` - Identity text such as a site ID, box ID or LabJack name.
31///
32/// # Returns
33///
34/// The sanitized token, or `unknown` if nothing usable remains.
35///
36/// # Examples
37///
38/// ```text
39/// sanitize_token("I69 MU1") -> "i69-mu1"
40/// sanitize_token("site.a/b") -> "site-a-b"
41/// sanitize_token(" *> ") -> "unknown"
42/// ```
43fn sanitize_token(raw: &str) -> String {
44 let mut out = String::with_capacity(raw.len());
45 for ch in raw.trim().chars() {
46 if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
47 out.push(ch.to_ascii_lowercase());
48 } else if ch.is_whitespace() || ch == '.' || ch == '/' {
49 out.push('-');
50 }
51 }
52
53 let out = out.trim_matches('-').to_string();
54 if out.is_empty() {
55 "unknown".to_string()
56 } else {
57 out
58 }
59}
60
61/// Formats a LabJack channel number as the `chNN` token used in subjects.
62///
63/// The number is zero-padded to at least two digits.
64///
65/// # Arguments
66///
67/// * `ch` - LabJack channel number.
68///
69/// # Examples
70///
71/// ```text
72/// pad_channel(3) -> "ch03"
73/// pad_channel(11) -> "ch11"
74/// pad_channel(120) -> "ch120"
75/// ```
76pub fn pad_channel(ch: u8) -> String {
77 format!("ch{ch:02}")
78}
79
80/// Formats an asset number with the three-digit legacy subject width.
81///
82/// The number is zero-padded to at least three digits; longer numbers are not
83/// truncated.
84///
85/// # Arguments
86///
87/// * `n` - Asset number.
88///
89/// # Examples
90///
91/// ```text
92/// pad_asset(7) -> "007"
93/// pad_asset(1456) -> "1456"
94/// ```
95pub fn pad_asset(n: u32) -> String {
96 format!("{n:03}")
97}
98
99/// Treats an empty identity field the same as a missing one.
100///
101/// The webapp builds the same subjects with JavaScript `||` fallbacks, where an empty
102/// string counts as missing. Doing the same here keeps both sides on one subject when
103/// a config has `""` in an identity field.
104///
105/// # Arguments
106///
107/// * `value` - Identity field from the config or environment.
108///
109/// # Returns
110///
111/// `None` if `value` is `None` or an empty string, otherwise `value`.
112fn present(value: Option<&str>) -> Option<&str> {
113 value.filter(|v| !v.is_empty())
114}
115
116/// Picks the source token shared by every structured subject.
117///
118/// The first of `source_id`, `labjack_name` and `asset<NNN>` that is present is used,
119/// passed through [`sanitize_token`]. Live channel subjects, the stream wildcard and
120/// the export request subject all go through this function, so a config without
121/// identity fields still produces a stream that matches its own channels and an
122/// export subject that the webapp can find. The webapp (`webapp/src/lib/subjects.ts`)
123/// uses the same order. `unknown-source` is only used when there is no asset number
124/// either, which can happen for the exporter when neither `SOURCE_ID`, `LABJACK_NAME`
125/// nor `ASSET_NUMBER` is set.
126///
127/// # Arguments
128///
129/// * `source_id` - Configured source ID, if any.
130/// * `labjack_name` - LabJack name, if any.
131/// * `asset` - Asset number, if known.
132///
133/// # Examples
134///
135/// ```text
136/// (Some("i69-lj2"), Some("Unit A"), Some(42)) -> "i69-lj2"
137/// (None, Some("Unit A"), Some(42)) -> "unit-a"
138/// (None, None, Some(42)) -> "asset042"
139/// (None, None, None) -> "unknown-source"
140/// ```
141fn source_token(source_id: Option<&str>, labjack_name: Option<&str>, asset: Option<u32>) -> String {
142 match (present(source_id).or(present(labjack_name)), asset) {
143 (Some(source), _) => sanitize_token(source),
144 (None, Some(asset)) => format!("asset{}", pad_asset(asset)),
145 (None, None) => "unknown-source".to_string(),
146 }
147}
148
149/// Decides whether a config should use the structured subject namespace.
150///
151/// # Arguments
152///
153/// * `nats_subject` - Configured subject root, for example `avenabox` or `avenars`.
154/// * `site_id` - Configured site ID, if any.
155/// * `box_id` - Configured box ID, if any.
156/// * `source_id` - Configured source ID, if any.
157///
158/// # Returns
159///
160/// `true` if the trimmed root is exactly `avenars` or any of the three IDs is set and
161/// not empty. A LabJack name alone does not switch to the structured layout.
162fn uses_structured_namespace(
163 nats_subject: &str,
164 site_id: Option<&str>,
165 box_id: Option<&str>,
166 source_id: Option<&str>,
167) -> bool {
168 nats_subject.trim() == "avenars"
169 || present(site_id).is_some()
170 || present(box_id).is_some()
171 || present(source_id).is_some()
172}
173
174/// Builds the live-data subject for one LabJack channel.
175///
176/// Legacy configs use `<root>.<asset>.data.<channel>`, with the root used as given.
177/// Structured configs use `<root>.<site>.<box>.<source>.live.<channel>`, with every
178/// token passed through [`sanitize_token`]. In the structured layout a missing or empty
179/// site or box becomes `unknown-site` or `unknown-box`, and the source is chosen by
180/// [`source_token`]: the first of `source_id`, `labjack_name` or `asset<NNN>` that is
181/// set.
182///
183/// # Arguments
184///
185/// * `nats_subject` - Subject root, for example `avenabox` or `avenars`.
186/// * `asset` - Asset number. Used in the legacy layout and as the last-resort source
187/// token in the structured layout.
188/// * `channel` - LabJack channel number, formatted by [`pad_channel`].
189/// * `site_id` - Site ID for the structured layout.
190/// * `box_id` - Box ID for the structured layout.
191/// * `labjack_name` - LabJack name, used as the source when `source_id` is `None`.
192/// * `source_type` - Accepted for config compatibility; not used in the subject.
193/// * `source_id` - Source ID for the structured layout.
194///
195/// # Examples
196///
197/// ```text
198/// ("avenabox", 1456, 11, None, None, None, None, None)
199/// -> "avenabox.1456.data.ch11"
200/// ("avenars", 1456, 11, Some("i69"), Some("i69-mu1"), Some("i69-lj2"), None, None)
201/// -> "avenars.i69.i69-mu1.i69-lj2.live.ch11"
202/// ("avenars", 42, 3, None, None, None, None, None)
203/// -> "avenars.unknown-site.unknown-box.asset042.live.ch03"
204/// ```
205pub fn live_labjack_channel_subject(
206 nats_subject: &str,
207 asset: u32,
208 channel: u8,
209 site_id: Option<&str>,
210 box_id: Option<&str>,
211 labjack_name: Option<&str>,
212 source_type: Option<&str>,
213 source_id: Option<&str>,
214) -> String {
215 if !uses_structured_namespace(nats_subject, site_id, box_id, source_id) {
216 return format!(
217 "{}.{}.data.{}",
218 nats_subject,
219 pad_asset(asset),
220 pad_channel(channel)
221 );
222 }
223
224 let root = sanitize_token(nats_subject);
225 let site_id = sanitize_token(present(site_id).unwrap_or("unknown-site"));
226 let box_id = sanitize_token(present(box_id).unwrap_or("unknown-box"));
227 let source_id = source_token(source_id, labjack_name, Some(asset));
228
229 let _ = source_type;
230
231 format!(
232 "{root}.{site_id}.{box_id}.{source_id}.live.{}",
233 pad_channel(channel)
234 )
235}
236
237/// Builds the JetStream subject wildcard for all live channels from one source.
238///
239/// Legacy configs get `<root>.*.data.*`, which matches every asset under the root.
240/// Structured configs get `<root>.<site>.<box>.<source>.live.*`, with the same tokens
241/// and fallbacks as [`live_labjack_channel_subject`], so the wildcard always matches
242/// the channel subjects built from the same config.
243///
244/// # Arguments
245///
246/// * `nats_subject` - Subject root, for example `avenabox` or `avenars`.
247/// * `asset` - Asset number, the last-resort source token in the structured layout.
248/// Not used in the legacy layout, whose wildcard covers every asset.
249/// * `site_id` - Site ID for the structured layout.
250/// * `box_id` - Box ID for the structured layout.
251/// * `labjack_name` - LabJack name, used as the source when `source_id` is `None`.
252/// * `source_type` - Accepted for config compatibility; not used in the subject.
253/// * `source_id` - Source ID for the structured layout.
254///
255/// # Examples
256///
257/// ```text
258/// ("avenabox", 1456, None, None, None, None, None)
259/// -> "avenabox.*.data.*"
260/// ("avenars", 1456, Some("i69"), Some("i69-mu1"), Some("i69-lj2"), None, None)
261/// -> "avenars.i69.i69-mu1.i69-lj2.live.*"
262/// ("avenars", 42, None, None, None, None, None)
263/// -> "avenars.unknown-site.unknown-box.asset042.live.*"
264/// ```
265pub fn live_labjack_stream_subject(
266 nats_subject: &str,
267 asset: u32,
268 site_id: Option<&str>,
269 box_id: Option<&str>,
270 labjack_name: Option<&str>,
271 source_type: Option<&str>,
272 source_id: Option<&str>,
273) -> String {
274 if !uses_structured_namespace(nats_subject, site_id, box_id, source_id) {
275 return format!("{nats_subject}.*.data.*");
276 }
277
278 let root = sanitize_token(nats_subject);
279 let site_id = sanitize_token(present(site_id).unwrap_or("unknown-site"));
280 let box_id = sanitize_token(present(box_id).unwrap_or("unknown-box"));
281 let source_id = source_token(source_id, labjack_name, Some(asset));
282
283 let _ = source_type;
284
285 format!("{root}.{site_id}.{box_id}.{source_id}.live.*")
286}
287
288/// Builds the NATS request subject that the exporter serves archive exports on.
289///
290/// Always uses the structured layout `<root>.<site>.<box>.<source>.export.request`,
291/// whatever the root, with every token passed through [`sanitize_token`]. Missing or
292/// empty IDs become `unknown-site` and `unknown-box`, and the source is chosen by
293/// [`source_token`] exactly as for the live subjects, so the webapp reaches the
294/// exporter with the subject it derives from the source's config.
295///
296/// # Arguments
297///
298/// * `nats_subject` - Subject root, for example `avenars`.
299/// * `asset` - Asset number, the last-resort source token, if known.
300/// * `site_id` - Site ID.
301/// * `box_id` - Box ID.
302/// * `labjack_name` - LabJack name, used as the source when `source_id` is not set.
303/// * `source_type` - Accepted for config compatibility; not used in the subject.
304/// * `source_id` - Source ID.
305///
306/// # Examples
307///
308/// ```text
309/// ("avenars", None, Some("i69"), Some("i69-mu1"), None, Some("labjack"), Some("i69-lj2"))
310/// -> "avenars.i69.i69-mu1.i69-lj2.export.request"
311/// ("avenars", Some(42), None, None, None, None, None)
312/// -> "avenars.unknown-site.unknown-box.asset042.export.request"
313/// ```
314pub fn archive_export_request_subject(
315 nats_subject: &str,
316 asset: Option<u32>,
317 site_id: Option<&str>,
318 box_id: Option<&str>,
319 labjack_name: Option<&str>,
320 source_type: Option<&str>,
321 source_id: Option<&str>,
322) -> String {
323 let root = sanitize_token(nats_subject);
324 let site = sanitize_token(present(site_id).unwrap_or("unknown-site"));
325 let box_id = sanitize_token(present(box_id).unwrap_or("unknown-box"));
326 let source_id = source_token(source_id, labjack_name, asset);
327
328 let _ = source_type;
329
330 format!("{root}.{site}.{box_id}.{source_id}.export.request")
331}
332
333/// Checks whether an existing stream subject is compatible with a desired one.
334///
335/// An exact match is compatible. When the desired subject is a legacy wildcard
336/// `<prefix>.*.data.*`, an existing subject is also compatible if it starts with
337/// `<prefix>.` and ends with `.data.*`, so an older per-asset pattern such as
338/// `avenabox.1456.data.*` is accepted. Anything else is incompatible.
339///
340/// # Arguments
341///
342/// * `existing` - A subject already configured on the JetStream stream.
343/// * `desired_namespace` - The subject the current config wants, typically from
344/// [`live_labjack_stream_subject`].
345///
346/// # Examples
347///
348/// ```text
349/// ("avenabox.1456.data.*", "avenabox.*.data.*") -> true
350/// ("avenabox.*.data.*", "avenabox.*.data.*") -> true
351/// ("avenars.i69.a.b.live.*", "avenars.i69.a.c.live.*") -> false
352/// ```
353pub fn stream_subject_is_compatible(existing: &str, desired_namespace: &str) -> bool {
354 if existing == desired_namespace {
355 return true;
356 }
357
358 if let Some(prefix) = desired_namespace.strip_suffix(".*.data.*") {
359 return existing.starts_with(&format!("{prefix}.")) && existing.ends_with(".data.*");
360 }
361
362 false
363}
364
365#[cfg(test)]
366mod tests {
367 use super::*;
368
369 /// Legacy configs keep the `<root>.<asset>.data.chNN` layout.
370 #[test]
371 fn legacy_subjects_are_preserved() {
372 assert_eq!(
373 live_labjack_channel_subject("avenabox", 1456, 11, None, None, None, None, None),
374 "avenabox.1456.data.ch11"
375 );
376 assert_eq!(
377 live_labjack_stream_subject("avenabox", 1456, None, None, None, None, None),
378 "avenabox.*.data.*"
379 );
380 }
381
382 /// Structured configs produce site, box and source tokens in every subject kind.
383 #[test]
384 fn structured_subjects_include_site_box_source_and_channel() {
385 assert_eq!(
386 live_labjack_channel_subject(
387 "avenars",
388 1456,
389 11,
390 Some("i69"),
391 Some("i69-mu1"),
392 Some("i69-lj2"),
393 None,
394 None,
395 ),
396 "avenars.i69.i69-mu1.i69-lj2.live.ch11"
397 );
398 assert_eq!(
399 live_labjack_stream_subject(
400 "avenars",
401 1456,
402 Some("i69"),
403 Some("i69-mu1"),
404 Some("i69-lj2"),
405 None,
406 None,
407 ),
408 "avenars.i69.i69-mu1.i69-lj2.live.*"
409 );
410 assert_eq!(
411 archive_export_request_subject(
412 "avenars",
413 Some(1456),
414 Some("i69"),
415 Some("i69-mu1"),
416 Some("Unit A"),
417 Some("labjack"),
418 Some("i69-lj2"),
419 ),
420 "avenars.i69.i69-mu1.i69-lj2.export.request"
421 );
422 }
423
424 /// Returns whether a NATS subject matches a pattern with `*` and `>` wildcards.
425 fn nats_subject_matches(pattern: &str, subject: &str) -> bool {
426 let mut pattern = pattern.split('.');
427 let mut subject = subject.split('.');
428 loop {
429 match (pattern.next(), subject.next()) {
430 (Some(">"), Some(_)) => return true,
431 (Some("*"), Some(_)) => {}
432 (Some(p), Some(s)) if p == s => {}
433 (None, None) => return true,
434 _ => return false,
435 }
436 }
437 }
438
439 /// Without identity fields, the stream wildcard still covers every channel subject
440 /// and all three subjects share the `asset<NNN>` source token.
441 #[test]
442 fn stream_wildcard_matches_channels_without_identity_fields() {
443 let stream = live_labjack_stream_subject("avenars", 42, None, None, None, None, None);
444 assert_eq!(stream, "avenars.unknown-site.unknown-box.asset042.live.*");
445 for channel in [0, 3, 11, 120] {
446 let live =
447 live_labjack_channel_subject("avenars", 42, channel, None, None, None, None, None);
448 assert!(
449 nats_subject_matches(&stream, &live),
450 "{stream} should match {live}"
451 );
452 }
453 assert_eq!(
454 archive_export_request_subject("avenars", Some(42), None, None, None, None, None),
455 "avenars.unknown-site.unknown-box.asset042.export.request"
456 );
457 }
458
459 /// Empty identity fields fall back the same way as missing ones, like the webapp.
460 #[test]
461 fn empty_identity_fields_fall_back_like_missing_ones() {
462 let stream =
463 live_labjack_stream_subject("avenars", 7, Some(""), Some(""), Some(""), None, Some(""));
464 let live = live_labjack_channel_subject(
465 "avenars",
466 7,
467 5,
468 Some(""),
469 Some(""),
470 Some(""),
471 None,
472 Some(""),
473 );
474 let export = archive_export_request_subject(
475 "avenars",
476 Some(7),
477 Some(""),
478 Some(""),
479 Some(""),
480 None,
481 Some(""),
482 );
483 assert_eq!(stream, "avenars.unknown-site.unknown-box.asset007.live.*");
484 assert_eq!(live, "avenars.unknown-site.unknown-box.asset007.live.ch05");
485 assert!(nats_subject_matches(&stream, &live));
486 assert_eq!(
487 export,
488 "avenars.unknown-site.unknown-box.asset007.export.request"
489 );
490
491 // With only a LabJack name, every subject uses it as the source.
492 let stream =
493 live_labjack_stream_subject("avenars", 7, None, None, Some("Unit A"), None, None);
494 let live =
495 live_labjack_channel_subject("avenars", 7, 5, None, None, Some("Unit A"), None, None);
496 let export = archive_export_request_subject(
497 "avenars",
498 Some(7),
499 None,
500 None,
501 Some("Unit A"),
502 None,
503 None,
504 );
505 assert!(nats_subject_matches(&stream, &live));
506 assert_eq!(
507 export,
508 "avenars.unknown-site.unknown-box.unit-a.export.request"
509 );
510
511 // An empty legacy config stays in the legacy layout.
512 assert_eq!(
513 live_labjack_channel_subject("avenabox", 7, 5, Some(""), None, None, None, None),
514 "avenabox.007.data.ch05"
515 );
516 }
517
518 /// The exporter falls back to `unknown-source` only when it has no asset number.
519 #[test]
520 fn export_subject_without_any_source_uses_unknown_source() {
521 assert_eq!(
522 archive_export_request_subject(
523 "avenars",
524 None,
525 Some("i69"),
526 Some("i69-mu1"),
527 None,
528 None,
529 None
530 ),
531 "avenars.i69.i69-mu1.unknown-source.export.request"
532 );
533 }
534}