moqtap_client/draft08/
namespace.rs

1/// SUBSCRIBE_ANNOUNCES lifecycle states (draft-08).
2#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3pub enum SubscribeAnnouncesState {
4    /// Initial state before any message is sent.
5    Idle,
6    /// SUBSCRIBE_ANNOUNCES has been sent; awaiting OK or ERROR.
7    Pending,
8    /// Namespace subscription is accepted and active.
9    Active,
10    /// Namespace subscription has ended.
11    Done,
12}
13
14/// ANNOUNCE lifecycle states (draft-08).
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum AnnounceState {
17    /// Initial state before any message is sent.
18    Idle,
19    /// ANNOUNCE has been sent; awaiting ANNOUNCE_OK or ANNOUNCE_ERROR.
20    Pending,
21    /// Namespace publication is accepted and active.
22    Active,
23    /// Namespace publication has ended (UNANNOUNCE or cancel).
24    Done,
25}
26
27/// Errors that can occur during namespace state transitions.
28#[derive(Debug, thiserror::Error, PartialEq, Eq)]
29pub enum NamespaceError {
30    /// An event was received that is not valid for the current state.
31    #[error("invalid transition from {from} on event {event}")]
32    InvalidTransition {
33        /// The state the machine was in when the invalid event arrived.
34        from: String,
35        /// The name of the event that was rejected.
36        event: String,
37    },
38}
39
40/// State machine for the SUBSCRIBE_ANNOUNCES flow (draft-08).
41/// Idle → Pending → Active → Done.
42pub struct SubscribeAnnouncesStateMachine {
43    state: SubscribeAnnouncesState,
44}
45
46impl Default for SubscribeAnnouncesStateMachine {
47    fn default() -> Self {
48        Self::new()
49    }
50}
51
52impl SubscribeAnnouncesStateMachine {
53    /// Creates a new machine in [`SubscribeAnnouncesState::Idle`].
54    pub fn new() -> Self {
55        Self { state: SubscribeAnnouncesState::Idle }
56    }
57
58    /// Returns the current state.
59    pub fn state(&self) -> SubscribeAnnouncesState {
60        self.state
61    }
62
63    /// Idle → Pending.
64    pub fn on_subscribe_announces_sent(&mut self) -> Result<(), NamespaceError> {
65        if self.state == SubscribeAnnouncesState::Idle {
66            self.state = SubscribeAnnouncesState::Pending;
67            Ok(())
68        } else {
69            Err(NamespaceError::InvalidTransition {
70                from: format!("{:?}", self.state),
71                event: "on_subscribe_announces_sent".to_string(),
72            })
73        }
74    }
75
76    /// Pending → Active.
77    pub fn on_subscribe_announces_ok(&mut self) -> Result<(), NamespaceError> {
78        if self.state == SubscribeAnnouncesState::Pending {
79            self.state = SubscribeAnnouncesState::Active;
80            Ok(())
81        } else {
82            Err(NamespaceError::InvalidTransition {
83                from: format!("{:?}", self.state),
84                event: "on_subscribe_announces_ok".to_string(),
85            })
86        }
87    }
88
89    /// Pending → Done.
90    pub fn on_subscribe_announces_error(&mut self) -> Result<(), NamespaceError> {
91        if self.state == SubscribeAnnouncesState::Pending {
92            self.state = SubscribeAnnouncesState::Done;
93            Ok(())
94        } else {
95            Err(NamespaceError::InvalidTransition {
96                from: format!("{:?}", self.state),
97                event: "on_subscribe_announces_error".to_string(),
98            })
99        }
100    }
101
102    /// Active → Done (UNSUBSCRIBE_ANNOUNCES sent).
103    pub fn on_unsubscribe_announces(&mut self) -> Result<(), NamespaceError> {
104        if self.state == SubscribeAnnouncesState::Active {
105            self.state = SubscribeAnnouncesState::Done;
106            Ok(())
107        } else {
108            Err(NamespaceError::InvalidTransition {
109                from: format!("{:?}", self.state),
110                event: "on_unsubscribe_announces".to_string(),
111            })
112        }
113    }
114
115    /// Idle → Pending (a SUBSCRIBE_ANNOUNCES arrived from the peer).
116    ///
117    /// The mirror of
118    /// [`on_subscribe_announces_sent`](Self::on_subscribe_announces_sent),
119    /// named for the direction it runs in rather than shared with it: the two
120    /// state edges coincide, so a message dispatched to the wrong one of them
121    /// would move the record silently instead of naming the event it was not.
122    pub fn on_subscribe_announces_received(&mut self) -> Result<(), NamespaceError> {
123        self.on_subscribe_announces_sent().map_err(|_| NamespaceError::InvalidTransition {
124            from: format!("{:?}", self.state()),
125            event: "on_subscribe_announces_received".to_string(),
126        })
127    }
128
129    /// Pending → Active (this endpoint accepted the peer's request with a
130    /// SUBSCRIBE_ANNOUNCES_OK).
131    ///
132    /// Section 4.1: "A publisher MUST send exactly one SUBSCRIBE_ANNOUNCES_OK
133    /// or SUBSCRIBE_ANNOUNCES_ERROR in response to a SUBSCRIBE_ANNOUNCES."
134    ///
135    /// One answer and no second one: the record leaves Pending on the first,
136    /// and a second call finds it somewhere else.
137    pub fn on_subscribe_announces_ok_sent(&mut self) -> Result<(), NamespaceError> {
138        self.on_subscribe_announces_ok().map_err(|_| NamespaceError::InvalidTransition {
139            from: format!("{:?}", self.state()),
140            event: "on_subscribe_announces_ok_sent".to_string(),
141        })
142    }
143
144    /// Pending → Done (this endpoint refused the peer's request with a
145    /// SUBSCRIBE_ANNOUNCES_ERROR).
146    ///
147    /// The other half of the same sentence: one message back, and this is the
148    /// other one it can be.
149    pub fn on_subscribe_announces_error_sent(&mut self) -> Result<(), NamespaceError> {
150        self.on_subscribe_announces_error().map_err(|_| NamespaceError::InvalidTransition {
151            from: format!("{:?}", self.state()),
152            event: "on_subscribe_announces_error_sent".to_string(),
153        })
154    }
155
156    /// Active → Done (the peer withdrew the namespace subscription with an
157    /// UNSUBSCRIBE_ANNOUNCES).
158    ///
159    /// Section 7.14: "A subscriber issues a UNSUBSCRIBE_ANNOUNCES message to
160    /// a publisher indicating it is no longer interested in ANNOUNCE and
161    /// UNANNOUNCE messages for the specified track namespace prefix."
162    ///
163    /// Active is the acceptance, which is the state a namespace subscription
164    /// reaches by being answered SUBSCRIBE_ANNOUNCES_OK and no other way, so
165    /// a withdrawal of one never answered is refused here.
166    pub fn on_unsubscribe_announces_received(&mut self) -> Result<(), NamespaceError> {
167        self.on_unsubscribe_announces().map_err(|_| NamespaceError::InvalidTransition {
168            from: format!("{:?}", self.state()),
169            event: "on_unsubscribe_announces_received".to_string(),
170        })
171    }
172}
173
174/// State machine for the ANNOUNCE flow (draft-08).
175/// Idle → Pending → Active → Done.
176pub struct AnnounceStateMachine {
177    state: AnnounceState,
178}
179
180impl Default for AnnounceStateMachine {
181    fn default() -> Self {
182        Self::new()
183    }
184}
185
186impl AnnounceStateMachine {
187    /// Creates a new machine in [`AnnounceState::Idle`].
188    pub fn new() -> Self {
189        Self { state: AnnounceState::Idle }
190    }
191
192    /// Returns the current state.
193    pub fn state(&self) -> AnnounceState {
194        self.state
195    }
196
197    /// Idle → Pending (ANNOUNCE sent).
198    pub fn on_announce_sent(&mut self) -> Result<(), NamespaceError> {
199        if self.state == AnnounceState::Idle {
200            self.state = AnnounceState::Pending;
201            Ok(())
202        } else {
203            Err(NamespaceError::InvalidTransition {
204                from: format!("{:?}", self.state),
205                event: "on_announce_sent".to_string(),
206            })
207        }
208    }
209
210    /// Pending → Active (ANNOUNCE_OK received).
211    pub fn on_announce_ok(&mut self) -> Result<(), NamespaceError> {
212        if self.state == AnnounceState::Pending {
213            self.state = AnnounceState::Active;
214            Ok(())
215        } else {
216            Err(NamespaceError::InvalidTransition {
217                from: format!("{:?}", self.state),
218                event: "on_announce_ok".to_string(),
219            })
220        }
221    }
222
223    /// Pending → Done (ANNOUNCE_ERROR received).
224    pub fn on_announce_error(&mut self) -> Result<(), NamespaceError> {
225        if self.state == AnnounceState::Pending {
226            self.state = AnnounceState::Done;
227            Ok(())
228        } else {
229            Err(NamespaceError::InvalidTransition {
230                from: format!("{:?}", self.state),
231                event: "on_announce_error".to_string(),
232            })
233        }
234    }
235
236    /// Active → Done (UNANNOUNCE sent — publisher withdrawing).
237    pub fn on_unannounce(&mut self) -> Result<(), NamespaceError> {
238        if self.state == AnnounceState::Active {
239            self.state = AnnounceState::Done;
240            Ok(())
241        } else {
242            Err(NamespaceError::InvalidTransition {
243                from: format!("{:?}", self.state),
244                event: "on_unannounce".to_string(),
245            })
246        }
247    }
248
249    /// Active → Done (ANNOUNCE_CANCEL received — subscriber cancelling).
250    pub fn on_announce_cancel(&mut self) -> Result<(), NamespaceError> {
251        if self.state == AnnounceState::Active {
252            self.state = AnnounceState::Done;
253            Ok(())
254        } else {
255            Err(NamespaceError::InvalidTransition {
256                from: format!("{:?}", self.state),
257                event: "on_announce_cancel".to_string(),
258            })
259        }
260    }
261}
262
263/// The same transitions, named for the end the advertisement arrives at.
264///
265/// An announcement this endpoint accepts passes through the states in the same
266/// order as one it makes, with every message going the other way: the ANNOUNCE
267/// arrives instead of leaving, the answer leaves instead of arriving, the
268/// withdrawal arrives and the cancellation leaves. Sharing the transitions and
269/// not the names is what lets a refusal say which event was refused, rather
270/// than naming the mirror image of it.
271impl AnnounceStateMachine {
272    /// Idle -> Pending (ANNOUNCE received from the peer).
273    pub fn on_announce_received(&mut self) -> Result<(), NamespaceError> {
274        self.on_announce_sent().map_err(|_| NamespaceError::InvalidTransition {
275            from: format!("{:?}", self.state()),
276            event: "on_announce_received".to_string(),
277        })
278    }
279
280    /// Pending -> Active (ANNOUNCE_OK sent, accepting the announcement).
281    pub fn on_announce_ok_sent(&mut self) -> Result<(), NamespaceError> {
282        self.on_announce_ok().map_err(|_| NamespaceError::InvalidTransition {
283            from: format!("{:?}", self.state()),
284            event: "on_announce_ok_sent".to_string(),
285        })
286    }
287
288    /// Pending -> Done (ANNOUNCE_ERROR sent, refusing the announcement).
289    pub fn on_announce_error_sent(&mut self) -> Result<(), NamespaceError> {
290        self.on_announce_error().map_err(|_| NamespaceError::InvalidTransition {
291            from: format!("{:?}", self.state()),
292            event: "on_announce_error_sent".to_string(),
293        })
294    }
295
296    /// Active -> Done (UNANNOUNCE received, the peer withdrawing).
297    pub fn on_unannounce_received(&mut self) -> Result<(), NamespaceError> {
298        self.on_unannounce().map_err(|_| NamespaceError::InvalidTransition {
299            from: format!("{:?}", self.state()),
300            event: "on_unannounce_received".to_string(),
301        })
302    }
303
304    /// Active -> Done (ANNOUNCE_CANCEL sent, revoking an acceptance).
305    ///
306    /// Active is the acceptance: it is the state an announcement reaches by
307    /// being answered ANNOUNCE_OK and no other way, which is why a cancellation
308    /// of one never answered is refused here rather than sent.
309    pub fn on_announce_cancel_sent(&mut self) -> Result<(), NamespaceError> {
310        self.on_announce_cancel().map_err(|_| NamespaceError::InvalidTransition {
311            from: format!("{:?}", self.state()),
312            event: "on_announce_cancel_sent".to_string(),
313        })
314    }
315}
moqtap_client/draft08/namespace.rs

moqtap_client/draft08/
namespace.rs
