io_jmap/rfc8620/
event_source.rs

1//! JMAP Event Source push channel (RFC 8620 §7.3 & §7.1).
2//!
3//! A streaming GET against
4//! [`JmapSession::event_source_url`](crate::rfc8620::session::JmapSession::event_source_url)
5//! yields W3C SSE frames carrying [`JmapStateChange`] payloads.
6
7use alloc::{
8    collections::BTreeMap,
9    string::{String, ToString},
10};
11
12use serde::{Deserialize, Serialize};
13use thiserror::Error;
14
15pub mod subscribe;
16
17/// Wire value of the `@type` property of a StateChange object.
18const DEFAULT_TYPE_TAG: &str = "StateChange";
19
20fn default_type_tag() -> String {
21    DEFAULT_TYPE_TAG.to_string()
22}
23
24/// Type-state map for one JMAP account, keyed by JMAP type name (`Email`,
25/// `Mailbox`, …); the value is the opaque state string. Callers diff it
26/// against their stored checkpoint and call `<Type>/changes` on a mismatch.
27pub type JmapTypeStates = BTreeMap<String, String>;
28
29/// JMAP StateChange push notification (RFC 8620 §7.1).
30///
31/// `changed` is keyed by account id, then JMAP type, then opaque new state.
32#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
33pub struct JmapStateChange {
34    /// The `@type` tag of the push object, always `StateChange`.
35    #[serde(rename = "@type", default = "default_type_tag")]
36    pub r#type: String,
37    /// The new type states, keyed by account id.
38    #[serde(default)]
39    pub changed: BTreeMap<String, JmapTypeStates>,
40}
41
42impl JmapStateChange {
43    /// Decodes one SSE frame's `data` field as a JMAP StateChange. Empty or
44    /// whitespace-only payloads return an empty `changed` map (keep-alive).
45    pub fn parse(data: &str) -> Result<Self, JmapStateChangeParseError> {
46        let trimmed = data.trim();
47        if trimmed.is_empty() {
48            return Ok(Self::default());
49        }
50
51        let change: Self = serde_json::from_str(trimmed)?;
52        if change.r#type != DEFAULT_TYPE_TAG {
53            return Err(JmapStateChangeParseError::UnexpectedType(change.r#type));
54        }
55
56        Ok(change)
57    }
58}
59
60/// Failure causes from [`JmapStateChange::parse`].
61#[derive(Debug, Error)]
62pub enum JmapStateChangeParseError {
63    /// The payload is not valid JSON.
64    #[error("Invalid StateChange JSON: {0}")]
65    InvalidJson(#[from] serde_json::Error),
66    /// The payload's `@type` tag is not `StateChange`.
67    #[error("Expected @type StateChange, got {0}")]
68    UnexpectedType(String),
69}
70
71/// JMAP EventSource `closeafter` query value (RFC 8620 §7.3): when the server
72/// closes the streaming response.
73#[derive(Clone, Copy, Debug)]
74pub enum JmapCloseAfter {
75    /// Never close: stream many [`JmapStateChange`] frames over one socket.
76    /// The socket is unavailable for parallel JMAP POSTs while the stream is
77    /// open.
78    No,
79    /// Close after the first [`JmapStateChange`]: frees the socket for
80    /// follow-up `*/changes` + `*/get` POSTs, then resubscribe (IMAP
81    /// IDLE-like pattern). Recommended for
82    /// [`JmapEventSource`](crate::rfc8620::event_source::subscribe::JmapEventSource).
83    State,
84}
85
86impl JmapCloseAfter {
87    pub(super) fn as_str(self) -> &'static str {
88        match self {
89            Self::No => "no",
90            Self::State => "state",
91        }
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use crate::rfc8620::event_source::*;
98
99    #[test]
100    fn parses_minimal_state_change() {
101        let json = r#"{"@type":"StateChange","changed":{"u1":{"Email":"s1"}}}"#;
102        let change = JmapStateChange::parse(json).unwrap();
103        assert_eq!(change.r#type, "StateChange");
104        assert_eq!(change.changed.len(), 1);
105        assert_eq!(change.changed["u1"]["Email"], "s1");
106    }
107
108    #[test]
109    fn parses_multi_account_multi_type() {
110        let json = r#"{
111            "@type": "StateChange",
112            "changed": {
113                "acc-a": {"Email": "e1", "Mailbox": "m1"},
114                "acc-b": {"Email": "e2"}
115            }
116        }"#;
117        let change = JmapStateChange::parse(json).unwrap();
118        assert_eq!(change.changed.len(), 2);
119        assert_eq!(change.changed["acc-a"]["Mailbox"], "m1");
120        assert_eq!(change.changed["acc-b"]["Email"], "e2");
121    }
122
123    #[test]
124    fn empty_data_is_keep_alive() {
125        let change = JmapStateChange::parse("").unwrap();
126        assert!(change.changed.is_empty());
127
128        let change = JmapStateChange::parse("   \n  ").unwrap();
129        assert!(change.changed.is_empty());
130    }
131
132    #[test]
133    fn wrong_type_field_rejected() {
134        let json = r#"{"@type":"NotAStateChange","changed":{}}"#;
135        match JmapStateChange::parse(json) {
136            Err(JmapStateChangeParseError::UnexpectedType(t)) => assert_eq!(t, "NotAStateChange"),
137            other => panic!("unexpected: {other:?}"),
138        }
139    }
140
141    #[test]
142    fn invalid_json_rejected() {
143        match JmapStateChange::parse("{not json") {
144            Err(JmapStateChangeParseError::InvalidJson(_)) => {}
145            other => panic!("unexpected: {other:?}"),
146        }
147    }
148
149    #[test]
150    fn missing_changed_field_defaults_to_empty() {
151        let json = r#"{"@type":"StateChange"}"#;
152        let change = JmapStateChange::parse(json).unwrap();
153        assert!(change.changed.is_empty());
154    }
155}
io_jmap/rfc8620/event_source.rs

io_jmap/rfc8620/
event_source.rs
