crates/signald/tests/streaming.rs

 1//! Live-streaming acceptance test (spec §1.3): a subscriber gets the current
 2//! value of each metric immediately on connect (last-value cache), then receives
 3//! updates as values change — over the real Unix socket, not just the hub.
 4
 5use std::io::BufReader;
 6use std::os::unix::net::UnixStream;
 7use std::path::{Path, PathBuf};
 8use std::thread;
 9use std::time::{Duration, Instant};
10
11use signal_schema::{wire, Signal, SignalName, Source, Value, SCHEMA_VERSION};
12use signald::hub::Hub;
13use signald::publish;
14
15fn sig(name: SignalName, value: f64) -> Signal {
16    Signal {
17        schema_version: SCHEMA_VERSION,
18        ts: 1,
19        source: Source::Terminal,
20        name,
21        value: Value(value),
22        tag: None,
23    }
24}
25
26#[test]
27fn subscriber_sees_cached_value_then_a_live_update() {
28    let socket = unique_socket();
29    let hub = Hub::new();
30
31    // A value published before anyone connects must still reach a subscriber
32    // (the last-value cache).
33    hub.publish(sig(SignalName::KeysPerMin, 12.0));
34
35    let serve_hub = hub.clone();
36    let serve_socket = socket.clone();
37    thread::spawn(move || {
38        let _ = publish::serve(&serve_socket, serve_hub);
39    });
40
41    let stream = connect(&socket);
42    let mut reader = BufReader::new(stream);
43
44    // 1. On connect: the cached snapshot arrives immediately.
45    let first = wire::read_frame(&mut reader).unwrap().expect("snapshot frame");
46    assert_eq!(first.name, SignalName::KeysPerMin);
47    assert_eq!(first.value, Value(12.0));
48
49    // 2. A change after subscribing is streamed live.
50    hub.publish(sig(SignalName::KeysPerMin, 99.0));
51    let update = wire::read_frame(&mut reader).unwrap().expect("live update frame");
52    assert_eq!(update.name, SignalName::KeysPerMin);
53    assert_eq!(update.value, Value(99.0));
54
55    let _ = std::fs::remove_file(&socket);
56}
57
58fn connect(socket: &Path) -> UnixStream {
59    let deadline = Instant::now() + Duration::from_secs(5);
60    loop {
61        if let Ok(s) = UnixStream::connect(socket) {
62            return s;
63        }
64        assert!(Instant::now() < deadline, "signald socket never came up");
65        thread::sleep(Duration::from_millis(20));
66    }
67}
68
69fn unique_socket() -> PathBuf {
70    let nanos = std::time::SystemTime::now()
71        .duration_since(std::time::UNIX_EPOCH)
72        .unwrap()
73        .as_nanos();
74    std::env::temp_dir().join(format!("signald-stream-{}-{nanos}.sock", std::process::id()))
75}