media_pp/elements/sink/
rtsp_sink.rs

1use std::{ffi::CString, ptr, sync::Arc};
2
3use crate::pp_log::{PpLog, pp_error, pp_info};
4use ffmpeg_next::{self as ffmpeg, ffi};
5use thiserror::Error as ThisError;
6
7use crate::{
8    buffer::MediaBuffer,
9    contract::{InputContract, MediaKind, PortContract},
10    control::ControlMsg,
11    element::{Element, ElementType, Sink, element_pp_log},
12    elements::RtspTransport,
13    error::Result,
14};
15
16/// Errors produced while opening or writing an [`RtspSink`].
17#[derive(Debug, ThisError)]
18pub enum RtspSinkError {
19    /// FFmpeg rejected connection setup or packet writing.
20    #[error("ffmpeg error: {0}")]
21    Ffmpeg(#[from] ffmpeg::Error),
22
23    /// The sink received a decoded frame instead of compressed packet data.
24    #[error(
25        "RtspSink only remuxes compressed Packets, got a decoded {0}; \
26         connect an encoder or demuxer packet pad instead"
27    )]
28    UnsupportedBuffer(&'static str),
29
30    /// The URL contains an interior NUL byte rejected by FFmpeg's C API.
31    #[error("RTSP URL contains a NUL byte")]
32    InvalidUrl,
33}
34
35/// Publishes one compressed packet stream to an already-running RTSP server.
36///
37/// [`RtspSink::open`] performs the RTSP `ANNOUNCE`/`SETUP`/`RECORD`
38/// handshake through FFmpeg, so the server must already be listening at
39/// `url` and must permit publishing to that path. The server can be
40/// MediaMTX or any other implementation that accepts RTSP publishing;
41/// this element does not start, stop, or otherwise depend on a particular
42/// server process.
43///
44/// This is a remuxing sink, not an encoder. Incoming buffers must be
45/// compressed [`MediaBuffer::Packet`] values whose codec parameters and
46/// time base match the values passed to [`RtspSink::open`]. Place a
47/// [`crate::elements::Pacer`] upstream when publishing packets from a file,
48/// otherwise the file will be sent faster than real time.
49///
50/// The current sink publishes one stream. Build separate sinks and RTSP
51/// paths when publishing independent streams.
52pub struct RtspSink {
53    pp_log: PpLog,
54    name: Arc<str>,
55    url: String,
56    /// The medium this session publishes; `None` for one this crate does
57    /// not model, which then declares nothing.
58    kind: Option<MediaKind>,
59    output: ffmpeg::format::context::Output,
60    input_time_base: ffmpeg::Rational,
61    last_output_dts: Option<i64>,
62    last_output_pts: Option<i64>,
63    pts_offset: i64,
64    pending_seek: bool,
65}
66
67impl RtspSink {
68    /// Connects to `url` and starts publishing.
69    ///
70    /// `params` and `time_base` must describe every packet subsequently
71    /// passed to [`Sink::consume`]. TCP is the most reliable transport for
72    /// general networks; UDP is useful when the network path and server
73    /// permit the negotiated RTP/RTCP ports.
74    pub fn open(
75        name: impl Into<String>,
76        url: impl Into<String>,
77        transport: RtspTransport,
78        params: ffmpeg::codec::Parameters,
79        time_base: ffmpeg::Rational,
80    ) -> Result<Self> {
81        let url = url.into();
82        let kind = MediaKind::packet_for(params.medium());
83        let mut output = alloc_output(&url)?;
84
85        {
86            let mut stream = output
87                .add_stream(ffmpeg::encoder::find(ffmpeg::codec::Id::None))
88                .map_err(RtspSinkError::from)?;
89            stream.set_parameters(params);
90            // Avoid codec-tag incompatibilities when the input packet came
91            // from a container with a different tag convention.
92            // SAFETY: `as_mut_ptr` on parameters this stream owns, written before the
93            // stream is handed to the muxer — see the comment beside it for why the tag
94            // is cleared at all.
95            unsafe {
96                (*stream.parameters().as_mut_ptr()).codec_tag = 0;
97            }
98            stream.set_time_base(time_base);
99        }
100
101        let mut options = ffmpeg::Dictionary::new();
102        options.set("rtsp_transport", transport.as_ffmpeg_option());
103        output
104            .write_header_with(options)
105            .map_err(RtspSinkError::from)?;
106
107        let name: Arc<str> = name.into().into();
108        let pp_log = element_pp_log(ElementType::RtspSink, &name, None);
109        pp_info!(pp_log: &pp_log, "publishing: url={url}, transport={transport:?}");
110
111        Ok(Self {
112            pp_log,
113            name,
114            url,
115            kind,
116            output,
117            input_time_base: time_base,
118            last_output_dts: None,
119            last_output_pts: None,
120            pts_offset: 0,
121            pending_seek: false,
122        })
123    }
124
125    /// URL this sink publishes to.
126    pub fn url(&self) -> &str {
127        &self.url
128    }
129}
130
131/// Allocates an RTSP muxer without opening a generic `AVIOContext`.
132///
133/// RTSP is a libavformat muxer, not a generic AVIO protocol. Its muxer
134/// owns the control and RTP sockets internally during header/packet writes,
135/// while `ffmpeg_next::format::output_as` attempts an incompatible generic
136/// `avio_open2` first on FFmpeg builds where `rtsp` is not an AVIO protocol.
137fn alloc_output(url: &str) -> Result<ffmpeg::format::context::Output> {
138    let c_url = CString::new(url).map_err(|_| RtspSinkError::InvalidUrl)?;
139    let c_format = CString::new("rtsp").expect("static format name contains no NUL");
140
141    // SAFETY: `c_format` and `c_url` are live NUL-terminated `CString`s, and
142    // `context` is a live local. Every path below checks it before use, and the
143    // failure paths free what was allocated.
144    unsafe {
145        let mut context: *mut ffi::AVFormatContext = ptr::null_mut();
146        let result = ffi::avformat_alloc_output_context2(
147            &mut context,
148            ptr::null_mut(),
149            c_format.as_ptr(),
150            c_url.as_ptr(),
151        );
152        if result < 0 {
153            return Err(RtspSinkError::Ffmpeg(ffmpeg::Error::from(result)).into());
154        }
155
156        Ok(ffmpeg::format::context::Output::wrap(context))
157    }
158}
159
160impl Element for RtspSink {
161    fn name(&self) -> Arc<str> {
162        self.name.clone()
163    }
164
165    fn element_type(&self) -> ElementType {
166        ElementType::RtspSink
167    }
168
169    fn pp_log(&self) -> &PpLog {
170        &self.pp_log
171    }
172
173    fn pp_log_mut(&mut self) -> &mut PpLog {
174        &mut self.pp_log
175    }
176}
177
178impl Sink for RtspSink {
179    /// Republishes encoded data as-is; it has no encoder of its own.
180    fn input_contract(&self) -> InputContract {
181        match self.kind {
182            Some(kind) => InputContract::Fixed(PortContract::packet(kind)),
183            None => InputContract::Unknown,
184        }
185    }
186
187    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
188        match buf {
189            MediaBuffer::Packet(packet) => {
190                let mut packet = (*packet).clone();
191                let output_time_base = self
192                    .output
193                    .stream(0)
194                    .expect("stream 0 was added by RtspSink::open")
195                    .time_base();
196                packet.rescale_ts(self.input_time_base, output_time_base);
197
198                if let Some(raw_pts) = packet.pts() {
199                    if self.pending_seek {
200                        // Keep the published timeline monotonic across an
201                        // upstream seek. DTS is the muxer's hard ordering
202                        // requirement; PTS is the fallback for packets that
203                        // carry no DTS.
204                        self.pts_offset = match (self.last_output_dts, packet.dts()) {
205                            (Some(last_dts), Some(raw_dts)) => last_dts + 1 - raw_dts,
206                            _ => match self.last_output_pts {
207                                Some(last_pts) => last_pts + 1 - raw_pts,
208                                None => 0,
209                            },
210                        };
211                        self.pending_seek = false;
212                    }
213
214                    let corrected_pts = raw_pts + self.pts_offset;
215                    packet.set_pts(Some(corrected_pts));
216                    if let Some(raw_dts) = packet.dts() {
217                        let corrected_dts = raw_dts + self.pts_offset;
218                        packet.set_dts(Some(corrected_dts));
219                        self.last_output_dts = Some(corrected_dts);
220                    }
221                    self.last_output_pts = Some(corrected_pts);
222                }
223
224                packet.set_stream(0);
225                packet.set_position(-1);
226                packet
227                    .write_interleaved(&mut self.output)
228                    .map_err(RtspSinkError::from)
229                    .map_err(Into::into)
230                    .inspect_err(|error| pp_error!(self, "write_interleaved failed: {error}"))
231            }
232            MediaBuffer::Eos => self
233                .output
234                .write_trailer()
235                .map_err(RtspSinkError::from)
236                .map_err(Into::into)
237                .inspect_err(|error| pp_error!(self, "write_trailer failed: {error}")),
238            MediaBuffer::Video(_) => {
239                pp_error!(self, "unsupported buffer: Video");
240                Err(RtspSinkError::UnsupportedBuffer("Video").into())
241            }
242            MediaBuffer::Audio(_) => {
243                pp_error!(self, "unsupported buffer: Audio");
244                Err(RtspSinkError::UnsupportedBuffer("Audio").into())
245            }
246        }
247    }
248
249    fn control(&mut self, msg: ControlMsg) -> Result<()> {
250        match msg {
251            ControlMsg::Seek(_) => self.pending_seek = true,
252            ControlMsg::Pause
253            | ControlMsg::Resume
254            | ControlMsg::Stop
255            | ControlMsg::Flush
256            | ControlMsg::CheckSeek(_)
257            | ControlMsg::Preroll(_) => {}
258        }
259        Ok(())
260    }
261}
262
263impl Drop for RtspSink {
264    fn drop(&mut self) {
265        pp_info!(
266            self,
267            "dropped: closing publisher connection to {}",
268            self.url
269        );
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use ffmpeg_next as ffmpeg;
276
277    use super::{RtspSink, RtspSinkError};
278    use crate::{elements::RtspTransport, error::Error};
279
280    #[test]
281    fn rejects_a_url_containing_a_nul_byte_before_connecting() {
282        let result = RtspSink::open(
283            "rtsp",
284            "rtsp://127.0.0.1:8554/stream\0invalid",
285            RtspTransport::Tcp,
286            ffmpeg::codec::Parameters::new(),
287            ffmpeg::Rational(1, 1_000),
288        );
289
290        assert!(matches!(
291            result,
292            Err(Error::RtspSinkError(RtspSinkError::InvalidUrl))
293        ));
294    }
295}
media_pp/elements/sink/rtsp_sink.rs

media_pp/elements/sink/
rtsp_sink.rs
