seydd/input/
rtsp.rs

1// Copyright 2026 Anton Gravestam
2// SPDX-License-Identifier: Apache-2.0
3//! RTSP input via `retina`, pulled over TCP.
4//!
5//! TCP rather than UDP on purpose: an IP camera's RTP/UDP has no FEC and its
6//! losses would arrive as corrupt access units that Seyd then spends parity
7//! protecting. The camera hop is a short LAN link where a retransmit costs
8//! microseconds; the lossy path worth protecting is the one after the agent.
9
10use super::{redact, Codec, VideoAu};
11use bytes::Bytes;
12use futures_util::StreamExt;
13use retina::client::{PlayOptions, SessionOptions, SetupOptions, Transport};
14use retina::codec::{CodecItem, FrameFormat};
15use std::time::{Duration, Instant};
16use tokio::sync::mpsc;
17
18const BASE_BACKOFF: Duration = Duration::from_secs(1);
19const MAX_BACKOFF: Duration = Duration::from_secs(30);
20/// A stream that ran this long was healthy, not flapping, so the failure that
21/// ends it starts the backoff over. Without this a camera that drops after
22/// hours costs the operator `MAX_BACKOFF` of black screen instead of a second.
23const STABLE_SESSION: Duration = Duration::from_secs(30);
24
25pub async fn run(url: String, codec: Codec, tx: mpsc::Sender<VideoAu>) {
26    let mut backoff = BASE_BACKOFF;
27    loop {
28        let started = Instant::now();
29        let outcome = session(&url, codec, &tx).await;
30        // Reset before logging, so the delay reported is the one actually slept.
31        if started.elapsed() >= STABLE_SESSION {
32            backoff = BASE_BACKOFF;
33        }
34        match outcome {
35            Ok(()) => tracing::info!(url = %redact(&url), "rtsp stream ended; reconnecting"),
36            Err(e) => {
37                tracing::warn!(url = %redact(&url), error = %redact(&e.to_string()), "rtsp input failed; retrying in {backoff:?}")
38            }
39        }
40        if tx.is_closed() {
41            return;
42        }
43        tokio::time::sleep(backoff).await;
44        backoff = (backoff * 2).min(MAX_BACKOFF);
45    }
46}
47
48/// Substitute `SEYD_RTSP_USER` / `SEYD_RTSP_PASSWORD` into a URL without userinfo.
49pub fn resolve_credentials(url: &str) -> String {
50    let Ok(mut u) = url::Url::parse(url) else {
51        return url.to_string();
52    };
53    if !u.username().is_empty() {
54        return url.to_string();
55    }
56    if let (Ok(user), Ok(pass)) = (
57        std::env::var("SEYD_RTSP_USER"),
58        std::env::var("SEYD_RTSP_PASSWORD"),
59    ) {
60        let _ = u.set_username(&user);
61        let _ = u.set_password(Some(&pass));
62    }
63    u.to_string()
64}
65
66async fn session(url: &str, codec: Codec, tx: &mpsc::Sender<VideoAu>) -> anyhow::Result<()> {
67    let mut parsed = url::Url::parse(url)?;
68    let creds = if !parsed.username().is_empty() {
69        let c = retina::client::Credentials {
70            username: parsed.username().to_string(),
71            password: parsed.password().unwrap_or("").to_string(),
72        };
73        let _ = parsed.set_username("");
74        let _ = parsed.set_password(None);
75        Some(c)
76    } else {
77        None
78    };
79
80    let mut sess = retina::client::Session::describe(
81        parsed,
82        SessionOptions::default()
83            .creds(creds)
84            .user_agent("seydd".into()),
85    )
86    .await?;
87
88    // Video only: a camera carrying audio would otherwise hand us AAC frames
89    // that a VideoDecoder would be fed as delta frames.
90    let video_idx = sess
91        .streams()
92        .iter()
93        .position(|s| {
94            s.media() == "video"
95                && s.encoding_name()
96                    .eq_ignore_ascii_case(codec.rtp_encoding_name())
97        })
98        .ok_or_else(|| anyhow::anyhow!("no H.264 video stream in RTSP DESCRIBE"))?;
99    sess.setup(
100        video_idx,
101        SetupOptions::default()
102            .transport(Transport::Tcp(Default::default()))
103            .frame_format(FrameFormat::SIMPLE),
104    )
105    .await?;
106    let playing = sess.play(PlayOptions::default()).await?;
107    let mut demuxed = playing.demuxed()?;
108    tracing::info!(url = %redact(url), "rtsp input playing");
109
110    let mut frames: u64 = 0;
111    let mut last_at = std::time::Instant::now();
112    while let Some(item) = demuxed.next().await {
113        let gap = last_at.elapsed();
114        last_at = std::time::Instant::now();
115        if gap.as_millis() > 120 {
116            tracing::debug!(gap_ms = gap.as_millis() as u64, "rtsp inter-frame gap");
117        }
118        if let CodecItem::VideoFrame(f) = item? {
119            {
120                if f.stream_id() != video_idx {
121                    continue;
122                }
123                // The camera's own sampling clock (RTP timestamp), not our
124                // arrival time: the pilot paces presentation on this timeline,
125                // and arrival already carries the jitter we are trying to
126                // remove — a keyframe reaches us several ms after its slot.
127                let ts = f.timestamp();
128                let capture_ts_us = (ts.elapsed().max(0) as u128 * 1_000_000
129                    / ts.clock_rate().get() as u128) as u64;
130                let au = VideoAu {
131                    keyframe: f.is_random_access_point(),
132                    input_loss: f.loss(),
133                    capture_ts_us,
134                    data: Bytes::from(f.into_data()),
135                };
136                frames += 1;
137                if frames == 1 {
138                    tracing::info!(
139                        bytes = au.data.len(),
140                        keyframe = au.keyframe,
141                        "first rtsp frame"
142                    );
143                }
144                if tx.send(au).await.is_err() {
145                    return Ok(());
146                }
147            }
148        }
149    }
150    Ok(())
151}
seydd/input/rtsp.rs

seydd/input/
rtsp.rs
