netget/client/rtsp/
mod.rs

1//! RTSP client: one control connection to a media server, requests made on the handler's
2//! say, UDP RTP/RTCP port pairs bound per SETUP, and the media that arrives summarised per
3//! stream by the RTP client's tracker.
4pub mod actions;
5use crate::client::llm_budget::call_llm_for_client;
6use crate::client::rtp::Tracker;
7use crate::llm::actions::client_trait::{Client, ClientActionResult};
8use crate::logging::emit::Log;
9use crate::protocol::{ConnectContext, Event};
10use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
11use crate::state::{AccessLogOwner, ClientStatus};
12pub use actions::RtspClientProtocol;
13use anyhow::{bail, Context, Result};
14use serde_json::{json, Map, Value};
15use std::collections::HashMap;
16use std::net::{IpAddr, SocketAddr};
17use std::time::Duration;
18use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, WriteHalf};
19use tokio::net::{TcpStream, UdpSocket};
20use tokio::sync::mpsc;
21
22/// A handler chain (answer → event → answer …) stops after this many follow-ups.
23pub const MAX_FOLLOWUP_DEPTH: usize = 8;
24/// A response's status line and headers at most.
25pub const MAX_HEADER_BYTES: usize = 64 * 1024;
26/// A response body (an SDP) at most.
27pub const MAX_BODY_BYTES: usize = 1024 * 1024;
28/// Requests awaiting an answer at once.
29pub const MAX_PENDING: usize = 16;
30/// Tracks one session may set up.
31pub const MAX_TRACKS: usize = 8;
32pub const IO_TIMEOUT: Duration = Duration::from_secs(30);
33const USER_AGENT: &str = concat!("NetGet/", env!("CARGO_PKG_VERSION"));
34
35pub struct Response {
36    pub status: u16,
37    pub reason: String,
38    pub headers: Vec<(String, String)>,
39    pub body: Vec<u8>,
40}
41
42impl Response {
43    pub fn header(&self, name: &str) -> Option<&str> {
44        self.headers
45            .iter()
46            .find(|(k, _)| k.eq_ignore_ascii_case(name))
47            .map(|(_, v)| v.as_str())
48    }
49}
50
51/// Read one RTSP response, bounded in header and body size.
52pub async fn read_response<R: tokio::io::AsyncBufRead + Unpin>(
53    r: &mut R,
54) -> Result<Option<Response>> {
55    let mut head = Vec::new();
56    loop {
57        let mut line = Vec::new();
58        let n = r.read_until(b'\n', &mut line).await?;
59        if n == 0 {
60            if head.is_empty() {
61                return Ok(None);
62            }
63            bail!("the connection closed inside a response");
64        }
65        if head.is_empty() && line.iter().all(|b| b.is_ascii_whitespace()) {
66            continue;
67        }
68        if head.len() + line.len() > MAX_HEADER_BYTES {
69            bail!("response headers longer than {MAX_HEADER_BYTES} bytes");
70        }
71        let blank = line == b"\r\n" || line == b"\n";
72        head.extend_from_slice(&line);
73        if blank {
74            break;
75        }
76    }
77    let text = String::from_utf8_lossy(&head);
78    let mut lines = text.lines();
79    let status_line = lines.next().unwrap_or_default();
80    let mut parts = status_line.splitn(3, ' ');
81    let version = parts.next().unwrap_or_default();
82    if !version.starts_with("RTSP/") {
83        bail!("not an RTSP response: {status_line}");
84    }
85    let status: u16 = parts
86        .next()
87        .unwrap_or_default()
88        .parse()
89        .context("bad status code")?;
90    let reason = parts.next().unwrap_or_default().trim().to_string();
91    let headers: Vec<(String, String)> = lines
92        .filter_map(|l| {
93            l.split_once(':')
94                .map(|(k, v)| (k.trim().to_string(), v.trim().to_string()))
95        })
96        .collect();
97    let len = headers
98        .iter()
99        .find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
100        .and_then(|(_, v)| v.parse::<usize>().ok())
101        .unwrap_or(0);
102    if len > MAX_BODY_BYTES {
103        bail!("response body of {len} bytes (at most {MAX_BODY_BYTES})");
104    }
105    let mut body = vec![0u8; len];
106    r.read_exact(&mut body).await?;
107    Ok(Some(Response {
108        status,
109        reason,
110        headers,
111        body,
112    }))
113}
114
115/// The parts of an SDP the model needs: each media section, and the session-level control.
116pub fn parse_sdp(sdp: &str) -> Value {
117    let mut media: Vec<Value> = Vec::new();
118    let mut session_control = Value::Null;
119    for line in sdp.lines().map(str::trim) {
120        if let Some(m) = line.strip_prefix("m=") {
121            let f: Vec<&str> = m.split_whitespace().collect();
122            media.push(json!({"type": f.first(), "port": f.get(1).and_then(|p| p.parse::<u64>().ok()),
123                              "protocol": f.get(2), "formats": f.get(3..).unwrap_or_default(), "rtpmap": {}, "control": null}));
124        } else if let Some(a) = line.strip_prefix("a=") {
125            let target = media.last_mut();
126            if let Some(c) = a.strip_prefix("control:") {
127                match target {
128                    Some(m) => m["control"] = json!(c),
129                    None => session_control = json!(c),
130                }
131            } else if let (Some(r), Some(m)) = (a.strip_prefix("rtpmap:"), target) {
132                if let Some((pt, enc)) = r.split_once(' ') {
133                    m["rtpmap"][pt] = json!(enc);
134                }
135            }
136        }
137    }
138    json!({"media": media, "control": session_control})
139}
140
141/// A control URL resolved against the base (Content-Base, else the request URL).
142pub fn resolve(base: &str, control: &str) -> String {
143    if control.starts_with("rtsp://") || control.starts_with("rtsps://") {
144        control.to_string()
145    } else if control == "*" || control.is_empty() {
146        base.to_string()
147    } else if base.ends_with('/') {
148        format!("{base}{control}")
149    } else {
150        format!("{base}/{control}")
151    }
152}
153
154/// An RTP/RTCP pair on consecutive ports, RTP's even (RFC 3550 §11).
155async fn port_pair(ip: IpAddr) -> Result<(UdpSocket, UdpSocket)> {
156    for _ in 0..64 {
157        let rtp = UdpSocket::bind((ip, 0)).await?;
158        let port = rtp.local_addr()?.port();
159        if port % 2 != 0 || port == u16::MAX {
160            continue;
161        }
162        if let Ok(rtcp) = UdpSocket::bind((ip, port + 1)).await {
163            return Ok((rtp, rtcp));
164        }
165    }
166    bail!("no free even/odd UDP port pair")
167}
168
169struct Pending {
170    method: &'static str,
171    depth: usize,
172    caller: Option<ClientCommand>,
173    /// For SETUP: the sockets the Transport header offered.
174    sockets: Option<(UdpSocket, UdpSocket)>,
175}
176
177pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
178    let url = if ctx.remote_addr.starts_with("rtsp://") {
179        ctx.remote_addr.clone()
180    } else {
181        format!("rtsp://{}", ctx.remote_addr)
182    };
183    let parsed =
184        reqwest::Url::parse(&url).context("remote_addr must be an rtsp:// URL or host:port")?;
185    let host = parsed
186        .host_str()
187        .context("the URL names no host")?
188        .to_string();
189    let port = parsed.port().unwrap_or(554);
190    let stream = tokio::time::timeout(IO_TIMEOUT, TcpStream::connect((host.as_str(), port)))
191        .await
192        .context("RTSP connect deadline")??;
193    let local = stream.local_addr()?;
194    let (reader, writer) = tokio::io::split(stream);
195    ctx.state
196        .update_client_status(ctx.client_id, ClientStatus::Connected)
197        .await;
198    let external =
199        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
200    let (resp_tx, resp_rx) = mpsc::channel::<Result<Response>>(16);
201    let reader_task = tokio::spawn(async move {
202        let mut r = BufReader::new(reader);
203        loop {
204            let item = read_response(&mut r).await;
205            let end = !matches!(item, Ok(Some(_)));
206            let sent = match item {
207                Ok(Some(resp)) => resp_tx.send(Ok(resp)).await,
208                Ok(None) => {
209                    resp_tx
210                        .send(Err(anyhow::anyhow!("the server closed the connection")))
211                        .await
212                }
213                Err(e) => resp_tx.send(Err(e)).await,
214            };
215            if end || sent.is_err() {
216                return;
217            }
218        }
219    });
220    ctx.state
221        .register_client_task(ctx.client_id, reader_task)
222        .await;
223    let (internal_tx, internal_rx) = mpsc::channel::<(Value, usize)>(32);
224    let (event_tx, mut event_rx) = mpsc::channel::<(Event, usize)>(64);
225    event_tx.try_send((
226        Event::new(&actions::CONNECTED_EVENT, json!({"url": url})),
227        0,
228    ))?;
229    let events_ctx = ctx.clone();
230    let dispatcher = tokio::spawn(async move {
231        while let Some((event, depth)) = event_rx.recv().await {
232            events_ctx
233                .state
234                .record_access_log(
235                    AccessLogOwner::Client(events_ctx.client_id.as_u32()),
236                    "RTSP",
237                    None,
238                    event.id(),
239                    event.data.clone(),
240                    vec![],
241                )
242                .await;
243            let instruction = events_ctx
244                .state
245                .get_instruction_for_client(events_ctx.client_id)
246                .await
247                .unwrap_or_default();
248            let memory = events_ctx
249                .state
250                .get_memory_for_client(events_ctx.client_id)
251                .await
252                .unwrap_or_default();
253            match call_llm_for_client(
254                &events_ctx.llm_client,
255                &events_ctx.state,
256                events_ctx.client_id.to_string(),
257                &instruction,
258                &memory,
259                Some(&event),
260                &RtspClientProtocol,
261                &events_ctx.status_tx,
262            )
263            .await
264            {
265                Ok(result) => {
266                    if let Some(memory) = result.memory_updates {
267                        events_ctx
268                            .state
269                            .set_memory_for_client(events_ctx.client_id, memory)
270                            .await;
271                    }
272                    for action in result.actions {
273                        if internal_tx.send((action, depth + 1)).await.is_err() {
274                            return;
275                        }
276                    }
277                }
278                Err(e) => {
279                    Log::new(Some(&events_ctx.status_tx)).warn(format!("RTSP client handler: {e}"))
280                }
281            }
282        }
283    });
284    let dispatcher_abort = dispatcher.abort_handle();
285    ctx.state
286        .register_client_task(ctx.client_id, dispatcher)
287        .await;
288    let session_ctx = ctx.clone();
289    let task = tokio::spawn(async move {
290        let mut s = Session {
291            writer,
292            url,
293            local_ip: local.ip(),
294            cseq: 0,
295            pending: HashMap::new(),
296            session: None,
297            base: None,
298            tracks: Vec::new(),
299            receivers: Vec::new(),
300            events: event_tx,
301        };
302        let result = s.run(&session_ctx, resp_rx, external, internal_rx).await;
303        for r in s.receivers.drain(..) {
304            r.abort();
305        }
306        dispatcher_abort.abort();
307        let status = match result {
308            Ok(()) => ClientStatus::Disconnected,
309            Err(e) => {
310                Log::new(Some(&session_ctx.status_tx)).warn(format!("RTSP client ended: {e:#}"));
311                ClientStatus::Error(e.to_string())
312            }
313        };
314        session_ctx
315            .state
316            .update_client_status(session_ctx.client_id, status)
317            .await;
318        session_ctx
319            .state
320            .remove_client_handle(session_ctx.client_id)
321            .await;
322        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
323    });
324    ctx.state.register_client_task(ctx.client_id, task).await;
325    Ok(local)
326}
327
328struct Session {
329    writer: WriteHalf<TcpStream>,
330    url: String,
331    local_ip: IpAddr,
332    cseq: u64,
333    pending: HashMap<u64, Pending>,
334    session: Option<String>,
335    base: Option<String>,
336    /// Control URLs from the last DESCRIBE, first audio/video first.
337    tracks: Vec<String>,
338    receivers: Vec<tokio::task::AbortHandle>,
339    events: mpsc::Sender<(Event, usize)>,
340}
341
342fn reply(caller: Option<ClientCommand>, outcome: ClientSendOutcome) {
343    if let Some(c) = caller {
344        crate::client::command_support::reply(c, Ok(outcome));
345    }
346}
347
348impl Session {
349    async fn request(
350        &mut self,
351        method: &'static str,
352        uri: &str,
353        extra: &[(&str, String)],
354        pending: Pending,
355    ) -> Result<()> {
356        if self.pending.len() >= MAX_PENDING {
357            bail!("too many requests awaiting the server");
358        }
359        self.cseq += 1;
360        let mut req = format!(
361            "{method} {uri} RTSP/1.0\r\nCSeq: {}\r\nUser-Agent: {USER_AGENT}\r\n",
362            self.cseq
363        );
364        if let Some(s) = &self.session {
365            req.push_str(&format!("Session: {s}\r\n"));
366        }
367        for (k, v) in extra {
368            req.push_str(&format!("{k}: {v}\r\n"));
369        }
370        req.push_str("\r\n");
371        tokio::time::timeout(IO_TIMEOUT, self.writer.write_all(req.as_bytes()))
372            .await
373            .context("RTSP write deadline")??;
374        self.pending.insert(self.cseq, pending);
375        Ok(())
376    }
377
378    async fn act(
379        &mut self,
380        action: Value,
381        depth: usize,
382        caller: Option<ClientCommand>,
383    ) -> Result<()> {
384        let pending = |method, caller| Pending {
385            method,
386            depth,
387            caller,
388            sockets: None,
389        };
390        let url = self.url.clone();
391        match action["type"].as_str().unwrap_or_default() {
392            actions::OPTIONS => {
393                self.request("OPTIONS", &url, &[], pending("OPTIONS", caller))
394                    .await
395            }
396            actions::DESCRIBE => {
397                self.request(
398                    "DESCRIBE",
399                    &url,
400                    &[("Accept", "application/sdp".into())],
401                    pending("DESCRIBE", caller),
402                )
403                .await
404            }
405            actions::SETUP => {
406                if self.receivers.len() / 2 >= MAX_TRACKS {
407                    reply(
408                        caller,
409                        ClientSendOutcome::Rejected {
410                            error: "too many tracks set up".into(),
411                        },
412                    );
413                    return Ok(());
414                }
415                let base = self.base.clone().unwrap_or(url);
416                let track = match action["track"].as_str() {
417                    Some(t) => resolve(&base, t),
418                    None => match self.tracks.first() {
419                        Some(t) => t.clone(),
420                        None => base,
421                    },
422                };
423                let (rtp, rtcp) = port_pair(self.local_ip).await?;
424                let p = rtp.local_addr()?.port();
425                let mut pend = pending("SETUP", caller);
426                pend.sockets = Some((rtp, rtcp));
427                self.request(
428                    "SETUP",
429                    &track,
430                    &[(
431                        "Transport",
432                        format!("RTP/AVP;unicast;client_port={p}-{}", p + 1),
433                    )],
434                    pend,
435                )
436                .await
437            }
438            actions::PLAY => {
439                let range: Vec<(&str, String)> = action["range"]
440                    .as_str()
441                    .map(|r| vec![("Range", r.to_string())])
442                    .unwrap_or_default();
443                let target = self.base.clone().unwrap_or(url);
444                self.request("PLAY", &target, &range, pending("PLAY", caller))
445                    .await
446            }
447            actions::PAUSE => {
448                let target = self.base.clone().unwrap_or(url);
449                self.request("PAUSE", &target, &[], pending("PAUSE", caller))
450                    .await
451            }
452            actions::TEARDOWN => {
453                let target = self.base.clone().unwrap_or(url);
454                self.request("TEARDOWN", &target, &[], pending("TEARDOWN", caller))
455                    .await
456            }
457            _ => Ok(()),
458        }
459    }
460
461    async fn on_response(
462        &mut self,
463        ctx: &ConnectContext,
464        r: Response,
465        media: &mpsc::Sender<(Vec<u8>, SocketAddr)>,
466    ) -> Result<()> {
467        let cseq = r
468            .header("CSeq")
469            .and_then(|c| c.trim().parse::<u64>().ok())
470            .unwrap_or(0);
471        let Some(p) = self.pending.remove(&cseq) else {
472            return Ok(());
473        };
474        let mut headers = Map::new();
475        for name in [
476            "Public",
477            "Content-Base",
478            "Session",
479            "Transport",
480            "RTP-Info",
481            "Range",
482        ] {
483            if let Some(v) = r.header(name) {
484                headers.insert(name.to_string(), json!(v));
485            }
486        }
487        let mut data =
488            json!({"method": p.method, "status": r.status, "reason": r.reason, "headers": headers});
489        let ok = (200..300).contains(&r.status);
490        if ok {
491            if let Some(s) = r.header("Session") {
492                self.session = Some(s.split(';').next().unwrap_or_default().trim().to_string());
493            }
494            match p.method {
495                "DESCRIBE" => {
496                    let base = r
497                        .header("Content-Base")
498                        .or(r.header("Content-Location"))
499                        .map(str::to_string)
500                        .unwrap_or_else(|| self.url.clone());
501                    let sdp = parse_sdp(&String::from_utf8_lossy(&r.body));
502                    let session_base = sdp["control"]
503                        .as_str()
504                        .map(|c| resolve(&base, c))
505                        .unwrap_or(base.clone());
506                    self.tracks = sdp["media"]
507                        .as_array()
508                        .into_iter()
509                        .flatten()
510                        .filter(|m| matches!(m["type"].as_str(), Some("audio" | "video")))
511                        .map(|m| resolve(&session_base, m["control"].as_str().unwrap_or("*")))
512                        .collect();
513                    self.base = Some(base);
514                    data["sdp"] = sdp;
515                }
516                "SETUP" => {
517                    if let Some((rtp, rtcp)) = p.sockets {
518                        for sock in [rtp, rtcp] {
519                            let tx = media.clone();
520                            let task = tokio::spawn(async move {
521                                let mut buf = vec![0u8; 2048];
522                                while let Ok((n, from)) = sock.recv_from(&mut buf).await {
523                                    if tx.send((buf[..n].to_vec(), from)).await.is_err() {
524                                        return;
525                                    }
526                                }
527                            });
528                            self.receivers.push(task.abort_handle());
529                            ctx.state.register_client_task(ctx.client_id, task).await;
530                        }
531                    }
532                }
533                "TEARDOWN" => {
534                    for h in self.receivers.drain(..) {
535                        h.abort();
536                    }
537                    self.session = None;
538                }
539                _ => {}
540            }
541        }
542        reply(
543            p.caller,
544            ClientSendOutcome::Executed {
545                detail: data.to_string(),
546            },
547        );
548        self.events
549            .try_send((Event::new(&actions::RESPONSE_EVENT, data), p.depth))
550            .context("RTSP event queue full; consumer stalled")
551    }
552
553    async fn run(
554        &mut self,
555        ctx: &ConnectContext,
556        mut responses: mpsc::Receiver<Result<Response>>,
557        mut external: mpsc::Receiver<ClientCommand>,
558        mut internal: mpsc::Receiver<(Value, usize)>,
559    ) -> Result<()> {
560        let log = Log::new(Some(&ctx.status_tx));
561        let (media_tx, mut media_rx) = mpsc::channel::<(Vec<u8>, SocketAddr)>(1024);
562        let mut tracker = Tracker::default();
563        let mut tick = tokio::time::interval(Duration::from_millis(250));
564        loop {
565            let (action, depth, mut injected) = tokio::select! {
566                r = responses.recv() => {
567                    let Some(r) = r else { return Ok(()) };
568                    self.on_response(ctx, r?, &media_tx).await?;
569                    continue;
570                }
571                m = media_rx.recv() => {
572                    if let Some((data, from)) = m {
573                        if !crate::server::rtp::media::is_rtcp(&data) {
574                            if let Some(started) = tracker.on_rtp(&data, from) {
575                                let _ = self.events.try_send((Event::new(&actions::STREAM_STARTED_EVENT, started), 0));
576                            }
577                        }
578                    }
579                    continue;
580                }
581                _ = tick.tick() => {
582                    for data in tracker.sweep() {
583                        let _ = self.events.try_send((Event::new(&actions::STREAM_ENDED_EVENT, data), 0));
584                    }
585                    continue;
586                }
587                command = external.recv() => match command {
588                    Some(c) => (c.action.clone(), 0, Some(c)),
589                    None => return Ok(()),
590                },
591                action = internal.recv() => match action {
592                    Some((a, depth)) => (a, depth, None),
593                    None => return Ok(()),
594                },
595            };
596            match RtspClientProtocol.execute_action(action.clone()) {
597                Ok(ClientActionResult::Disconnect) => {
598                    reply(injected.take(), ClientSendOutcome::Disconnected);
599                    return Ok(());
600                }
601                Ok(_) => {}
602                Err(e) => {
603                    log.warn(format!("RTSP client action refused: {e}"));
604                    reply(
605                        injected.take(),
606                        ClientSendOutcome::Rejected {
607                            error: e.to_string(),
608                        },
609                    );
610                    continue;
611                }
612            }
613            if depth > MAX_FOLLOWUP_DEPTH {
614                log.warn(format!(
615                    "RTSP client: handler chain stopped after {MAX_FOLLOWUP_DEPTH} follow-ups"
616                ));
617                continue;
618            }
619            if injected.is_some() {
620                ctx.state
621                    .record_access_log(
622                        AccessLogOwner::Client(ctx.client_id.as_u32()),
623                        "RTSP",
624                        None,
625                        "injected_action",
626                        action.clone(),
627                        vec![],
628                    )
629                    .await;
630            }
631            self.act(action, depth, injected.take()).await?;
632        }
633    }
634}
netget/client/rtsp/mod.rs

netget/client/rtsp/
mod.rs
