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    /// Number of currently registered event subscribers (not pruned).
80    pub fn subscriber_count(&self) -> usize {
81        self.subscribers
82            .lock()
83            .map(|s| s.len())
84            .unwrap_or(0)
85    }
86}
87
88/// Connect to the daemon event socket for wakeup notices.
89pub fn subscribe(home: &UnifierHome) -> Result<UnixStream> {
90    let path = events_socket_path(home);
91    let stream = UnixStream::connect(&path).map_err(|e| {
92        Error::msg(format!(
93            "event socket not reachable at {}: {e}",
94            path.display()
95        ))
96    })?;
97    stream.set_nonblocking(false)?;
98    stream.set_read_timeout(None)?;
99    Ok(stream)
100}
101
102/// Print notices from the event socket until the daemon disconnects.
103pub fn watch(home: &UnifierHome) -> Result<()> {
104    let stream = subscribe(home)?;
105    let mut reader = BufReader::new(stream);
106    loop {
107        let mut line = String::new();
108        match reader.read_line(&mut line) {
109            Ok(0) => break,
110            Ok(_) => print!("{line}"),
111            Err(e)
112                if matches!(
113                    e.kind(),
114                    ErrorKind::Interrupted | ErrorKind::WouldBlock | ErrorKind::TimedOut
115                ) =>
116            {
117                continue
118            }
119            Err(e) => return Err(e.into()),
120        }
121    }
122    Ok(())
123}
124
125pub fn event_name(payload: &str) -> Option<String> {
126    let value: serde_json::Value = serde_json::from_str(payload).ok()?;
127    value.get("name")?.as_str().map(str::to_string)
128}
129
130fn lock_err<E: std::fmt::Display>(e: E) -> Error {
131    Error::msg(format!("event hub lock poisoned: {e}"))
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    use std::io::BufRead;
138
139    #[test]
140    fn broadcast_reaches_paired_subscriber() {
141        let hub = EventHub::default();
142        let (tx, rx) = UnixStream::pair().unwrap();
143        hub.add(tx).unwrap();
144
145        let id = Uuid::new_v4();
146        hub.broadcast(&Notice::mailbox(id, "alice", "bob"));
147
148        let mut reader = BufReader::new(rx);
149        let mut line = String::new();
150        reader.read_line(&mut line).unwrap();
151        let notice = Notice::from_json(&line).unwrap();
152        assert_eq!(
153            notice,
154            Notice::Mailbox {
155                id,
156                from: "alice".into(),
157                to: "bob".into(),
158            }
159        );
160    }
161}
unifier/daemon/notify.rs

unifier/daemon/
notify.rs
