rtsp_runtime/
server.rs

1//! Server-side RTSP session engine — RFC 2326 Appendix A.2.
2//!
3//! [`ServerSession`] is sans-IO: [`ServerSession::handle_request`] parses an
4//! inbound request, validates it against the §A.2 server state table (see
5//! [`docs/state-machines.md`](../docs/state-machines.md)), and returns the
6//! serialized response bytes plus typed [`ServerEvent`]s. A method not valid in
7//! the current state yields a `455 Method Not Valid In This State`
8//! (see [`docs/methods-and-status.md`](../docs/methods-and-status.md)); state
9//! advances only when a `2xx` is actually sent.
10//!
11//! On `SETUP` the server allocates a `Session` id (if none yet) and echoes the
12//! client's `Transport` back (a minimal negotiation: the first offered spec is
13//! accepted, else `461 Unsupported Transport`).
14
15use rtsp_types::{Message, Method, Request, Response, StatusCode, Version, headers};
16
17use crate::error::{Error, Result};
18use crate::state::{SessionState, server_next_state};
19use crate::transport::Transport;
20
21/// A message body type: owned bytes.
22type Body = Vec<u8>;
23
24/// An event produced by [`ServerSession::handle_request`].
25#[non_exhaustive]
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub enum ServerEvent {
28    /// A request was accepted and the state machine advanced (2xx sent).
29    RequestAccepted {
30        /// The request method.
31        method: Method,
32        /// The `CSeq` echoed on the response.
33        cseq: u32,
34        /// The state after handling.
35        state: SessionState,
36    },
37    /// A request was rejected in the current state; a `455` was returned.
38    MethodNotValid {
39        /// The rejected method.
40        method: Method,
41        /// The state the request was rejected in.
42        state: SessionState,
43    },
44    /// A `SETUP` completed and a session id was allocated / reused.
45    SessionSetup {
46        /// The session id.
47        session_id: String,
48        /// The negotiated transport.
49        transport: Transport,
50    },
51}
52
53/// Arbitrary default seed for allocated RTSP `Session` ids; callers may
54/// override it with [`ServerSession::with_session_seed`]. Not
55/// security-sensitive — session ids are opaque handles.
56const DEFAULT_SESSION_SEED: u64 = 0x1234_5678;
57
58/// A driveable RTSP server session (RFC 2326 §A.2).
59#[derive(Debug)]
60pub struct ServerSession {
61    state: SessionState,
62    session_id: Option<String>,
63    session_timeout: Option<u64>,
64    next_session_seed: u64,
65    negotiated_transport: Option<Transport>,
66    server_header: String,
67}
68
69impl Default for ServerSession {
70    fn default() -> Self {
71        Self::new()
72    }
73}
74
75impl ServerSession {
76    /// Creates a fresh server session in the `Init` state.
77    pub fn new() -> Self {
78        ServerSession {
79            state: SessionState::Init,
80            session_id: None,
81            session_timeout: None,
82            next_session_seed: DEFAULT_SESSION_SEED,
83            negotiated_transport: None,
84            server_header: "rtsp-runtime".to_string(),
85        }
86    }
87
88    /// Sets the timeout (seconds) advertised in the `Session` header on SETUP.
89    pub fn with_session_timeout(mut self, seconds: u64) -> Self {
90        self.session_timeout = Some(seconds);
91        self
92    }
93
94    /// Overrides the seed used to allocate session ids (deterministic for tests).
95    pub fn with_session_seed(mut self, seed: u64) -> Self {
96        self.next_session_seed = seed;
97        self
98    }
99
100    /// The current session state.
101    pub fn state(&self) -> SessionState {
102        self.state
103    }
104
105    /// The allocated session id, once a SETUP has been handled.
106    pub fn session_id(&self) -> Option<&str> {
107        self.session_id.as_deref()
108    }
109
110    /// The negotiated transport, once a SETUP has been handled.
111    pub fn negotiated_transport(&self) -> Option<&Transport> {
112        self.negotiated_transport.as_ref()
113    }
114
115    /// Parses an inbound request and returns the serialized response bytes plus
116    /// the events produced.
117    pub fn handle_request(&mut self, data: &[u8]) -> Result<(Vec<u8>, Vec<ServerEvent>)> {
118        let (message, _consumed) =
119            Message::<Body>::parse(data).map_err(|e| Error::MessageParse(format!("{e:?}")))?;
120        let request = match message {
121            Message::Request(r) => r,
122            _ => return Err(Error::MessageParse("expected an RTSP request".into())),
123        };
124        self.handle_parsed(request)
125    }
126
127    fn handle_parsed(&mut self, request: Request<Body>) -> Result<(Vec<u8>, Vec<ServerEvent>)> {
128        let method = request.method().clone();
129        let cseq = header_value(request.header(&headers::CSEQ))
130            .and_then(|s| s.trim().parse::<u32>().ok())
131            .ok_or(Error::MissingCSeq)?;
132
133        // Validate the method against the current state (§A.2).
134        let next_state = match server_next_state(self.state, &method) {
135            Ok(s) => s,
136            Err(_) => {
137                let resp = self.build_response(StatusCode::MethodNotValidInThisState, cseq, |b| b);
138                let bytes = serialize(&Message::from(resp))?;
139                return Ok((
140                    bytes,
141                    vec![ServerEvent::MethodNotValid {
142                        method,
143                        state: self.state,
144                    }],
145                ));
146            }
147        };
148
149        let mut events = Vec::new();
150
151        // SETUP: allocate session + negotiate transport.
152        if method == Method::Setup {
153            let transport = match header_value(request.header(&headers::TRANSPORT)) {
154                Some(t) => Transport::parse(t)?,
155                None => {
156                    let resp = self.build_response(StatusCode::UnsupportedTransport, cseq, |b| b);
157                    return Ok((serialize(&Message::from(resp))?, events));
158                }
159            };
160            if transport.first().is_none() {
161                let resp = self.build_response(StatusCode::UnsupportedTransport, cseq, |b| b);
162                return Ok((serialize(&Message::from(resp))?, events));
163            }
164            let session_id = self
165                .session_id
166                .clone()
167                .unwrap_or_else(|| self.allocate_session());
168            self.session_id = Some(session_id.clone());
169            self.negotiated_transport = Some(transport.clone());
170
171            let sid = session_id.clone();
172            let session_hdr = match self.session_timeout {
173                Some(t) => format!("{sid};timeout={t}"),
174                None => sid.clone(),
175            };
176            let transport_hdr = transport.to_header_value();
177            let resp = self.build_response(StatusCode::Ok, cseq, |b| {
178                b.header(headers::SESSION, session_hdr)
179                    .header(headers::TRANSPORT, transport_hdr)
180            });
181            self.state = next_state;
182            events.push(ServerEvent::SessionSetup {
183                session_id,
184                transport,
185            });
186            events.push(ServerEvent::RequestAccepted {
187                method,
188                cseq,
189                state: self.state,
190            });
191            return Ok((serialize(&Message::from(resp))?, events));
192        }
193
194        // Other methods: echo Session if we have one, send 200, transition.
195        let session_hdr = self.session_id.clone();
196        let resp = self.build_response(StatusCode::Ok, cseq, |mut b| {
197            if let Some(sid) = &session_hdr {
198                b = b.header(headers::SESSION, sid.clone());
199            }
200            b
201        });
202        self.state = next_state;
203        if method == Method::Teardown {
204            self.session_id = None;
205            self.negotiated_transport = None;
206        }
207        events.push(ServerEvent::RequestAccepted {
208            method,
209            cseq,
210            state: self.state,
211        });
212        Ok((serialize(&Message::from(resp))?, events))
213    }
214
215    fn allocate_session(&mut self) -> String {
216        let id = self.next_session_seed;
217        // Advance deterministically for any subsequent allocation.
218        self.next_session_seed = self.next_session_seed.wrapping_add(1);
219        format!("{id:08}")
220    }
221
222    /// Builds a response with Version 1.0, CSeq, and Server headers, plus any
223    /// headers added by `f`.
224    fn build_response<F>(&self, status: StatusCode, cseq: u32, f: F) -> Response<Body>
225    where
226        F: FnOnce(rtsp_types::ResponseBuilder) -> rtsp_types::ResponseBuilder,
227    {
228        let builder = Response::builder(Version::V1_0, status)
229            .header(headers::CSEQ, cseq.to_string())
230            .header(headers::SERVER, self.server_header.clone());
231        f(builder).build(Vec::new())
232    }
233}
234
235fn header_value(h: Option<&headers::HeaderValue>) -> Option<&str> {
236    h.map(|v| v.as_str())
237}
238
239fn serialize(message: &Message<Body>) -> Result<Vec<u8>> {
240    let mut out = Vec::new();
241    message
242        .write(&mut out)
243        .map_err(|e| Error::MessageWrite(e.to_string()))?;
244    Ok(out)
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250
251    fn req(bytes: &str) -> Vec<u8> {
252        bytes.replace('\n', "\r\n").into_bytes()
253    }
254
255    #[test]
256    fn setup_transitions_init_to_ready() {
257        let mut s = ServerSession::new();
258        let (resp, events) = s
259            .handle_request(&req(
260                "SETUP rtsp://h/s RTSP/1.0\nCSeq: 1\nTransport: RTP/AVP/TCP;interleaved=0-1\n\n",
261            ))
262            .unwrap();
263        assert_eq!(s.state(), SessionState::Ready);
264        let text = String::from_utf8_lossy(&resp);
265        assert!(text.contains("200"));
266        assert!(text.contains("Session:"));
267        assert!(text.contains("Transport:"));
268        assert!(
269            events
270                .iter()
271                .any(|e| matches!(e, ServerEvent::SessionSetup { .. }))
272        );
273    }
274
275    #[test]
276    fn play_in_init_returns_455() {
277        let mut s = ServerSession::new();
278        let (resp, events) = s
279            .handle_request(&req("PLAY rtsp://h/s RTSP/1.0\nCSeq: 1\n\n"))
280            .unwrap();
281        assert_eq!(s.state(), SessionState::Init);
282        assert!(String::from_utf8_lossy(&resp).contains("455"));
283        assert!(
284            events
285                .iter()
286                .any(|e| matches!(e, ServerEvent::MethodNotValid { .. }))
287        );
288    }
289}
rtsp_runtime/server.rs

rtsp_runtime/
server.rs
