svc-gw/src/bus.rs

//! The real input: envelopes from the bus, latest per signal.
//!
//! Subscribes to the envelope topics and keeps the newest envelope of each signal,
//! keyed by the topic's last segment (the signal name, contracts guideline 1).
//! Staleness is judged here, by arrival: a signal not refreshed within `STALE`,
//! or known only from the broker's retained copy (it may predate a reboot), is
//! sent with `quality: "stale"`. Signals are never dropped once seen; an old value
//! shown stale is the honest picture.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use rumqttc::{Client, Event, MqttOptions, Packet, QoS};
use serde_json::{json, Map, Value};

const TOPICS: [&str; 3] = ["yukart/vehicle/+", "yukart/host/+", "yukart/gnss/+"];
const STALE: Duration = Duration::from_secs(5);

struct Entry {
    envelope: Value,
    at: Instant,
    retained: bool,
}

#[derive(Clone)]
pub struct Bus {
    latest: Arc<Mutex<HashMap<String, Entry>>>,
}

impl Bus {
    /// Connects in the background and reconnects forever; never blocks the caller.
    pub fn start(host: &str, port: u16) -> Self {
        let bus = Bus { latest: Arc::default() };
        let mut opts = MqttOptions::new("svc-gw", host, port);
        opts.set_keep_alive(Duration::from_secs(5));
        let (client, mut connection) = Client::new(opts, 64);
        let latest = bus.latest.clone();
        thread::spawn(move || {
            let mut up = false;
            for ev in connection.iter() {
                match ev {
                    Ok(Event::Incoming(Packet::ConnAck(_))) => {
                        up = true;
                        eprintln!("bus: connected; subscribing to {}", TOPICS.join(", "));
                        // A clean session forgets subscriptions: renew them on every connect.
                        for t in TOPICS {
                            let _ = client.try_subscribe(t, QoS::AtMostOnce);
                        }
                    }
                    Ok(Event::Incoming(Packet::Publish(p))) => {
                        let Some(name) = p.topic.rsplit('/').next() else { continue };
                        let Ok(envelope) = serde_json::from_slice::<Value>(&p.payload) else { continue };
                        if !envelope.is_object() {
                            continue;
                        }
                        let entry = Entry { envelope, at: Instant::now(), retained: p.retain };
                        latest.lock().unwrap().insert(name.to_string(), entry);
                    }
                    Ok(_) => {}
                    Err(e) => {
                        if std::mem::take(&mut up) {
                            eprintln!("bus: {e}; retrying");
                        }
                        thread::sleep(Duration::from_secs(1));
                    }
                }
            }
        });
        bus
    }

    /// Every signal seen, newest envelope each, marked stale where it is.
    pub fn snapshot(&self) -> Map<String, Value> {
        let latest = self.latest.lock().unwrap();
        latest
            .iter()
            .map(|(name, e)| {
                let mut env = e.envelope.clone();
                if (e.retained || e.at.elapsed() > STALE) && env.get("quality").is_none() {
                    env["quality"] = json!("stale");
                }
                (name.clone(), env)
            })
            .collect()
    }
}