moqtap_client/draft16/
subscription.rs

1/// Subscription lifecycle states.
2#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3pub enum SubscriptionState {
4    /// Initial state before any SUBSCRIBE message is sent.
5    Idle,
6    /// SUBSCRIBE has been sent; awaiting OK or ERROR.
7    Subscribing,
8    /// Subscription is accepted and data may be flowing.
9    Active,
10    /// Subscription has ended (error, unsubscribe, or publish done).
11    Done,
12}
13
14/// Errors that can occur during subscription state transitions.
15#[derive(Debug, thiserror::Error, PartialEq, Eq)]
16pub enum SubscriptionError {
17    /// An event was received that is not valid for the current state.
18    #[error("invalid transition from {from:?} on event {event}")]
19    InvalidTransition {
20        /// The state the machine was in when the invalid event arrived.
21        from: SubscriptionState,
22        /// The name of the event that was rejected.
23        event: String,
24    },
25}
26
27/// Pure state machine for a MoQT subscription.
28/// Transitions: Idle -> Subscribing -> Active -> Done.
29pub struct SubscriptionStateMachine {
30    state: SubscriptionState,
31}
32
33impl Default for SubscriptionStateMachine {
34    fn default() -> Self {
35        Self::new()
36    }
37}
38
39impl SubscriptionStateMachine {
40    /// Creates a new state machine in the [`SubscriptionState::Idle`] state.
41    pub fn new() -> Self {
42        Self { state: SubscriptionState::Idle }
43    }
44
45    /// Returns the current state of the subscription.
46    pub fn state(&self) -> SubscriptionState {
47        self.state
48    }
49
50    /// Idle -> Subscribing (SUBSCRIBE sent).
51    pub fn on_subscribe_sent(&mut self) -> Result<(), SubscriptionError> {
52        if self.state == SubscriptionState::Idle {
53            self.state = SubscriptionState::Subscribing;
54            Ok(())
55        } else {
56            Err(SubscriptionError::InvalidTransition {
57                from: self.state,
58                event: "on_subscribe_sent".to_string(),
59            })
60        }
61    }
62
63    /// Subscribing -> Active (SUBSCRIBE_OK received).
64    pub fn on_subscribe_ok(&mut self) -> Result<(), SubscriptionError> {
65        if self.state == SubscriptionState::Subscribing {
66            self.state = SubscriptionState::Active;
67            Ok(())
68        } else {
69            Err(SubscriptionError::InvalidTransition {
70                from: self.state,
71                event: "on_subscribe_ok".to_string(),
72            })
73        }
74    }
75
76    /// Subscribing -> Done (REQUEST_ERROR received).
77    pub fn on_subscribe_error(&mut self) -> Result<(), SubscriptionError> {
78        if self.state == SubscriptionState::Subscribing {
79            self.state = SubscriptionState::Done;
80            Ok(())
81        } else {
82            Err(SubscriptionError::InvalidTransition {
83                from: self.state,
84                event: "on_subscribe_error".to_string(),
85            })
86        }
87    }
88
89    /// Active -> Done (UNSUBSCRIBE sent).
90    pub fn on_unsubscribe(&mut self) -> Result<(), SubscriptionError> {
91        if self.state == SubscriptionState::Active {
92            self.state = SubscriptionState::Done;
93            Ok(())
94        } else {
95            Err(SubscriptionError::InvalidTransition {
96                from: self.state,
97                event: "on_unsubscribe".to_string(),
98            })
99        }
100    }
101
102    /// REQUEST_UPDATE received -- a self-transition, from Subscribing as well as
103    /// from Active.
104    ///
105    /// Section 9.11 orders an update against the request rather than against
106    /// the request's answer: the sender of a SUBSCRIBE "can later send a
107    /// REQUEST_UPDATE to modify it", where later is later than the SUBSCRIBE.
108    /// The message names what it updates in its own Existing Request ID field,
109    /// which the SUBSCRIBE has already established.
110    ///
111    /// So a peer that sends SUBSCRIBE and REQUEST_UPDATE back to back breaks no
112    /// rule this draft states, and an update arriving before the answer leaves
113    /// the subscription where it found it. `Idle` and `Done` are still refused:
114    /// in neither does the subscription an update names exist.
115    pub fn on_subscribe_update(&mut self) -> Result<(), SubscriptionError> {
116        if matches!(self.state, SubscriptionState::Subscribing | SubscriptionState::Active) {
117            Ok(())
118        } else {
119            Err(SubscriptionError::InvalidTransition {
120                from: self.state,
121                event: "on_subscribe_update".to_string(),
122            })
123        }
124    }
125
126    /// Active -> Done (PUBLISH_DONE received).
127    pub fn on_publish_done(&mut self) -> Result<(), SubscriptionError> {
128        if self.state == SubscriptionState::Active {
129            self.state = SubscriptionState::Done;
130            Ok(())
131        } else {
132            Err(SubscriptionError::InvalidTransition {
133                from: self.state,
134                event: "on_publish_done".to_string(),
135            })
136        }
137    }
138}
139
140/// The same six transitions, named for the end that sees them.
141///
142/// A subscription this endpoint publishes runs through the states in the same
143/// order as one it subscribes to, with every message going the other way: the
144/// SUBSCRIBE arrives instead of leaving, the answer leaves instead of
145/// arriving. Sharing the transitions and not the names is what lets a refusal
146/// say which event was refused, rather than naming the mirror image of it.
147impl SubscriptionStateMachine {
148    /// Idle -> Subscribing (SUBSCRIBE received from a subscribing peer).
149    pub fn on_subscribe_received(&mut self) -> Result<(), SubscriptionError> {
150        self.on_subscribe_sent().map_err(|_| SubscriptionError::InvalidTransition {
151            from: self.state(),
152            event: "on_subscribe_received".to_string(),
153        })
154    }
155
156    /// Subscribing -> Active (SUBSCRIBE_OK sent to the subscribing peer).
157    pub fn on_subscribe_ok_sent(&mut self) -> Result<(), SubscriptionError> {
158        self.on_subscribe_ok().map_err(|_| SubscriptionError::InvalidTransition {
159            from: self.state(),
160            event: "on_subscribe_ok_sent".to_string(),
161        })
162    }
163
164    /// Subscribing -> Done (REQUEST_ERROR sent to the subscribing peer).
165    pub fn on_request_error_sent(&mut self) -> Result<(), SubscriptionError> {
166        self.on_subscribe_error().map_err(|_| SubscriptionError::InvalidTransition {
167            from: self.state(),
168            event: "on_request_error_sent".to_string(),
169        })
170    }
171
172    /// Active -> Done (UNSUBSCRIBE received from the subscribing peer).
173    pub fn on_unsubscribe_received(&mut self) -> Result<(), SubscriptionError> {
174        self.on_unsubscribe().map_err(|_| SubscriptionError::InvalidTransition {
175            from: self.state(),
176            event: "on_unsubscribe_received".to_string(),
177        })
178    }
179
180    /// Active -> Done (PUBLISH_DONE sent to the subscribing peer).
181    pub fn on_publish_done_sent(&mut self) -> Result<(), SubscriptionError> {
182        self.on_publish_done().map_err(|_| SubscriptionError::InvalidTransition {
183            from: self.state(),
184            event: "on_publish_done_sent".to_string(),
185        })
186    }
187
188    /// Subscribing or Active, unchanged (REQUEST_UPDATE received from the subscribing peer).
189    pub fn on_request_update_received(&mut self) -> Result<(), SubscriptionError> {
190        self.on_subscribe_update().map_err(|_| SubscriptionError::InvalidTransition {
191            from: self.state(),
192            event: "on_request_update_received".to_string(),
193        })
194    }
195}
moqtap_client/draft16/subscription.rs

moqtap_client/draft16/
subscription.rs
