moqtap_client/draft16/
namespace.rs

1/// SUBSCRIBE_NAMESPACE lifecycle states.
2#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3pub enum SubscribeNamespaceState {
4    /// Initial state before any message is sent.
5    Idle,
6    /// SUBSCRIBE_NAMESPACE 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/// PUBLISH_NAMESPACE lifecycle states.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum PublishNamespaceState {
17    /// Initial state before any message is sent.
18    Idle,
19    /// PUBLISH_NAMESPACE has been sent; awaiting OK or ERROR.
20    Pending,
21    /// Namespace publication is accepted and active.
22    Active,
23    /// Namespace publication has ended.
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 SUBSCRIBE_NAMESPACE flow.
41/// Idle -> Pending -> Active -> Done.
42pub struct SubscribeNamespaceStateMachine {
43    state: SubscribeNamespaceState,
44}
45
46impl Default for SubscribeNamespaceStateMachine {
47    fn default() -> Self {
48        Self::new()
49    }
50}
51
52impl SubscribeNamespaceStateMachine {
53    /// Creates a new machine in [`SubscribeNamespaceState::Idle`].
54    pub fn new() -> Self {
55        Self { state: SubscribeNamespaceState::Idle }
56    }
57
58    /// Returns the current state of the subscribe-namespace flow.
59    pub fn state(&self) -> SubscribeNamespaceState {
60        self.state
61    }
62
63    /// Idle -> Pending.
64    pub fn on_subscribe_namespace_sent(&mut self) -> Result<(), NamespaceError> {
65        if self.state == SubscribeNamespaceState::Idle {
66            self.state = SubscribeNamespaceState::Pending;
67            Ok(())
68        } else {
69            Err(NamespaceError::InvalidTransition {
70                from: format!("{:?}", self.state),
71                event: "on_subscribe_namespace_sent".to_string(),
72            })
73        }
74    }
75
76    /// Pending -> Active.
77    pub fn on_subscribe_namespace_ok(&mut self) -> Result<(), NamespaceError> {
78        if self.state == SubscribeNamespaceState::Pending {
79            self.state = SubscribeNamespaceState::Active;
80            Ok(())
81        } else {
82            Err(NamespaceError::InvalidTransition {
83                from: format!("{:?}", self.state),
84                event: "on_subscribe_namespace_ok".to_string(),
85            })
86        }
87    }
88
89    /// Pending -> Done.
90    pub fn on_subscribe_namespace_error(&mut self) -> Result<(), NamespaceError> {
91        if self.state == SubscribeNamespaceState::Pending {
92            self.state = SubscribeNamespaceState::Done;
93            Ok(())
94        } else {
95            Err(NamespaceError::InvalidTransition {
96                from: format!("{:?}", self.state),
97                event: "on_subscribe_namespace_error".to_string(),
98            })
99        }
100    }
101
102    /// Pending | Active -> Done, Done -> Done (this subscription's stream was
103    /// closed).
104    ///
105    /// Draft-16 has no UNSUBSCRIBE_NAMESPACE. The subscription is withdrawn by
106    /// ending the stream it was made on — Section 6.1: "A SUBSCRIBE_NAMESPACE
107    /// can be cancelled by closing the stream with either a FIN or
108    /// RESET_STREAM." Either form is a withdrawal, so both arrive here.
109    ///
110    /// `Pending` is accepted because the stream is open from the
111    /// SUBSCRIBE_NAMESPACE that opened it, and the sentence asks for nothing
112    /// more: a namespace subscription can be withdrawn before it is ever
113    /// answered.
114    ///
115    /// `Idle` is refused, on the other half of the same sentence: nothing has
116    /// been written, so there is no stream to close. `Done` stays `Done` — a
117    /// subscription that has already ended is still carried on a stream, and
118    /// closing that stream afterwards is the ordinary end rather than a fault.
119    pub fn on_request_cancelled(&mut self) -> Result<(), NamespaceError> {
120        match self.state {
121            SubscribeNamespaceState::Pending | SubscribeNamespaceState::Active => {
122                self.state = SubscribeNamespaceState::Done;
123                Ok(())
124            }
125            SubscribeNamespaceState::Done => Ok(()),
126            SubscribeNamespaceState::Idle => Err(NamespaceError::InvalidTransition {
127                from: format!("{:?}", self.state),
128                event: "on_request_cancelled".to_string(),
129            }),
130        }
131    }
132
133    /// Idle -> Pending (the peer opened a stream with a SUBSCRIBE_NAMESPACE on
134    /// it).
135    ///
136    /// The mirror of [`on_subscribe_namespace_sent`](Self::on_subscribe_namespace_sent),
137    /// named for the direction it runs in rather than sharing that one: the
138    /// state edges coincide, so a mis-dispatch would succeed silently instead
139    /// of naming the wrong event in an `InvalidTransition`.
140    pub fn on_subscribe_namespace_received(&mut self) -> Result<(), NamespaceError> {
141        if self.state == SubscribeNamespaceState::Idle {
142            self.state = SubscribeNamespaceState::Pending;
143            Ok(())
144        } else {
145            Err(NamespaceError::InvalidTransition {
146                from: format!("{:?}", self.state),
147                event: "on_subscribe_namespace_received".to_string(),
148            })
149        }
150    }
151
152    /// Pending -> Active (this endpoint answered the peer's request with a
153    /// REQUEST_OK).
154    pub fn on_subscribe_namespace_ok_sent(&mut self) -> Result<(), NamespaceError> {
155        if self.state == SubscribeNamespaceState::Pending {
156            self.state = SubscribeNamespaceState::Active;
157            Ok(())
158        } else {
159            Err(NamespaceError::InvalidTransition {
160                from: format!("{:?}", self.state),
161                event: "on_subscribe_namespace_ok_sent".to_string(),
162            })
163        }
164    }
165
166    /// Pending -> Done (this endpoint refused the peer's request).
167    pub fn on_subscribe_namespace_error_sent(&mut self) -> Result<(), NamespaceError> {
168        if self.state == SubscribeNamespaceState::Pending {
169            self.state = SubscribeNamespaceState::Done;
170            Ok(())
171        } else {
172            Err(NamespaceError::InvalidTransition {
173                from: format!("{:?}", self.state),
174                event: "on_subscribe_namespace_error_sent".to_string(),
175            })
176        }
177    }
178}
179
180/// State machine for PUBLISH_NAMESPACE flow.
181/// Idle -> Pending -> Active -> Done.
182pub struct PublishNamespaceStateMachine {
183    state: PublishNamespaceState,
184}
185
186impl Default for PublishNamespaceStateMachine {
187    fn default() -> Self {
188        Self::new()
189    }
190}
191
192impl PublishNamespaceStateMachine {
193    /// Creates a new machine in [`PublishNamespaceState::Idle`].
194    pub fn new() -> Self {
195        Self { state: PublishNamespaceState::Idle }
196    }
197
198    /// Returns the current state of the publish-namespace flow.
199    pub fn state(&self) -> PublishNamespaceState {
200        self.state
201    }
202
203    /// Idle -> Pending.
204    pub fn on_publish_namespace_sent(&mut self) -> Result<(), NamespaceError> {
205        if self.state == PublishNamespaceState::Idle {
206            self.state = PublishNamespaceState::Pending;
207            Ok(())
208        } else {
209            Err(NamespaceError::InvalidTransition {
210                from: format!("{:?}", self.state),
211                event: "on_publish_namespace_sent".to_string(),
212            })
213        }
214    }
215
216    /// Pending -> Active.
217    pub fn on_publish_namespace_ok(&mut self) -> Result<(), NamespaceError> {
218        if self.state == PublishNamespaceState::Pending {
219            self.state = PublishNamespaceState::Active;
220            Ok(())
221        } else {
222            Err(NamespaceError::InvalidTransition {
223                from: format!("{:?}", self.state),
224                event: "on_publish_namespace_ok".to_string(),
225            })
226        }
227    }
228
229    /// Pending -> Done.
230    pub fn on_publish_namespace_error(&mut self) -> Result<(), NamespaceError> {
231        if self.state == PublishNamespaceState::Pending {
232            self.state = PublishNamespaceState::Done;
233            Ok(())
234        } else {
235            Err(NamespaceError::InvalidTransition {
236                from: format!("{:?}", self.state),
237                event: "on_publish_namespace_error".to_string(),
238            })
239        }
240    }
241
242    /// Active -> Done (publisher withdrawing).
243    pub fn on_publish_namespace_done(&mut self) -> Result<(), NamespaceError> {
244        if self.state == PublishNamespaceState::Active {
245            self.state = PublishNamespaceState::Done;
246            Ok(())
247        } else {
248            Err(NamespaceError::InvalidTransition {
249                from: format!("{:?}", self.state),
250                event: "on_publish_namespace_done".to_string(),
251            })
252        }
253    }
254
255    /// Active -> Done (subscriber cancelling).
256    pub fn on_publish_namespace_cancel(&mut self) -> Result<(), NamespaceError> {
257        if self.state == PublishNamespaceState::Active {
258            self.state = PublishNamespaceState::Done;
259            Ok(())
260        } else {
261            Err(NamespaceError::InvalidTransition {
262                from: format!("{:?}", self.state),
263                event: "on_publish_namespace_cancel".to_string(),
264            })
265        }
266    }
267}
268
269/// The same transitions, named for the end the advertisement arrives at.
270///
271/// An announcement this endpoint accepts passes through the states in the same
272/// order as one it makes, with every message going the other way: the PUBLISH_NAMESPACE
273/// arrives instead of leaving, the answer leaves instead of arriving, the
274/// withdrawal arrives and the cancellation leaves. Sharing the transitions and
275/// not the names is what lets a refusal say which event was refused, rather
276/// than naming the mirror image of it.
277impl PublishNamespaceStateMachine {
278    /// Idle -> Pending (PUBLISH_NAMESPACE received from the peer).
279    pub fn on_publish_namespace_received(&mut self) -> Result<(), NamespaceError> {
280        self.on_publish_namespace_sent().map_err(|_| NamespaceError::InvalidTransition {
281            from: format!("{:?}", self.state()),
282            event: "on_publish_namespace_received".to_string(),
283        })
284    }
285
286    /// Pending -> Active (REQUEST_OK sent, accepting the announcement).
287    pub fn on_publish_namespace_ok_sent(&mut self) -> Result<(), NamespaceError> {
288        self.on_publish_namespace_ok().map_err(|_| NamespaceError::InvalidTransition {
289            from: format!("{:?}", self.state()),
290            event: "on_publish_namespace_ok_sent".to_string(),
291        })
292    }
293
294    /// Pending -> Done (REQUEST_ERROR sent, refusing the announcement).
295    pub fn on_publish_namespace_error_sent(&mut self) -> Result<(), NamespaceError> {
296        self.on_publish_namespace_error().map_err(|_| NamespaceError::InvalidTransition {
297            from: format!("{:?}", self.state()),
298            event: "on_publish_namespace_error_sent".to_string(),
299        })
300    }
301
302    /// Active -> Done (PUBLISH_NAMESPACE_DONE received, the peer withdrawing).
303    pub fn on_publish_namespace_done_received(&mut self) -> Result<(), NamespaceError> {
304        self.on_publish_namespace_done().map_err(|_| NamespaceError::InvalidTransition {
305            from: format!("{:?}", self.state()),
306            event: "on_publish_namespace_done_received".to_string(),
307        })
308    }
309
310    /// Active -> Done (PUBLISH_NAMESPACE_CANCEL sent, revoking an acceptance).
311    ///
312    /// Active is the acceptance: it is the state an announcement reaches by
313    /// being answered REQUEST_OK and no other way, which is why a cancellation
314    /// of one never answered is refused here rather than sent.
315    pub fn on_publish_namespace_cancel_sent(&mut self) -> Result<(), NamespaceError> {
316        self.on_publish_namespace_cancel().map_err(|_| NamespaceError::InvalidTransition {
317            from: format!("{:?}", self.state()),
318            event: "on_publish_namespace_cancel_sent".to_string(),
319        })
320    }
321}
moqtap_client/draft16/namespace.rs

moqtap_client/draft16/
namespace.rs
