1use std::{
2 borrow::Cow,
3 ffi::CString,
4 path::{Path, PathBuf},
5 ptr,
6 sync::{Arc, Mutex},
7 time::Duration,
8};
9
10use crate::pp_log::{PpLog, pp_error};
11use ffmpeg_next as ffmpeg;
12use thiserror::Error as ThisError;
13
14use crate::{
15 buffer::MediaBuffer,
16 control::ControlMsg,
17 element::{Element, ElementType, Sink, element_pp_log},
18 error::Result,
19};
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum HlsMode {
25 Live {
29 window_size: usize,
30 delete_old_segments: bool,
31 },
32 Event,
35 Vod,
38}
39
40#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum HlsSegmentFormat {
43 MpegTs,
45 Fmp4,
48}
49
50#[derive(Debug, Clone)]
52pub struct HlsOptions {
53 pub playlist_path: PathBuf,
55 pub segment_pattern: PathBuf,
58 pub segment_duration: Duration,
61 pub mode: HlsMode,
62 pub segment_format: HlsSegmentFormat,
63 pub init_filename: String,
66 pub base_url: Option<String>,
69}
70
71impl HlsOptions {
72 pub fn new(playlist_path: impl Into<PathBuf>, segment_pattern: impl Into<PathBuf>) -> Self {
75 Self {
76 playlist_path: playlist_path.into(),
77 segment_pattern: segment_pattern.into(),
78 segment_duration: Duration::from_secs(2),
79 mode: HlsMode::Live {
80 window_size: 6,
81 delete_old_segments: true,
82 },
83 segment_format: HlsSegmentFormat::Fmp4,
84 init_filename: "init.mp4".into(),
85 base_url: None,
86 }
87 }
88
89 fn validate(&self) -> std::result::Result<(), HlsMuxerError> {
90 if self.segment_duration.is_zero() {
91 return Err(HlsMuxerError::ZeroSegmentDuration);
92 }
93 if let HlsMode::Live { window_size, .. } = self.mode
94 && (window_size == 0 || window_size > i32::MAX as usize)
95 {
96 return Err(HlsMuxerError::InvalidWindowSize(window_size));
97 }
98 if self.segment_format == HlsSegmentFormat::Fmp4 && self.init_filename.is_empty() {
99 return Err(HlsMuxerError::EmptyInitFilename);
100 }
101
102 let pattern = path_as_utf8(&self.segment_pattern, "segment_pattern")?;
103 if !has_integer_conversion(pattern) {
104 return Err(HlsMuxerError::MissingSegmentIndex(
105 self.segment_pattern.clone(),
106 ));
107 }
108 path_as_utf8(&self.playlist_path, "playlist_path")?;
109 reject_nul(pattern, "segment_pattern")?;
110 reject_nul(&self.init_filename, "init_filename")?;
111 if let Some(base_url) = &self.base_url {
112 reject_nul(base_url, "base_url")?;
113 }
114 Ok(())
115 }
116
117 fn header_options(&self) -> std::result::Result<ffmpeg::Dictionary<'static>, HlsMuxerError> {
118 let mut options = ffmpeg::Dictionary::new();
119 options.set("hls_time", &self.segment_duration.as_secs_f64().to_string());
120 let segment_pattern = path_for_ffmpeg(&self.segment_pattern, "segment_pattern")?;
121 options.set("hls_segment_filename", &segment_pattern);
122
123 let mut flags = vec!["temp_file", "independent_segments"];
124 match self.mode {
125 HlsMode::Live {
126 window_size,
127 delete_old_segments,
128 } => {
129 options.set("hls_list_size", &window_size.to_string());
130 if delete_old_segments {
131 flags.push("delete_segments");
132 }
133 }
134 HlsMode::Event => {
135 options.set("hls_playlist_type", "event");
136 options.set("hls_list_size", "0");
137 }
138 HlsMode::Vod => {
139 options.set("hls_playlist_type", "vod");
140 options.set("hls_list_size", "0");
141 }
142 }
143
144 match self.segment_format {
145 HlsSegmentFormat::MpegTs => options.set("hls_segment_type", "mpegts"),
146 HlsSegmentFormat::Fmp4 => {
147 options.set("hls_segment_type", "fmp4");
148 options.set("hls_fmp4_init_filename", &self.init_filename);
149 }
150 }
151 options.set("hls_flags", &flags.join("+"));
152 if let Some(base_url) = &self.base_url {
153 options.set("hls_base_url", base_url);
154 }
155 Ok(options)
156 }
157}
158
159fn path_as_utf8<'a>(
160 path: &'a Path,
161 field: &'static str,
162) -> std::result::Result<&'a str, HlsMuxerError> {
163 path.to_str().ok_or_else(|| HlsMuxerError::NonUtf8Path {
164 field,
165 path: path.to_path_buf(),
166 })
167}
168
169fn path_for_ffmpeg<'a>(
173 path: &'a Path,
174 field: &'static str,
175) -> std::result::Result<Cow<'a, str>, HlsMuxerError> {
176 let path = path_as_utf8(path, field)?;
177 #[cfg(windows)]
178 {
179 Ok(Cow::Owned(path.replace('\\', "/")))
180 }
181 #[cfg(not(windows))]
182 {
183 Ok(Cow::Borrowed(path))
184 }
185}
186
187fn reject_nul(value: &str, field: &'static str) -> std::result::Result<(), HlsMuxerError> {
188 if value.contains('\0') {
189 Err(HlsMuxerError::EmbeddedNul { field })
190 } else {
191 Ok(())
192 }
193}
194
195fn allocate_output(
199 options: &HlsOptions,
200) -> std::result::Result<ffmpeg::format::context::Output, HlsMuxerError> {
201 let path = path_for_ffmpeg(&options.playlist_path, "playlist_path")?;
202 let path = CString::new(path.as_ref()).map_err(|_| HlsMuxerError::EmbeddedNul {
203 field: "playlist_path",
204 })?;
205 let format = CString::new("hls").expect("static HLS format name contains no NUL");
206 let mut context = ptr::null_mut();
207 let result = unsafe {
208 ffmpeg::ffi::avformat_alloc_output_context2(
209 &mut context,
210 ptr::null_mut(),
211 format.as_ptr(),
212 path.as_ptr(),
213 )
214 };
215 if result < 0 {
216 return Err(HlsMuxerError::Ffmpeg(ffmpeg::Error::from(result)));
217 }
218 if context.is_null() {
219 return Err(HlsMuxerError::Ffmpeg(ffmpeg::Error::Unknown));
220 }
221 Ok(unsafe { ffmpeg::format::context::Output::wrap(context) })
222}
223
224fn has_integer_conversion(pattern: &str) -> bool {
227 let bytes = pattern.as_bytes();
228 let mut index = 0;
229 while index < bytes.len() {
230 if bytes[index] != b'%' {
231 index += 1;
232 continue;
233 }
234 index += 1;
235 if index < bytes.len() && bytes[index] == b'%' {
236 index += 1;
237 continue;
238 }
239 while index < bytes.len() && (bytes[index] == b'0' || bytes[index].is_ascii_digit()) {
240 index += 1;
241 }
242 if index < bytes.len() && bytes[index] == b'd' {
243 return true;
244 }
245 }
246 false
247}
248
249#[derive(Debug, ThisError)]
251pub enum HlsMuxerError {
252 #[error("HlsMuxer stream sinks only accept Packet or Eos buffers, got {0}")]
253 UnsupportedBuffer(&'static str),
254
255 #[error("HLS segment duration must be greater than zero")]
256 ZeroSegmentDuration,
257
258 #[error(
259 "HLS live window size must be between 1 and {max}, got {0}",
260 max = i32::MAX
261 )]
262 InvalidWindowSize(usize),
263
264 #[error("HLS fMP4 init filename must not be empty")]
265 EmptyInitFilename,
266
267 #[error("HLS segment pattern must contain an integer conversion such as %05d: {0:?}")]
268 MissingSegmentIndex(PathBuf),
269
270 #[error("HLS {field} must be valid UTF-8: {path:?}")]
271 NonUtf8Path { field: &'static str, path: PathBuf },
272
273 #[error("HLS {field} must not contain a NUL byte")]
274 EmbeddedNul { field: &'static str },
275
276 #[error("HLS muxer requires at least one stream")]
277 NoStreams,
278
279 #[error("ffmpeg error: {0}")]
280 Ffmpeg(#[from] ffmpeg::Error),
281}
282
283struct PendingStream {
284 name: Arc<str>,
285 input_time_base: ffmpeg::Rational,
286}
287
288pub struct HlsMuxer {
299 output: ffmpeg::format::context::Output,
300 streams: Vec<PendingStream>,
301 options: HlsOptions,
302}
303
304impl HlsMuxer {
305 pub fn create(options: HlsOptions) -> Result<Self> {
308 options.validate()?;
309 let output = allocate_output(&options)?;
310 Ok(Self {
311 output,
312 streams: Vec::new(),
313 options,
314 })
315 }
316
317 pub fn add_stream(
320 &mut self,
321 name: impl Into<String>,
322 parameters: ffmpeg::codec::Parameters,
323 time_base: ffmpeg::Rational,
324 ) -> Result<()> {
325 let mut stream = self
326 .output
327 .add_stream(parameters.id())
328 .map_err(HlsMuxerError::from)?;
329 stream.set_time_base(time_base);
330 stream.set_parameters(parameters);
331 self.streams.push(PendingStream {
332 name: name.into().into(),
333 input_time_base: time_base,
334 });
335 Ok(())
336 }
337
338 pub fn open(mut self) -> Result<Vec<Box<dyn Sink>>> {
342 if self.streams.is_empty() {
343 return Err(HlsMuxerError::NoStreams.into());
344 }
345 let options = self.options.header_options()?;
346 let unused = self
347 .output
348 .write_header_with(options)
349 .map_err(HlsMuxerError::from)?;
350 drop(unused);
351 let total = self.streams.len();
352 let shared = Arc::new(HlsMuxerShared {
353 state: Mutex::new(MuxerState {
354 output: self.output,
355 done: 0,
356 finished: false,
357 }),
358 total,
359 });
360 Ok(self
361 .streams
362 .into_iter()
363 .enumerate()
364 .map(|(index, stream)| -> Box<dyn Sink> {
365 Box::new(HlsMuxerStreamSink {
366 pp_log: element_pp_log(ElementType::HlsMuxer, &stream.name, None),
367 name: stream.name,
368 shared: shared.clone(),
369 stream_index: index,
370 input_time_base: stream.input_time_base,
371 done: false,
372 })
373 })
374 .collect())
375 }
376}
377
378struct MuxerState {
379 output: ffmpeg::format::context::Output,
380 done: usize,
381 finished: bool,
382}
383
384struct HlsMuxerShared {
385 state: Mutex<MuxerState>,
386 total: usize,
387}
388
389impl HlsMuxerShared {
390 fn write_packet(
391 &self,
392 stream_index: usize,
393 input_time_base: ffmpeg::Rational,
394 packet: &ffmpeg::Packet,
395 ) -> Result<()> {
396 let mut state = self.state.lock().unwrap();
397 if state.finished {
398 return Ok(());
399 }
400 let mut packet = packet.clone();
401 let output_time_base = state
402 .output
403 .stream(stream_index)
404 .expect("stream was added in HlsMuxer::add_stream")
405 .time_base();
406 packet.rescale_ts(input_time_base, output_time_base);
407 packet.set_stream(stream_index);
408 packet.set_position(-1);
409 packet
410 .write_interleaved(&mut state.output)
411 .map_err(HlsMuxerError::from)?;
412 Ok(())
413 }
414
415 fn finish_track(&self) -> Result<()> {
416 let mut state = self.state.lock().unwrap();
417 state.done += 1;
418 if state.finished || state.done < self.total {
419 return Ok(());
420 }
421 state.finished = true;
422 state.output.write_trailer().map_err(HlsMuxerError::from)?;
423 Ok(())
424 }
425}
426
427pub struct HlsMuxerStreamSink {
430 pp_log: PpLog,
431 name: Arc<str>,
432 shared: Arc<HlsMuxerShared>,
433 stream_index: usize,
434 input_time_base: ffmpeg::Rational,
435 done: bool,
436}
437
438impl HlsMuxerStreamSink {
439 fn finish(&mut self) -> Result<()> {
440 if self.done {
441 return Ok(());
442 }
443 self.done = true;
444 self.shared
445 .finish_track()
446 .inspect_err(|error| pp_error!(self, "write_trailer failed: {error}"))
447 }
448}
449
450impl Element for HlsMuxerStreamSink {
451 fn name(&self) -> Arc<str> {
452 self.name.clone()
453 }
454
455 fn element_type(&self) -> ElementType {
456 ElementType::HlsMuxer
457 }
458
459 fn pp_log(&self) -> &PpLog {
460 &self.pp_log
461 }
462
463 fn pp_log_mut(&mut self) -> &mut PpLog {
464 &mut self.pp_log
465 }
466}
467
468impl Sink for HlsMuxerStreamSink {
469 fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
470 match buf {
471 MediaBuffer::Packet(packet) => self
472 .shared
473 .write_packet(self.stream_index, self.input_time_base, &packet)
474 .inspect_err(|error| pp_error!(self, "write_interleaved failed: {error}")),
475 MediaBuffer::Eos => self.finish(),
476 other => Err(HlsMuxerError::UnsupportedBuffer(other.kind()).into()),
477 }
478 }
479
480 fn control(&mut self, msg: ControlMsg) -> Result<()> {
481 if msg == ControlMsg::Stop {
482 self.finish()?;
483 }
484 Ok(())
485 }
486}
487
488#[cfg(test)]
489mod tests {
490 use std::time::{SystemTime, UNIX_EPOCH};
491
492 use super::*;
493 use crate::{
494 element::Source,
495 elements::{AudioCodec, SwAudioEncoder, SwAudioEncoderOptions},
496 };
497
498 fn open_aac_encoder(sample_rate: u32, channels: u16) -> SwAudioEncoder {
499 SwAudioEncoder::new(
500 "encoder",
501 SwAudioEncoderOptions {
502 codec: AudioCodec::Aac,
503 sample_rate,
504 channels,
505 time_base: ffmpeg::Rational::new(1, sample_rate as i32),
506 bit_rate: 64_000,
507 },
508 )
509 .expect("aac encoder must be available")
510 }
511
512 fn silent_frame(
513 sample_rate: u32,
514 channels: u16,
515 samples: usize,
516 pts: i64,
517 ) -> ffmpeg::frame::Audio {
518 let mut frame = ffmpeg::frame::Audio::new(
519 ffmpeg::format::Sample::F32(ffmpeg::format::sample::Type::Packed),
520 samples,
521 ffmpeg::ChannelLayout::default(channels as i32),
522 );
523 frame.set_rate(sample_rate);
524 frame.set_pts(Some(pts));
525 frame.data_mut(0).fill(0);
526 frame
527 }
528
529 fn unique_test_dir(label: &str) -> PathBuf {
530 let nonce = SystemTime::now()
531 .duration_since(UNIX_EPOCH)
532 .unwrap()
533 .as_nanos();
534 std::env::temp_dir().join(format!("media_pp_{label}_{}_{}", std::process::id(), nonce))
535 }
536
537 fn encode_silence(options: HlsOptions, ticks: i64) {
538 let mut encoder = open_aac_encoder(48_000, 1);
539 let mut muxer = HlsMuxer::create(options).expect("HLS muxer must open");
540 muxer
541 .add_stream(
542 "audio",
543 encoder.parameters(),
544 ffmpeg::Rational::new(1, 48_000),
545 )
546 .expect("add_stream must succeed");
547 let mut sinks = muxer.open().expect("HLS header must be written");
548 encoder.src_pads()[0].link(sinks.pop().unwrap());
549
550 for tick in 0..ticks {
551 encoder
552 .consume(MediaBuffer::Audio(Arc::new(silent_frame(
553 48_000,
554 1,
555 960,
556 tick * 960,
557 ))))
558 .expect("encoding and muxing must succeed");
559 }
560 encoder
561 .consume(MediaBuffer::Eos)
562 .expect("EOS must finalize the HLS playlist");
563 }
564
565 #[test]
566 fn segment_pattern_requires_an_integer_conversion() {
567 assert!(has_integer_conversion("segment_%d.m4s"));
568 assert!(has_integer_conversion("segment_%05d.m4s"));
569 assert!(!has_integer_conversion("segment_%%05d.m4s"));
570 assert!(!has_integer_conversion("segment.m4s"));
571 }
572
573 #[test]
574 fn writes_a_playable_fmp4_vod_playlist() {
575 let dir = unique_test_dir("hls_vod");
576 std::fs::create_dir_all(&dir).unwrap();
577 let playlist_path = dir.join("index.m3u8");
578 let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.m4s"));
579 options.segment_duration = Duration::from_secs(1);
580 options.mode = HlsMode::Vod;
581
582 encode_silence(options, 80);
583
584 let playlist = std::fs::read_to_string(&playlist_path).unwrap();
585 assert!(playlist.starts_with("#EXTM3U"));
586 assert!(playlist.contains("#EXT-X-PLAYLIST-TYPE:VOD"));
587 assert!(playlist.contains("#EXT-X-MAP:URI=\"init.mp4\""));
588 assert!(playlist.contains("#EXT-X-ENDLIST"));
589
590 let segment_uris: Vec<_> = playlist
591 .lines()
592 .filter(|line| !line.is_empty() && !line.starts_with('#'))
593 .collect();
594 assert!(
595 segment_uris.len() >= 2,
596 "expected multiple media segments, got {segment_uris:?}\n{playlist}"
597 );
598 assert!(dir.join("init.mp4").metadata().unwrap().len() > 0);
599 for uri in segment_uris {
600 let path = PathBuf::from(uri);
601 let path = if path.is_absolute() {
602 path
603 } else {
604 dir.join(path)
605 };
606 assert!(
607 path.metadata().unwrap().len() > 0,
608 "empty segment: {path:?}"
609 );
610 }
611
612 let mut input = ffmpeg::format::input(&playlist_path)
613 .expect("FFmpeg must be able to read the generated playlist");
614 assert_eq!(input.streams().count(), 1);
615 let mut packet = ffmpeg::Packet::empty();
616 packet
617 .read(&mut input)
618 .expect("the generated playlist must contain media packets");
619 drop(input);
620 std::fs::remove_dir_all(&dir).unwrap();
621 }
622
623 #[test]
624 fn writes_a_playable_mpegts_vod_playlist() {
625 let dir = unique_test_dir("hls_mpegts");
626 std::fs::create_dir_all(&dir).unwrap();
627 let playlist_path = dir.join("index.m3u8");
628 let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.ts"));
629 options.segment_duration = Duration::from_secs(1);
630 options.mode = HlsMode::Vod;
631 options.segment_format = HlsSegmentFormat::MpegTs;
632
633 encode_silence(options, 80);
634
635 let playlist = std::fs::read_to_string(&playlist_path).unwrap();
636 assert!(playlist.contains("#EXT-X-PLAYLIST-TYPE:VOD"));
637 assert!(!playlist.contains("#EXT-X-MAP"));
638 assert!(playlist.contains("#EXT-X-ENDLIST"));
639 assert!(
640 playlist
641 .lines()
642 .filter(|line| !line.starts_with('#'))
643 .any(|line| line.ends_with(".ts"))
644 );
645
646 let mut input = ffmpeg::format::input(&playlist_path)
647 .expect("FFmpeg must be able to read the generated MPEG-TS playlist");
648 assert_eq!(input.streams().count(), 1);
649 let mut packet = ffmpeg::Packet::empty();
650 packet.read(&mut input).unwrap();
651 drop(input);
652 std::fs::remove_dir_all(&dir).unwrap();
653 }
654
655 #[test]
656 fn live_playlist_keeps_its_window_and_deletes_old_segments() {
657 let dir = unique_test_dir("hls_live");
658 std::fs::create_dir_all(&dir).unwrap();
659 let playlist_path = dir.join("index.m3u8");
660 let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.m4s"));
661 options.segment_duration = Duration::from_secs(1);
662 options.mode = HlsMode::Live {
663 window_size: 2,
664 delete_old_segments: true,
665 };
666
667 encode_silence(options, 200);
668
669 let playlist = std::fs::read_to_string(&playlist_path).unwrap();
670 let segment_uris: Vec<_> = playlist
671 .lines()
672 .filter(|line| !line.is_empty() && !line.starts_with('#'))
673 .collect();
674 assert_eq!(segment_uris.len(), 2, "{playlist}");
675 let media_sequence = playlist
676 .lines()
677 .find_map(|line| line.strip_prefix("#EXT-X-MEDIA-SEQUENCE:"))
678 .unwrap()
679 .parse::<u64>()
680 .unwrap();
681 assert!(media_sequence > 0, "{playlist}");
682 assert!(playlist.contains("#EXT-X-ENDLIST"));
683
684 let media_files = std::fs::read_dir(&dir)
685 .unwrap()
686 .filter_map(|entry| entry.ok())
687 .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "m4s"))
688 .count();
689 assert!(
690 (2..=3).contains(&media_files),
691 "the two listed segments plus at most one deletion-threshold file should remain; \
692 found {media_files}"
693 );
694 assert!(
695 std::fs::read_dir(&dir)
696 .unwrap()
697 .filter_map(|entry| entry.ok())
698 .all(|entry| entry.path().extension().is_none_or(|ext| ext != "tmp")),
699 "atomic temp files must not remain after finalization"
700 );
701 std::fs::remove_dir_all(&dir).unwrap();
702 }
703
704 #[test]
705 fn playlist_finalizes_only_after_every_track_finishes() {
706 let dir = unique_test_dir("hls_tracks");
707 std::fs::create_dir_all(&dir).unwrap();
708 let playlist_path = dir.join("index.m3u8");
709 let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.m4s"));
710 options.segment_duration = Duration::from_secs(1);
711 options.mode = HlsMode::Vod;
712
713 let mut encoder_a = open_aac_encoder(48_000, 1);
714 let mut encoder_b = open_aac_encoder(48_000, 1);
715 let mut muxer = HlsMuxer::create(options).unwrap();
716 muxer
717 .add_stream(
718 "audio-a",
719 encoder_a.parameters(),
720 ffmpeg::Rational::new(1, 48_000),
721 )
722 .unwrap();
723 muxer
724 .add_stream(
725 "audio-b",
726 encoder_b.parameters(),
727 ffmpeg::Rational::new(1, 48_000),
728 )
729 .unwrap();
730 let mut sinks = muxer.open().unwrap();
731 let sink_b = sinks.pop().unwrap();
732 let sink_a = sinks.pop().unwrap();
733 encoder_a.src_pads()[0].link(sink_a);
734 encoder_b.src_pads()[0].link(sink_b);
735
736 for tick in 0..60i64 {
737 encoder_a
738 .consume(MediaBuffer::Audio(Arc::new(silent_frame(
739 48_000,
740 1,
741 960,
742 tick * 960,
743 ))))
744 .unwrap();
745 encoder_b
746 .consume(MediaBuffer::Audio(Arc::new(silent_frame(
747 48_000,
748 1,
749 960,
750 tick * 960,
751 ))))
752 .unwrap();
753 }
754
755 encoder_a.consume(MediaBuffer::Eos).unwrap();
756 if let Ok(unfinished) = std::fs::read_to_string(&playlist_path) {
757 assert!(
758 !unfinished.contains("#EXT-X-ENDLIST"),
759 "the first finished track must not finalize the shared playlist"
760 );
761 }
762
763 for tick in 60..80i64 {
764 encoder_b
765 .consume(MediaBuffer::Audio(Arc::new(silent_frame(
766 48_000,
767 1,
768 960,
769 tick * 960,
770 ))))
771 .unwrap();
772 }
773 encoder_b.consume(MediaBuffer::Eos).unwrap();
774 let finished = std::fs::read_to_string(&playlist_path).unwrap();
775 assert!(finished.contains("#EXT-X-ENDLIST"));
776
777 drop(encoder_a);
778 drop(encoder_b);
779 std::fs::remove_dir_all(&dir).unwrap();
780 }
781}