kernoid_client/
push.rs

1// SPDX-License-Identifier: MIT
2//! Push over EventSource (RFC 8620, 7.3): a task that listens to the stream and tells which types changed; after a
3//! break it connects again, waiting longer each time (up to a minute).
4
5use std::{collections::BTreeSet, time::Duration};
6
7use serde_json::Value;
8use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader};
9
10use crate::Client;
11
12/// What the stream tells.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub enum Push {
15    /// The types whose state changed, in any account (`Email`, `Mailbox`, `FileNode` …).
16    Changed(BTreeSet<String>),
17    /// The stream broke; it is being opened again.
18    Lost(String),
19    /// The stream is open (again).
20    Connected,
21}
22
23/// The events of an `text/event-stream` (HTML, 9.2): name and data of each, as they arrive.
24#[derive(Debug)]
25pub struct Events<R> {
26    reader: BufReader<R>,
27}
28
29impl<R: AsyncRead + Unpin> Events<R> {
30    pub fn new(reader: R) -> Self {
31        Self { reader: BufReader::new(reader) }
32    }
33
34    /// The next event; `None` when the stream ends.
35    pub async fn next(&mut self) -> Option<std::io::Result<(String, String)>> {
36        let (mut name, mut data) = (String::new(), Vec::<String>::new());
37        let mut line = String::new();
38        loop {
39            line.clear();
40            match self.reader.read_line(&mut line).await {
41                Ok(0) => return None,
42                Ok(_) => {}
43                Err(error) => return Some(Err(error)),
44            }
45            let line = line.trim_end_matches(['\r', '\n']);
46            if line.is_empty() {
47                if data.is_empty() && name.is_empty() {
48                    continue;
49                }
50                let name = if name.is_empty() { "message".to_owned() } else { std::mem::take(&mut name) };
51                return Some(Ok((name, data.join("\n"))));
52            }
53            if line.starts_with(':') {
54                continue;
55            }
56            let (field, value) = line.split_once(':').unwrap_or((line, ""));
57            let value = value.strip_prefix(' ').unwrap_or(value);
58            match field {
59                "event" => name = value.to_owned(),
60                "data" => data.push(value.to_owned()),
61                _ => {}
62            }
63        }
64    }
65}
66
67/// The types a `StateChange` names (RFC 8620, 7.1).
68pub fn changed_types(data: &str) -> BTreeSet<String> {
69    let Ok(value) = serde_json::from_str::<Value>(data) else { return BTreeSet::new() };
70    value
71        .get("changed")
72        .and_then(Value::as_object)
73        .into_iter()
74        .flat_map(|accounts| accounts.values())
75        .filter_map(Value::as_object)
76        .flat_map(|types| types.keys().cloned())
77        .collect()
78}
79
80/// Listens in a task of its own (on the current tokio runtime) until `out` returns false.
81pub fn listen(client: Client, out: impl Fn(Push) -> bool + Send + 'static) -> tokio::task::JoinHandle<()> {
82    tokio::spawn(async move {
83        let mut wait = Duration::from_secs(2);
84        loop {
85            match client.event_source(60).await {
86                Ok(stream) => {
87                    if !out(Push::Connected) {
88                        return;
89                    }
90                    wait = Duration::from_secs(2);
91                    let mut events = Events::new(tokio_util::io::StreamReader::new(stream));
92                    while let Some(event) = events.next().await {
93                        let Ok((name, data)) = event else { break };
94                        if name != "state" {
95                            continue;
96                        }
97                        let types = changed_types(&data);
98                        if !types.is_empty() && !out(Push::Changed(types)) {
99                            return;
100                        }
101                    }
102                    if !out(Push::Lost("the stream ended".into())) {
103                        return;
104                    }
105                }
106                Err(error) => {
107                    if !out(Push::Lost(error.to_string())) {
108                        return;
109                    }
110                }
111            }
112            tokio::time::sleep(wait).await;
113            wait = (wait * 2).min(Duration::from_secs(60));
114        }
115    })
116}
117
118#[cfg(test)]
119mod tests {
120    use super::*;
121
122    #[tokio::test]
123    async fn events() {
124        let stream = ": hello\n\nevent: state\ndata: {\"@type\":\"StateChange\",\ndata: \"changed\":{\"a\":{\"Email\":\"1\",\"Mailbox\":\"2\"}}}\n\nevent: ping\ndata: {\"interval\":60}\n\n";
125        let mut reader = Events::new(stream.as_bytes());
126        let mut events = Vec::new();
127        while let Some(event) = reader.next().await {
128            events.push(event.unwrap());
129        }
130        assert_eq!(events.len(), 2);
131        assert_eq!(events[0].0, "state");
132        let types: Vec<_> = changed_types(&events[0].1).into_iter().collect();
133        assert_eq!(types, ["Email", "Mailbox"]);
134        assert_eq!(events[1].0, "ping");
135        assert!(changed_types(&events[1].1).is_empty());
136    }
137}
kernoid_client/push.rs

kernoid_client/
push.rs
