//! Everything heard, and the bus kept in step with it.
//!
//! One task owns the state. Sources send it events; it applies them, then
//! puts the difference on the bus. Nothing else writes to the objects, so
//! there is nothing to lock.
//!
//! The last of it is kept on disk, so a widget at login shows what was true
//! when the machine went down -- marked stale -- rather than nothing until
//! a broker answers.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::PathBuf;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::sync::mpsc;
use tracing::{debug, warn};
use zbus::Connection;
use crate::bus::{self, Origin, Query, Root};
use crate::points::{self, Point};
use crate::source::Event;
use crate::wire::Reading;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Entry {
pub reading: Reading,
pub source: String,
pub updated: i64,
}
#[derive(Debug, Clone, Default, Serialize)]
pub struct SourceStatus {
pub name: String,
pub connected: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
#[derive(Debug, Default, PartialEq)]
pub struct Change {
pub updated: Vec<String>,
pub removed: Vec<String>,
pub sources: bool,
}
#[derive(Debug, Default)]
pub struct State {
pub entries: BTreeMap<String, Entry>,
/// Plus readings by key. Apart from the entries, since the Plus board can
/// arrive before the board it extends.
pub plus: HashMap<String, Value>,
pub sources: BTreeMap<String, SourceStatus>,
pub stale_after: i64,
}
#[derive(Serialize, Deserialize)]
struct Cache {
entries: BTreeMap<String, Entry>,
plus: HashMap<String, Value>,
}
pub fn now() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or_default()
}
impl State {
pub fn new(sources: impl IntoIterator<Item = String>, stale_after: Duration) -> State {
State {
sources: sources
.into_iter()
.map(|s| {
(
s.clone(),
SourceStatus {
name: s,
..Default::default()
},
)
})
.collect(),
stale_after: stale_after.as_secs() as i64,
..Default::default()
}
}
pub fn is_stale(&self, entry: &Entry, now: i64) -> bool {
now - entry.updated > self.stale_after
}
pub fn apply(&mut self, event: Event, now: i64) -> Change {
let mut change = Change::default();
match event {
Event::Status {
source,
connected,
error,
} => {
let status = self.sources.entry(source.clone()).or_default();
status.name = source;
change.sources = status.connected != connected || status.error != error;
status.connected = connected;
status.error = error;
}
Event::Readings {
source,
readings,
whole,
} => {
let keys: HashSet<String> = readings.iter().map(|r| r.key.clone()).collect();
for reading in readings {
change.updated.push(reading.key.clone());
self.entries.insert(
reading.key.clone(),
Entry {
reading,
source: source.clone(),
updated: now,
},
);
}
// A board is the whole vision: what this source sent before
// and has not sent now has been taken out of it. Another
// source's copy of the same key is that source's to retract.
if whole {
let gone: Vec<String> = self
.entries
.iter()
.filter(|(k, e)| e.source == source && !keys.contains(*k))
.map(|(k, _)| k.clone())
.collect();
for k in gone {
self.entries.remove(&k);
self.plus.remove(&k);
change.removed.push(k);
}
}
}
Event::Plus { entries } => {
for (key, value) in entries {
if self.entries.contains_key(&key) {
change.updated.push(key.clone());
}
self.plus.insert(key, value);
}
}
}
change
}
fn points_of(entry: &Entry) -> Vec<Point> {
points::summarise(&entry.reading.kind, &entry.reading.data)
}
fn name_of(entry: &Entry) -> String {
entry
.reading
.name
.clone()
.unwrap_or_else(|| entry.reading.key.clone())
}
pub fn query(&self, key: &str, now: i64) -> Option<Query> {
let e = self.entries.get(key)?;
Some(Query {
key: key.to_string(),
kind: e.reading.kind.clone(),
id: e.reading.id.clone(),
query_id: e.reading.query_id,
name: State::name_of(e),
source: e.source.clone(),
updated: e.updated,
stale: self.is_stale(e, now),
data: e.reading.data.clone(),
meta: e.reading.meta.clone(),
plus: self.plus.get(key).cloned().unwrap_or(Value::Null),
points: State::points_of(e).into_iter().map(|p| p.text).collect(),
})
}
pub fn board_json(&self, now: i64) -> String {
let board: serde_json::Map<String, Value> = self
.entries
.iter()
.map(|(k, e)| {
(
k.clone(),
json!({
"name": State::name_of(e),
"type": e.reading.kind,
"id": e.reading.id,
"query_id": e.reading.query_id,
"source": e.source,
"updated": e.updated,
"stale": self.is_stale(e, now),
"data": e.reading.data,
"meta": e.reading.meta,
"plus": self.plus.get(k).cloned().unwrap_or(Value::Null),
"points": State::points_of(e),
}),
)
})
.collect();
Value::Object(board).to_string()
}
pub fn points_json(&self, now: i64) -> String {
let points: Vec<Value> = self
.entries
.iter()
.flat_map(|(k, e)| {
let stale = self.is_stale(e, now);
State::points_of(e).into_iter().map(move |p| {
let mut v = json!({
"key": k,
"name": State::name_of(e),
"type": e.reading.kind,
"text": p.text,
"stale": stale,
});
if let Some(c) = p.color {
v["color"] = json!(c);
}
v
})
})
.collect();
Value::Array(points).to_string()
}
pub fn sources_json(&self) -> String {
serde_json::to_string(&self.sources.values().collect::<Vec<_>>()).unwrap_or_default()
}
pub fn origins(&self) -> HashMap<String, Origin> {
self.entries
.iter()
.map(|(k, e)| {
(
k.clone(),
Origin {
source: e.source.clone(),
query_id: e.reading.query_id,
},
)
})
.collect()
}
pub fn load(&mut self, path: &PathBuf) {
let Ok(bytes) = std::fs::read(path) else {
return;
};
match serde_json::from_slice::<Cache>(&bytes) {
Ok(cache) => {
// Only what a configured source would have sent: a source
// since removed from the config would never take its own back.
self.entries = cache
.entries
.into_iter()
.filter(|(_, e)| self.sources.contains_key(&e.source))
.collect();
self.plus = cache.plus;
}
Err(e) => warn!("ignoring unreadable cache {}: {e}", path.display()),
}
}
fn save(&self, path: &PathBuf) {
let cache = Cache {
entries: self.entries.clone(),
plus: self.plus.clone(),
};
let write = || -> std::io::Result<()> {
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir)?;
}
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, serde_json::to_vec(&cache)?)?;
std::fs::rename(tmp, path)
};
if let Err(e) = write() {
warn!("could not keep the last board at {}: {e}", path.display());
}
}
}
pub fn cache_path() -> PathBuf {
dirs::cache_dir()
.unwrap_or_else(|| PathBuf::from("."))
.join("apollos-dbus")
.join("last.json")
}
/// How often staleness is looked at. A minute late is not worth more wakeups.
const TICK: Duration = Duration::from_secs(30);
pub async fn run(
conn: Connection,
mut state: State,
mut events: mpsc::Receiver<Event>,
cache: PathBuf,
) {
let mut stale: HashMap<String, bool> = HashMap::new();
let everything = Change {
updated: state.entries.keys().cloned().collect(),
removed: vec![],
sources: true,
};
sync(&conn, &state, &everything, &mut stale, false).await;
let mut tick = tokio::time::interval(TICK);
loop {
tokio::select! {
Some(event) = events.recv() => {
let change = state.apply(event, now());
if change == Change::default() {
continue;
}
sync(&conn, &state, &change, &mut stale, true).await;
if !change.updated.is_empty() || !change.removed.is_empty() {
state.save(&cache);
}
}
_ = tick.tick() => {
let t = now();
let flipped: Vec<String> = state
.entries
.iter()
.filter(|(k, e)| stale.get(*k) != Some(&state.is_stale(e, t)))
.map(|(k, _)| k.clone())
.collect();
if !flipped.is_empty() {
let change = Change { updated: flipped, ..Default::default() };
sync(&conn, &state, &change, &mut stale, false).await;
}
}
}
}
}
/// Put a change on the bus. `fresh` when it is news, rather than the clock
/// moving on -- only news gets a `QueryUpdated`.
async fn sync(
conn: &Connection,
state: &State,
change: &Change,
stale: &mut HashMap<String, bool>,
fresh: bool,
) {
let server = conn.object_server();
let t = now();
for key in &change.removed {
stale.remove(key);
if let Err(e) = server.remove::<Query, _>(bus::query_path(key)).await {
debug!("removing {key}: {e}");
}
}
for key in &change.updated {
let Some(q) = state.query(key, t) else {
continue;
};
stale.insert(key.clone(), q.stale);
let path = bus::query_path(key);
match server.interface::<_, Query>(path.as_str()).await {
Ok(iref) => {
let mut current = iref.get_mut().await;
*current = q;
if let Err(e) = bus::announce_query(¤t, iref.signal_emitter()).await {
warn!("announcing {key}: {e}");
}
}
Err(_) => {
if let Err(e) = server.at(path.as_str(), q).await {
warn!("adding {key}: {e}");
}
}
}
}
let Ok(root) = server.interface::<_, Root>(bus::ROOT).await else {
warn!("the root object has gone");
return;
};
let emitter = root.signal_emitter();
let mut r = root.get_mut().await;
let queries: Vec<String> = state.entries.keys().cloned().collect();
let entries_changed = !change.updated.is_empty() || !change.removed.is_empty();
let result: zbus::Result<()> = async {
if entries_changed {
r.board = state.board_json(t);
r.points = state.points_json(t);
r.origins = state.origins();
r.board_changed(emitter).await?;
r.points_changed(emitter).await?;
}
if r.queries != queries {
r.queries = queries;
r.queries_changed(emitter).await?;
}
if change.sources {
r.sources = state.sources_json();
r.sources_changed(emitter).await?;
let connected = state.sources.values().any(|s| s.connected);
if r.connected != connected {
r.connected = connected;
r.connected_changed(emitter).await?;
}
}
if fresh {
for key in &change.updated {
Root::query_updated(emitter, key).await?;
}
}
Ok(())
}
.await;
if let Err(e) = result {
warn!("updating the root object: {e}");
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::wire;
fn readings(payload: &str) -> Vec<Reading> {
wire::board(payload.as_bytes()).unwrap()
}
#[test]
fn a_board_retracts_what_it_no_longer_has_but_only_its_own() {
let mut s = State::new(
["ankyra:home".into(), "keryx:desk".into()],
Duration::from_secs(60),
);
s.apply(
Event::Readings {
source: "ankyra:home".into(),
readings: readings(r#"{"gtfs-1": [], "weather-2": []}"#),
whole: true,
},
0,
);
s.apply(
Event::Readings {
source: "keryx:desk".into(),
readings: readings(r#"{"gbfs-3": []}"#),
whole: false,
},
0,
);
let change = s.apply(
Event::Readings {
source: "ankyra:home".into(),
readings: readings(r#"{"gtfs-1": []}"#),
whole: true,
},
10,
);
assert_eq!(change.removed, vec!["weather-2".to_string()]);
assert_eq!(
s.entries.keys().collect::<Vec<_>>(),
vec!["gbfs-3", "gtfs-1"]
);
}
#[test]
fn plus_before_its_board_is_kept_for_it() {
let mut s = State::new(["ankyra:home".into()], Duration::from_secs(60));
let change = s.apply(
Event::Plus {
entries: vec![("gtfs-1".into(), json!([1]))],
},
0,
);
assert!(change.updated.is_empty(), "nothing to announce yet");
s.apply(
Event::Readings {
source: "ankyra:home".into(),
readings: readings(r#"{"gtfs-1": []}"#),
whole: true,
},
0,
);
assert_eq!(s.query("gtfs-1", 0).unwrap().plus, json!([1]));
}
#[test]
fn staleness_is_by_the_clock() {
let mut s = State::new(["ankyra:home".into()], Duration::from_secs(60));
s.apply(
Event::Readings {
source: "ankyra:home".into(),
readings: readings(r#"{"gtfs-1": []}"#),
whole: true,
},
100,
);
assert!(!s.query("gtfs-1", 160).unwrap().stale);
assert!(s.query("gtfs-1", 161).unwrap().stale);
}
#[test]
fn a_status_that_says_nothing_new_changes_nothing() {
let mut s = State::new(["ankyra:home".into()], Duration::from_secs(60));
let ev = || Event::Status {
source: "ankyra:home".into(),
connected: true,
error: None,
};
assert!(s.apply(ev(), 0).sources);
assert_eq!(s.apply(ev(), 0), Change::default());
}
}
neiam
/apollos-dbus
public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages
⭐
Log in to mark this repository.
16.4 KiB · text
History
4f1f7f4