Skip to main content

arcly_stream/packager/
mod.rs

1//! Egress packaging: turn a live frame stream into HLS/LL-HLS segments.
2//!
3//! Gated behind `hls`. Provides the [`Muxer`] and [`Packager`] contracts, a
4//! sliding-window [`HlsPlaylist`] generator, and an [`HlsSegmenter`] that cuts
5//! segments on keyframe boundaries and writes both media segments and the
6//! `.m3u8` playlist through any [`StorageBackend`].
7//!
8//! The container byte-format is pluggable via [`Muxer`]: a [`PassthroughMuxer`]
9//! (elementary-stream concatenation) ships here; production MPEG-TS or fMP4/CMAF
10//! muxers implement the same trait and drop in unchanged.
11//!
12//! ```no_run
13//! # #[cfg(feature = "storage-fs")]
14//! # async fn demo(handle: arcly_stream::StreamHandle) -> arcly_stream::Result<()> {
15//! use arcly_stream::packager::{HlsSegmenter, Packager, PassthroughMuxer};
16//! use arcly_stream::storage::FsStorage;
17//!
18//! let storage = FsStorage::new("/var/hls");
19//! let mut seg = HlsSegmenter::new(PassthroughMuxer::new("ts"), storage, "live/cam", 6, 5);
20//! let mut sub = handle.subscribe_resilient();
21//! while let Some(frame) = sub.recv().await {
22//!     seg.push(&frame).await?;
23//! }
24//! seg.finish().await?;
25//! # Ok(())
26//! # }
27//! ```
28
29mod engine;
30mod playlist;
31
32use engine::{SegmentEngine, BANDWIDTH_HINT};
33
34#[cfg(feature = "mpegts")]
35#[cfg_attr(docsrs, doc(cfg(feature = "mpegts")))]
36mod mpegts;
37
38#[cfg(feature = "mpegts")]
39#[cfg_attr(docsrs, doc(cfg(feature = "mpegts")))]
40pub use mpegts::MpegTsMuxer;
41
42#[cfg(feature = "fmp4")]
43#[cfg_attr(docsrs, doc(cfg(feature = "fmp4")))]
44mod fmp4;
45
46#[cfg(feature = "fmp4")]
47#[cfg_attr(docsrs, doc(cfg(feature = "fmp4")))]
48pub use fmp4::Fmp4Muxer;
49
50pub use playlist::{render_master, HlsPlaylist, Part, Segment};
51
52#[cfg(feature = "dash")]
53#[cfg_attr(docsrs, doc(cfg(feature = "dash")))]
54mod dash;
55
56#[cfg(feature = "dash")]
57#[cfg_attr(docsrs, doc(cfg(feature = "dash")))]
58pub use dash::{DashManifest, DashPackager};
59
60use crate::traits::StorageBackend;
61use crate::{CodecId, FrameFlags, MediaFrame, Result};
62use async_trait::async_trait;
63use bytes::{BufMut, Bytes, BytesMut};
64
65/// Container muxer: accumulates frames into one segment's bytes.
66///
67/// Implement this for MPEG-TS, fMP4/CMAF, etc. The segmenter calls
68/// [`start_segment`](Self::start_segment) at each boundary, [`write`](Self::write)
69/// per frame, and [`finish_segment`](Self::finish_segment) to take the bytes.
70pub trait Muxer: Send {
71    /// File extension for produced segments (e.g. `"ts"`, `"m4s"`).
72    fn extension(&self) -> &'static str;
73
74    /// Begin a fresh segment, discarding any partial state.
75    fn start_segment(&mut self) -> Result<()>;
76
77    /// Append one frame to the current segment.
78    fn write(&mut self, frame: &MediaFrame) -> Result<()>;
79
80    /// Take the finished segment's bytes, resetting for the next one.
81    fn finish_segment(&mut self) -> Result<Bytes>;
82
83    /// The exact media duration (ms) of the most recently finished segment, when
84    /// the container can report it (fMP4 sums its sample durations). The packager
85    /// prefers this for the manifest/playlist so the declared segment length
86    /// matches the media's internal decode-time advance — independently derived
87    /// durations drift and stutter the player at each boundary. `None` (the
88    /// default) means "use the caller's measured duration".
89    fn last_segment_duration_ms(&self) -> Option<u64> {
90        None
91    }
92
93    /// Flush the frames buffered since the last part/segment start as an
94    /// independently-appendable **partial segment** (LL-HLS `#EXT-X-PART`),
95    /// *without* ending the current segment. Returns `None` when nothing is
96    /// buffered or the container doesn't support parts.
97    ///
98    /// The concatenation of a segment's parts must equal the full segment, so a
99    /// container's parts are self-contained chunks (fMP4: one `moof`+`mdat`
100    /// fragment each). The default returns `None` (no LL-HLS parts).
101    fn take_partial(&mut self) -> Result<Option<Bytes>> {
102        Ok(None)
103    }
104
105    /// Build an fMP4 initialization segment (`ftyp`+`moov` carrying the
106    /// `avcC`/`hvcC`/`av1C`/`vvcC` decoder config) from a CONFIG access unit.
107    ///
108    /// Returning `Some` makes [`HlsSegmenter`] write it once and reference it via
109    /// `#EXT-X-MAP`. The default returns `None` — correct for self-initializing
110    /// containers like MPEG-TS and the [`PassthroughMuxer`].
111    fn init_segment(&mut self, _codec: CodecId, _config_record: &[u8]) -> Result<Option<Bytes>> {
112        Ok(None)
113    }
114
115    /// Build the initialization segment from **all** the stream's CONFIG access
116    /// units seen before the first media sample — e.g. an `(H264, avc)` video
117    /// config plus an `(AAC, asc)` audio config — so a container that carries
118    /// multiple tracks (fMP4 with muxed audio) can emit a multi-track `moov`.
119    ///
120    /// The default delegates to [`init_segment`](Self::init_segment) with the
121    /// first config, so single-track and self-initializing muxers (MPEG-TS,
122    /// passthrough) need no changes. The [`Fmp4Muxer`] overrides this
123    /// to add an audio track when an AAC config is present.
124    fn build_init_from(&mut self, configs: &[(CodecId, Bytes)]) -> Result<Option<Bytes>> {
125        match configs.first() {
126            Some((codec, data)) => self.init_segment(*codec, data),
127            None => Ok(None),
128        }
129    }
130
131    /// The HLS `CODECS` attribute for this rendition's master-playlist entry
132    /// (e.g. `"hvc1.1.6.L120.B0"`), once known.
133    fn codec_string(&self) -> Option<String> {
134        None
135    }
136}
137
138/// A trivial muxer that concatenates frame payloads (elementary stream).
139///
140/// Useful for testing and raw recording; not a real container. Swap in a TS or
141/// fMP4 muxer for player-compatible output.
142pub struct PassthroughMuxer {
143    ext: &'static str,
144    buf: BytesMut,
145}
146
147impl PassthroughMuxer {
148    /// New passthrough muxer producing segments with the given extension.
149    pub fn new(ext: &'static str) -> Self {
150        Self {
151            ext,
152            buf: BytesMut::new(),
153        }
154    }
155}
156
157impl Muxer for PassthroughMuxer {
158    fn extension(&self) -> &'static str {
159        self.ext
160    }
161    fn start_segment(&mut self) -> Result<()> {
162        self.buf.clear();
163        Ok(())
164    }
165    fn write(&mut self, frame: &MediaFrame) -> Result<()> {
166        self.buf.put_slice(&frame.data);
167        Ok(())
168    }
169    fn finish_segment(&mut self) -> Result<Bytes> {
170        Ok(std::mem::take(&mut self.buf).freeze())
171    }
172    fn take_partial(&mut self) -> Result<Option<Bytes>> {
173        if self.buf.is_empty() {
174            return Ok(None);
175        }
176        Ok(Some(std::mem::take(&mut self.buf).freeze()))
177    }
178}
179
180/// Receives a segment's CMAF fragments (`moof`+`mdat`) as they are produced, so a
181/// delivery layer can stream the in-progress segment over HTTP chunked transfer
182/// (real LL-DASH) instead of waiting for the whole segment file.
183///
184/// The chunks for one segment are exactly the [`Muxer::take_partial`] outputs,
185/// whose concatenation equals the full segment file — so a client reading the
186/// chunked stream and a client fetching the completed file see identical bytes.
187pub trait ChunkSink: Send + Sync {
188    /// A new segment `seq` has started; any chunks buffered for a previous open
189    /// segment should be considered closed.
190    fn open(&self, seq: u64);
191    /// One CMAF fragment of segment `seq`.
192    fn chunk(&self, seq: u64, bytes: Bytes);
193    /// Segment `seq` is complete (its full file has been written).
194    fn close(&self, seq: u64);
195}
196
197/// Consumes a live frame stream and produces a packaged rendition.
198#[async_trait]
199pub trait Packager: Send {
200    /// Feed one frame; may finalize and emit a segment as a side effect.
201    async fn push(&mut self, frame: &MediaFrame) -> Result<()>;
202    /// Flush the final segment and close the rendition.
203    async fn finish(&mut self) -> Result<()>;
204}
205
206/// Keyframe-boundary HLS segmenter writing through a [`StorageBackend`].
207///
208/// Starts at the first keyframe, cuts a new segment at the first keyframe at or
209/// after `target_duration` seconds, and after each cut writes the media segment
210/// and the regenerated `index.m3u8` to storage under `prefix`.
211pub struct HlsSegmenter<M: Muxer, S: StorageBackend> {
212    engine: SegmentEngine<M, S>,
213    playlist: HlsPlaylist,
214    /// Whether the multivariant (master) playlist has been written yet. It is
215    /// emitted once, as soon as the rendition's `CODECS` string is known.
216    master_written: bool,
217    /// The HLS `CODECS` string, derived from the CONFIG access unit. Cached here
218    /// (not just read from the muxer) because the segment clock skips the CONFIG
219    /// frame before it reaches a self-initializing muxer like MPEG-TS, so the
220    /// muxer would never see it — yet the master playlist still needs it.
221    codec_string: Option<String>,
222    // ── LL-HLS partial-segment state (inactive when `part_target_ms == 0`) ──
223    /// Part target in milliseconds; `0` disables LL-HLS part emission.
224    part_target_ms: i64,
225    /// Index of the next part within the in-progress segment.
226    part_idx: u64,
227    /// PTS (ms) at which the current part began, once a segment is open.
228    part_anchor_pts: Option<i64>,
229    /// PTS (ms) of the most recently written frame (for the final part's length).
230    last_pts: i64,
231    /// Byte chunks of the in-progress segment's parts, concatenated into the full
232    /// segment file at the cut so non-LL clients can fetch the whole segment.
233    seg_part_bytes: Vec<Bytes>,
234}
235
236impl<M: Muxer, S: StorageBackend> HlsSegmenter<M, S> {
237    /// New segmenter writing under `prefix`, targeting `target_duration`-second
238    /// segments and a `window`-segment live playlist.
239    pub fn new(
240        muxer: M,
241        storage: S,
242        prefix: impl Into<String>,
243        target_duration: u64,
244        window: usize,
245    ) -> Self {
246        Self {
247            engine: SegmentEngine::new(muxer, storage, prefix, target_duration),
248            playlist: HlsPlaylist::new(target_duration, window),
249            master_written: false,
250            codec_string: None,
251            part_target_ms: 0,
252            part_idx: 0,
253            part_anchor_pts: None,
254            last_pts: 0,
255            seg_part_bytes: Vec::new(),
256        }
257    }
258
259    /// The HLS `CODECS` attribute for this rendition's master-playlist entry.
260    pub fn codec_string(&self) -> Option<String> {
261        self.codec_string
262            .clone()
263            .or_else(|| self.engine.muxer.codec_string())
264    }
265
266    /// On the first CONFIG access unit: write the muxer's fMP4 init segment (if
267    /// any) and reference it via `#EXT-X-MAP`, and derive the HLS `CODECS` string
268    /// from the parameter sets. The codec string is read straight from the CONFIG
269    /// here — not from the muxer — because the segment clock skips the CONFIG
270    /// frame before a self-initializing muxer (MPEG-TS) ever sees it.
271    async fn ensure_init_segment(&mut self, frame: &MediaFrame) -> Result<()> {
272        if self.codec_string.is_none()
273            && frame.is_video()
274            && frame.flags.contains(FrameFlags::CONFIG)
275        {
276            self.codec_string = crate::codec::dispatch::parse_config(frame.codec, &frame.data)
277                .and_then(|p| crate::codec::dispatch::hls_codec_string(frame.codec, &p));
278        }
279        if let Some(uri) = self.engine.ensure_init(frame).await? {
280            self.playlist.set_map(uri);
281        }
282        Ok(())
283    }
284
285    /// Enable LL-HLS output: the playlist advertises parts and the segmenter
286    /// emits a partial segment roughly every `part_target` seconds (writing the
287    /// part files and `#EXT-X-PART`/preload-hint lines) in addition to full
288    /// segments. Requires a muxer that implements
289    /// [`take_partial`](Muxer::take_partial) (fMP4 / passthrough).
290    pub fn low_latency(mut self, part_target: f64) -> Self {
291        self.playlist = self.playlist.low_latency(part_target);
292        self.part_target_ms = (part_target.max(0.05) * 1000.0) as i64;
293        self
294    }
295
296    /// The storage key of the media playlist.
297    pub fn playlist_key(&self) -> String {
298        format!("{}/index.m3u8", self.engine.prefix)
299    }
300
301    /// The storage key of the multivariant (master) playlist.
302    pub fn master_key(&self) -> String {
303        format!("{}/master.m3u8", self.engine.prefix)
304    }
305
306    /// Once the muxer can report a `CODECS` string, write the multivariant
307    /// (master) playlist a single time. This is what advertises `CODECS="hvc1.*"`
308    /// for HEVC (and `av01.*`/`vvc1.*`); without it players such as Safari will
309    /// not decode H.265, falling back as if the rendition were unplayable.
310    async fn ensure_master_playlist(&mut self) -> Result<()> {
311        if self.master_written {
312            return Ok(());
313        }
314        let Some(codecs) = self.codec_string() else {
315            return Ok(());
316        };
317        let body = playlist::render_master("index.m3u8", Some(&codecs), BANDWIDTH_HINT);
318        self.engine
319            .storage
320            .put(
321                &self.engine.key("master.m3u8"),
322                Bytes::from(body.into_bytes()),
323            )
324            .await?;
325        self.master_written = true;
326        Ok(())
327    }
328
329    /// Flush the buffered samples as one LL-HLS partial segment: write the part
330    /// file, append `#EXT-X-PART` + a preload hint for the next part, and stash
331    /// the bytes so the full segment file can be assembled at the cut.
332    async fn flush_part(&mut self, end_pts: i64) -> Result<()> {
333        let Some(bytes) = self.engine.muxer.take_partial()? else {
334            return Ok(());
335        };
336        let anchor = self.part_anchor_pts.unwrap_or(end_pts);
337        // Prefer the muxer's exact fragment duration (sum of sample durations,
338        // just set by `take_partial`) over the DTS span: the span `end - anchor`
339        // excludes the last frame's own duration, so the declared PART duration
340        // drifts ~one frame short of the media every part and the LL-HLS player
341        // stutters at part boundaries. Aligning them keeps playback smooth.
342        let duration = self
343            .engine
344            .muxer
345            .last_segment_duration_ms()
346            .filter(|&d| d > 0)
347            .map(|d| d as f64 / 1000.0)
348            .unwrap_or_else(|| (end_pts - anchor).max(0) as f64 / 1000.0);
349        let uri = self.engine.part_uri(self.engine.seq, self.part_idx);
350        let key = self.engine.key(&uri);
351        self.engine.storage.put(&key, bytes.clone()).await?;
352        self.seg_part_bytes.push(bytes);
353        self.playlist.add_pending_part(Part {
354            uri,
355            duration,
356            // The first part of a segment begins on the anchoring keyframe.
357            independent: self.part_idx == 0,
358        });
359        self.playlist
360            .set_preload_hint(self.engine.part_uri(self.engine.seq, self.part_idx + 1));
361        self.part_idx += 1;
362        self.part_anchor_pts = Some(end_pts);
363        self.write_playlist().await
364    }
365
366    async fn cut(&mut self, duration: f64) -> Result<()> {
367        let uri = self.engine.segment_uri(self.engine.seq);
368
369        if self.part_target_ms > 0 {
370            // LL-HLS: flush the segment's final part, then assemble the full
371            // segment file from its parts (concatenation == the whole segment).
372            self.flush_part(self.last_pts).await?;
373            let mut full = BytesMut::new();
374            for b in &self.seg_part_bytes {
375                full.put_slice(b);
376            }
377            // Defensive: if the muxer produced no parts (it doesn't implement
378            // `take_partial`), fall back to a whole-segment flush so LL-HLS still
379            // emits a non-empty segment instead of a 0-byte file.
380            let full = if full.is_empty() {
381                self.engine.muxer.finish_segment()?
382            } else {
383                full.freeze()
384            };
385            let key = self.engine.key(&uri);
386            self.engine.storage.put(&key, full).await?;
387            self.playlist.commit_segment(Segment {
388                seq: self.engine.seq,
389                duration,
390                uri,
391                discontinuity: false,
392                parts: Vec::new(), // filled from pending parts by commit_segment
393            });
394            self.seg_part_bytes.clear();
395            self.part_idx = 0;
396            self.part_anchor_pts = None;
397        } else {
398            let bytes = self.engine.muxer.finish_segment()?;
399            let key = self.engine.key(&uri);
400            self.engine.storage.put(&key, bytes).await?;
401            self.playlist.push(Segment {
402                seq: self.engine.seq,
403                duration,
404                uri,
405                discontinuity: false,
406                parts: Vec::new(),
407            });
408        }
409        self.write_playlist().await?;
410        self.engine.seq += 1;
411        Ok(())
412    }
413
414    async fn write_playlist(&mut self) -> Result<()> {
415        let body = self.playlist.render();
416        self.engine
417            .storage
418            .put(
419                &self.engine.key("index.m3u8"),
420                Bytes::from(body.into_bytes()),
421            )
422            .await
423    }
424}
425
426#[async_trait]
427impl<M: Muxer, S: StorageBackend> Packager for HlsSegmenter<M, S> {
428    async fn push(&mut self, frame: &MediaFrame) -> Result<()> {
429        self.ensure_init_segment(frame).await?;
430        // Emit the master playlist as soon as the codec string is known — which
431        // `ensure_init_segment` derives from the CONFIG frame. This must run
432        // before the clock-skip below, since the clock skips that very CONFIG
433        // frame (it is not a keyframe), and HEVC/AV1 are unviewable without it.
434        self.ensure_master_playlist().await?;
435        let decision = self.engine.observe(frame);
436        if decision.skip {
437            return Ok(());
438        }
439        if let Some(duration) = decision.cut_previous {
440            self.cut(duration).await?;
441        }
442        if decision.open_new {
443            self.engine.muxer.start_segment()?;
444            // A fresh segment anchors a new part run at this keyframe. Use DTS
445            // (decode/arrival order) so part timing stays monotonic with B-frame
446            // sources — PTS reorders and would make part boundaries jitter.
447            self.part_anchor_pts = Some(frame.dts);
448            self.part_idx = 0;
449            self.seg_part_bytes.clear();
450        }
451        self.engine.muxer.write(frame)?;
452        self.last_pts = frame.dts;
453
454        // LL-HLS: emit a partial segment once the part target has elapsed. Parts
455        // are cut at frame boundaries, so a part never splits an access unit.
456        if self.part_target_ms > 0 {
457            if let Some(anchor) = self.part_anchor_pts {
458                if frame.dts - anchor >= self.part_target_ms {
459                    self.flush_part(frame.dts).await?;
460                }
461            }
462        }
463        Ok(())
464    }
465
466    async fn finish(&mut self) -> Result<()> {
467        if let Some(duration) = self.engine.flush() {
468            self.cut(duration).await?;
469        }
470        self.playlist.finish();
471        self.write_playlist().await
472    }
473}
474
475#[cfg(test)]
476mod tests {
477    use super::*;
478    use crate::testing::{video_frame, InMemoryStorage};
479
480    #[tokio::test]
481    async fn segments_on_keyframe_after_target_and_writes_playlist() {
482        let store = InMemoryStorage::new();
483        let mut seg =
484            HlsSegmenter::new(PassthroughMuxer::new("ts"), store.clone(), "live/cam", 2, 5);
485
486        // Keyframes every 1s; with a 2s target, a cut happens at the keyframe
487        // at/after 2s. pts in ms.
488        for i in 0..5 {
489            let pts = i * 1000;
490            seg.push(&video_frame(pts, true)).await.unwrap();
491            // a delta in between
492            seg.push(&video_frame(pts + 500, false)).await.unwrap();
493        }
494        seg.finish().await.unwrap();
495
496        let playlist = store.get("live/cam/index.m3u8").await.unwrap();
497        let text = String::from_utf8(playlist.to_vec()).unwrap();
498        assert!(text.contains("#EXTM3U"));
499        assert!(text.contains("#EXT-X-ENDLIST"));
500        // At least the first full segment must have been written to storage.
501        assert!(store.get("live/cam/seg0.ts").await.is_ok());
502        // Segment payload is the concatenation of its frames' bytes.
503        assert!(!store.get("live/cam/seg0.ts").await.unwrap().is_empty());
504    }
505
506    #[tokio::test]
507    async fn ll_hls_emits_part_files_and_ext_x_part() {
508        let store = InMemoryStorage::new();
509        // 2s segments, parts every ~0.5s.
510        let mut seg =
511            HlsSegmenter::new(PassthroughMuxer::new("m4s"), store.clone(), "live/ll", 2, 5)
512                .low_latency(0.5);
513
514        // Keyframe at 0, deltas every 250ms; next keyframe at 2000 cuts segment 0.
515        seg.push(&video_frame(0, true)).await.unwrap();
516        for i in 1..8 {
517            seg.push(&video_frame(i * 250, false)).await.unwrap();
518        }
519        seg.push(&video_frame(2000, true)).await.unwrap(); // cut seg 0
520        seg.push(&video_frame(2250, false)).await.unwrap();
521        seg.finish().await.unwrap();
522
523        let text =
524            String::from_utf8(store.get("live/ll/index.m3u8").await.unwrap().to_vec()).unwrap();
525        // LL-HLS headers + actual part lines for segment 0.
526        assert!(text.contains("#EXT-X-PART-INF:PART-TARGET=0.500"));
527        assert!(
528            text.contains("#EXT-X-PART:DURATION="),
529            "renders EXT-X-PART lines"
530        );
531        assert!(
532            text.contains("INDEPENDENT=YES"),
533            "first part is independent"
534        );
535        // Part files were written, and the first full segment too.
536        assert!(
537            store.get("live/ll/seg0.0.m4s").await.is_ok(),
538            "part 0 written"
539        );
540        assert!(
541            store.get("live/ll/seg0.1.m4s").await.is_ok(),
542            "part 1 written"
543        );
544        assert!(
545            store.get("live/ll/seg0.m4s").await.is_ok(),
546            "full segment 0 written"
547        );
548
549        // The full segment equals the concatenation of its parts.
550        let mut concat = Vec::new();
551        let mut p = 0;
552        while let Ok(part) = store.get(&format!("live/ll/seg0.{p}.m4s")).await {
553            concat.extend_from_slice(&part);
554            p += 1;
555        }
556        assert!(p >= 2, "segment 0 had multiple parts");
557        let full = store.get("live/ll/seg0.m4s").await.unwrap();
558        assert_eq!(&concat[..], &full[..], "full segment == concat of parts");
559    }
560
561    #[tokio::test]
562    async fn ll_hls_falls_back_to_full_segment_when_muxer_has_no_parts() {
563        // InitMuxer does NOT implement `take_partial`; in LL-HLS mode the cut path
564        // must still write a non-empty segment (regression: 0-byte .ts segments).
565        let store = InMemoryStorage::new();
566        let mut seg = HlsSegmenter::new(
567            InitMuxer {
568                buf: BytesMut::new(),
569            },
570            store.clone(),
571            "live/nopart",
572            2,
573            5,
574        )
575        .low_latency(0.5);
576
577        seg.push(&video_frame(0, true)).await.unwrap();
578        for i in 1..6 {
579            seg.push(&video_frame(i * 1000, i % 2 == 0)).await.unwrap();
580        }
581        seg.finish().await.unwrap();
582
583        let seg0 = store.get("live/nopart/seg0.m4s").await.unwrap();
584        assert!(!seg0.is_empty(), "LL-HLS segment must not be empty");
585    }
586
587    #[tokio::test]
588    async fn writes_master_playlist_with_hevc_codecs() {
589        use crate::FrameFlags;
590        let store = InMemoryStorage::new();
591        let mut seg = HlsSegmenter::new(
592            InitMuxer {
593                buf: BytesMut::new(),
594            },
595            store.clone(),
596            "live/hevc",
597            2,
598            5,
599        );
600
601        let mut cfg = video_frame(0, true);
602        cfg.codec = CodecId::H265;
603        cfg.flags |= FrameFlags::CONFIG;
604        seg.push(&cfg).await.unwrap();
605        for i in 1..4 {
606            seg.push(&video_frame(i * 1000, true)).await.unwrap();
607        }
608        seg.finish().await.unwrap();
609
610        let master =
611            String::from_utf8(store.get("live/hevc/master.m3u8").await.unwrap().to_vec()).unwrap();
612        assert!(master.contains("#EXT-X-STREAM-INF:"));
613        assert!(
614            master.contains("CODECS=\"hvc1.1.6.L120.B0\""),
615            "master advertises HEVC codec string: {master}"
616        );
617        assert!(master.contains("index.m3u8"));
618    }
619
620    /// A muxer that emits an fMP4-style init segment, to exercise EXT-X-MAP.
621    struct InitMuxer {
622        buf: BytesMut,
623    }
624    impl Muxer for InitMuxer {
625        fn extension(&self) -> &'static str {
626            "m4s"
627        }
628        fn start_segment(&mut self) -> Result<()> {
629            self.buf.clear();
630            Ok(())
631        }
632        fn write(&mut self, frame: &MediaFrame) -> Result<()> {
633            self.buf.put_slice(&frame.data);
634            Ok(())
635        }
636        fn finish_segment(&mut self) -> Result<Bytes> {
637            Ok(std::mem::take(&mut self.buf).freeze())
638        }
639        fn init_segment(&mut self, _codec: CodecId, config_record: &[u8]) -> Result<Option<Bytes>> {
640            // A real muxer would wrap this in ftyp+moov; here we just echo it.
641            Ok(Some(Bytes::copy_from_slice(config_record)))
642        }
643        fn codec_string(&self) -> Option<String> {
644            Some("hvc1.1.6.L120.B0".into())
645        }
646    }
647
648    #[tokio::test]
649    async fn fmp4_muxer_writes_init_segment_and_ext_x_map() {
650        use crate::FrameFlags;
651        let store = InMemoryStorage::new();
652        let mut seg = HlsSegmenter::new(
653            InitMuxer {
654                buf: BytesMut::new(),
655            },
656            store.clone(),
657            "live/hevc",
658            2,
659            5,
660        );
661        assert_eq!(seg.codec_string().as_deref(), Some("hvc1.1.6.L120.B0"));
662
663        // First frame carries CONFIG → triggers the init segment.
664        let mut cfg = video_frame(0, true);
665        cfg.codec = CodecId::H265;
666        cfg.flags |= FrameFlags::CONFIG;
667        seg.push(&cfg).await.unwrap();
668        for i in 1..6 {
669            seg.push(&video_frame(i * 1000, true)).await.unwrap();
670        }
671        seg.finish().await.unwrap();
672
673        assert!(store.get("live/hevc/init.m4s").await.is_ok());
674        let pl =
675            String::from_utf8(store.get("live/hevc/index.m3u8").await.unwrap().to_vec()).unwrap();
676        assert!(pl.contains("#EXT-X-MAP:URI=\"init.m4s\""));
677    }
678}