unifier/daemon/
notify.rs

1//! Outbound event socket: wake listeners (e.g. Jan cron) when mailboxes change.
2
3use std::io::{BufRead, BufReader, ErrorKind, Write};
4use std::os::unix::net::UnixStream;
5use std::sync::Mutex;
6use std::time::Duration;
7
8use serde::{Deserialize, Serialize};
9use uuid::Uuid;
10
11use crate::daemon::paths::events_socket_path;
12use crate::error::{Error, Result};
13use crate::home::UnifierHome;
14
15/// Line-delimited JSON notice written to `.daemon/events.sock`.
16///
17/// Downstream schedulers need at least `to` and `id` to locate the mailbox
18/// file at `mailbox/<to>/<id>.txt`.
19#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
20#[serde(tag = "kind", rename_all = "snake_case")]
21pub enum Notice {
22    Mailbox {
23        id: Uuid,
24        from: String,
25        to: String,
26    },
27    Event {
28        id: Uuid,
29        #[serde(default, skip_serializing_if = "Option::is_none")]
30        name: Option<String>,
31    },
32}
33
34impl Notice {
35    pub fn mailbox(id: Uuid, from: impl Into<String>, to: impl Into<String>) -> Self {
36        Self::Mailbox {
37            id,
38            from: from.into(),
39            to: to.into(),
40        }
41    }
42
43    pub fn event(id: Uuid, name: Option<String>) -> Self {
44        Self::Event { id, name }
45    }
46
47    pub fn from_json(line: &str) -> Result<Self> {
48        Ok(serde_json::from_str(line.trim())?)
49    }
50}
51
52/// Fan-out hub for connected event listeners.
53#[derive(Default)]
54pub struct EventHub {
55    subscribers: Mutex<Vec<UnixStream>>,
56}
57
58impl EventHub {
59    pub fn add(&self, stream: UnixStream) -> Result<()> {
60        stream.set_nonblocking(false)?;
61        stream.set_write_timeout(Some(Duration::from_millis(250)))?;
62        self.subscribers.lock().map_err(lock_err)?.push(stream);
63        Ok(())
64    }
65
66    pub fn broadcast(&self, notice: &Notice) {
67        let Ok(json) = serde_json::to_string(notice) else {
68            return;
69        };
70        let line = format!("{json}\n");
71        let Ok(mut subs) = self.subscribers.lock() else {
72            return;
73        };
74        subs.retain_mut(|stream| {
75            stream.write_all(line.as_bytes()).is_ok() && stream.flush().is_ok()
76        });
77    }
78}
79
80/// Connect to the daemon event socket for wakeup notices.
81pub fn subscribe(home: &UnifierHome) -> Result<UnixStream> {
82    let path = events_socket_path(home);
83    let stream = UnixStream::connect(&path).map_err(|e| {
84        Error::msg(format!(
85            "event socket not reachable at {}: {e}",
86            path.display()
87        ))
88    })?;
89    stream.set_nonblocking(false)?;
90    stream.set_read_timeout(None)?;
91    Ok(stream)
92}
93
94/// Print notices from the event socket until the daemon disconnects.
95pub fn watch(home: &UnifierHome) -> Result<()> {
96    let stream = subscribe(home)?;
97    let mut reader = BufReader::new(stream);
98    loop {
99        let mut line = String::new();
100        match reader.read_line(&mut line) {
101            Ok(0) => break,
102            Ok(_) => print!("{line}"),
103            Err(e)
104                if matches!(
105                    e.kind(),
106                    ErrorKind::Interrupted | ErrorKind::WouldBlock | ErrorKind::TimedOut
107                ) =>
108            {
109                continue
110            }
111            Err(e) => return Err(e.into()),
112        }
113    }
114    Ok(())
115}
116
117pub fn event_name(payload: &str) -> Option<String> {
118    let value: serde_json::Value = serde_json::from_str(payload).ok()?;
119    value.get("name")?.as_str().map(str::to_string)
120}
121
122fn lock_err<E: std::fmt::Display>(e: E) -> Error {
123    Error::msg(format!("event hub lock poisoned: {e}"))
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129    use std::io::BufRead;
130
131    #[test]
132    fn broadcast_reaches_paired_subscriber() {
133        let hub = EventHub::default();
134        let (tx, rx) = UnixStream::pair().unwrap();
135        hub.add(tx).unwrap();
136
137        let id = Uuid::new_v4();
138        hub.broadcast(&Notice::mailbox(id, "alice", "bob"));
139
140        let mut reader = BufReader::new(rx);
141        let mut line = String::new();
142        reader.read_line(&mut line).unwrap();
143        let notice = Notice::from_json(&line).unwrap();
144        assert_eq!(
145            notice,
146            Notice::Mailbox {
147                id,
148                from: "alice".into(),
149                to: "bob".into(),
150            }
151        );
152    }
153}
unifier/daemon/notify.rs

unifier/daemon/
notify.rs
