Struct HlsOrigin 

pub struct HlsOrigin { /* private fields */ }

Available on crate feature std only.

The LL-HLS origin ServedEgress: renders playlists and resolves blocking-reload/part-availability requests for one stream, backed by a shared Trunk. See this module’s own doc for exactly what comes straight from the Trunk and what needs the small synced Window.

Implementations

impl HlsOrigin

pub fn builder(trunk: Arc<Trunk>) -> HlsOriginBuilder

Start building an HlsOrigin over trunk — see HlsOriginBuilder for the required fields (target_duration_secs/window_segments), the container choice, and how to opt into LL-HLS.

Examples found in repository

examples/client_stepping.rs (line 41)

fn canned_playlist() -> String {
    let trunk = Trunk::new(TrunkConfig::new(nz(16), nz(4), nz(8), nz(4), nz(16)));
    let writer = trunk
        .segment_writer()
        .expect("first (and only) segment writer");
    let origin = HlsOrigin::builder(std::sync::Arc::clone(&trunk))
        .target_duration_secs(1.0)
        .window_segments(nz(4))
        .low_latency(500)
        .build()
        .expect("both required fields set");
    origin.set_init(vec![0xAA; 32]);

    writer.publish_part(PartEntry::new(
        vec![0x01; 16],
        1,
        0,
        Duration::from_millis(500),
        true,
    ));
    writer.publish_segment(SegmentEntry::new(
        vec![0x02; 32],
        1,
        Duration::from_secs(1),
        Timestamp::from_nanos(0),
        SegmentMeta {
            discontinuous: false,
        },
    ));
    writer.publish_part(PartEntry::new(
        vec![0x03; 16],
        2,
        0,
        Duration::from_millis(500),
        true,
    ));

    match origin.resolve(
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
        Timestamp::from_nanos(0),
        AwaitPolicy::new(Timestamp::from_nanos(0)),
    ) {
        EgressResponse::Ready {
            body: HlsBody::Playlist(m),
            ..
        } => m,
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }
}

examples/origin_playlist.rs (line 57)

fn main() {
    let trunk = Trunk::new(TrunkConfig::new(nz(16), nz(4), nz(8), nz(4), nz(16)));
    let writer = trunk
        .segment_writer()
        .expect("first (and only) segment writer");
    let origin = HlsOrigin::builder(std::sync::Arc::clone(&trunk))
        .target_duration_secs(TARGET_DURATION_SECS)
        .window_segments(nz(WINDOW_SEGMENTS))
        .low_latency(PART_TARGET_MS)
        .build()
        .expect("both required fields set");
    origin.set_init(vec![0xAA; 32]);

    // Segment 1 closes with two parts.
    writer.publish_part(PartEntry::new(
        vec![0x01; 16],
        1,
        0,
        Duration::from_millis(500),
        true,
    ));
    writer.publish_part(PartEntry::new(
        vec![0x02; 16],
        1,
        1,
        Duration::from_millis(500),
        false,
    ));
    writer.publish_segment(SegmentEntry::new(
        vec![0x03; 32],
        1,
        Duration::from_secs(1),
        Timestamp::from_nanos(0),
        SegmentMeta {
            discontinuous: false,
        },
    ));

    // Segment 2 is still open, with only its first part landed so far.
    writer.publish_part(PartEntry::new(
        vec![0x04; 16],
        2,
        0,
        Duration::from_millis(500),
        true,
    ));

    println!("--- master.m3u8 ---");
    println!("{}", master_playlist_m3u8("media.m3u8"));

    println!("--- media.m3u8 ---");
    match resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    ) {
        EgressResponse::Ready {
            body: HlsBody::Playlist(m),
            ..
        } => println!("{m}"),
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }

    // A plain (non-blocking) request is Ready immediately.
    let outcome = resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    );
    assert!(matches!(
        outcome,
        EgressResponse::Ready {
            body: HlsBody::Playlist(_),
            ..
        }
    ));
    println!("resolve(Playlist, no query)     -> Ready");

    // A blocking-reload request for a segment that hasn't closed yet: with
    // `await_policy`'s deadline already at `now`, this immediately reports
    // the awaited condition has run out of patience rather than serving a
    // fabricated Ready.
    let outcome = resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            // Segment 2 is the in-progress one, so msn=3 is the next segment
            // that has not closed. Keep this within
            // `ABUSE_MSN_FUTURE_BOUND` (live edge + 2) — a larger value is
            // rejected outright as abuse and never reaches the Await path
            // this block is demonstrating.
            query: BlockingQuery {
                hls_msn: Some(3),
                hls_part: None,
            },
        },
    );
    assert_eq!(outcome, EgressResponse::NotFound);
    println!("resolve(Playlist, _HLS_msn=3)   -> NotFound (Await's patience already expired)");

    // A `_HLS_msn` unreasonably far beyond the live edge is rejected outright
    // (RFC 8216bis §6.2.5.2 abuse prevention) rather than ever Await-ing.
    let outcome = resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery {
                hls_msn: Some(999),
                hls_part: None,
            },
        },
    );
    assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
    println!("resolve(Playlist, _HLS_msn=999) -> BadRequest (abuse bound)");

    // `Resource`: the init segment and the closed segment are Ready...
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "init-1.mp4".to_string(),
        },
    ) {
        EgressResponse::Ready { .. } => println!("resolve(Resource, init-1.mp4)     -> Ready"),
        other => panic!("expected Ready, got {other:?}"),
    }
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "seg-1-1.m4s".to_string(),
        },
    ) {
        EgressResponse::Ready { .. } => println!("resolve(Resource, seg-1-1.m4s)    -> Ready"),
        other => panic!("expected Ready, got {other:?}"),
    }
    // ...a live part of the still-open segment is Ready too...
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "part-1-2.0.m4s".to_string(),
        },
    ) {
        EgressResponse::Ready { .. } => println!("resolve(Resource, part-1-2.0.m4s) -> Ready"),
        other => panic!("expected Ready, got {other:?}"),
    }
    // ...a preload-hinted part not yet produced reports NotFound once this
    // call's `await_policy` has already expired (a real HTTP adapter would
    // instead give it a real deadline and block on `Trunk::listen()`)...
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "part-1-2.1.m4s".to_string(),
        },
    ) {
        EgressResponse::NotFound => {
            println!(
                "resolve(Resource, part-1-2.1.m4s) -> NotFound (Await's patience already expired)"
            )
        }
        other => panic!("expected NotFound, got {other:?}"),
    }
    // ...and an unrecognised filename is a plain 404.
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "nope.txt".to_string(),
        },
    ) {
        EgressResponse::NotFound => println!("resolve(Resource, nope.txt)       -> NotFound"),
        other => panic!("expected NotFound, got {other:?}"),
    }
}

pub fn set_init(&self, bytes: impl Into<Bytes>)

Store the init segment bytes — see this module’s doc for why an init segment is not something any Trunk ring holds.

Documented no-op under Container::MpegTs: this origin’s MpegTs grammar has no init resource and never emits #EXT-X-MAP by default (see Container’s own doc), so bytes stored here are never advertised or served in that mode. The method stays callable regardless of container so a caller sharing one code path across both (e.g. a segmenter that always calls set_init once available) does not need to branch on which container it configured.

Examples found in repository

examples/client_stepping.rs (line 47)

fn canned_playlist() -> String {
    let trunk = Trunk::new(TrunkConfig::new(nz(16), nz(4), nz(8), nz(4), nz(16)));
    let writer = trunk
        .segment_writer()
        .expect("first (and only) segment writer");
    let origin = HlsOrigin::builder(std::sync::Arc::clone(&trunk))
        .target_duration_secs(1.0)
        .window_segments(nz(4))
        .low_latency(500)
        .build()
        .expect("both required fields set");
    origin.set_init(vec![0xAA; 32]);

    writer.publish_part(PartEntry::new(
        vec![0x01; 16],
        1,
        0,
        Duration::from_millis(500),
        true,
    ));
    writer.publish_segment(SegmentEntry::new(
        vec![0x02; 32],
        1,
        Duration::from_secs(1),
        Timestamp::from_nanos(0),
        SegmentMeta {
            discontinuous: false,
        },
    ));
    writer.publish_part(PartEntry::new(
        vec![0x03; 16],
        2,
        0,
        Duration::from_millis(500),
        true,
    ));

    match origin.resolve(
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
        Timestamp::from_nanos(0),
        AwaitPolicy::new(Timestamp::from_nanos(0)),
    ) {
        EgressResponse::Ready {
            body: HlsBody::Playlist(m),
            ..
        } => m,
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }
}

examples/origin_playlist.rs (line 63)

fn main() {
    let trunk = Trunk::new(TrunkConfig::new(nz(16), nz(4), nz(8), nz(4), nz(16)));
    let writer = trunk
        .segment_writer()
        .expect("first (and only) segment writer");
    let origin = HlsOrigin::builder(std::sync::Arc::clone(&trunk))
        .target_duration_secs(TARGET_DURATION_SECS)
        .window_segments(nz(WINDOW_SEGMENTS))
        .low_latency(PART_TARGET_MS)
        .build()
        .expect("both required fields set");
    origin.set_init(vec![0xAA; 32]);

    // Segment 1 closes with two parts.
    writer.publish_part(PartEntry::new(
        vec![0x01; 16],
        1,
        0,
        Duration::from_millis(500),
        true,
    ));
    writer.publish_part(PartEntry::new(
        vec![0x02; 16],
        1,
        1,
        Duration::from_millis(500),
        false,
    ));
    writer.publish_segment(SegmentEntry::new(
        vec![0x03; 32],
        1,
        Duration::from_secs(1),
        Timestamp::from_nanos(0),
        SegmentMeta {
            discontinuous: false,
        },
    ));

    // Segment 2 is still open, with only its first part landed so far.
    writer.publish_part(PartEntry::new(
        vec![0x04; 16],
        2,
        0,
        Duration::from_millis(500),
        true,
    ));

    println!("--- master.m3u8 ---");
    println!("{}", master_playlist_m3u8("media.m3u8"));

    println!("--- media.m3u8 ---");
    match resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    ) {
        EgressResponse::Ready {
            body: HlsBody::Playlist(m),
            ..
        } => println!("{m}"),
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }

    // A plain (non-blocking) request is Ready immediately.
    let outcome = resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    );
    assert!(matches!(
        outcome,
        EgressResponse::Ready {
            body: HlsBody::Playlist(_),
            ..
        }
    ));
    println!("resolve(Playlist, no query)     -> Ready");

    // A blocking-reload request for a segment that hasn't closed yet: with
    // `await_policy`'s deadline already at `now`, this immediately reports
    // the awaited condition has run out of patience rather than serving a
    // fabricated Ready.
    let outcome = resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            // Segment 2 is the in-progress one, so msn=3 is the next segment
            // that has not closed. Keep this within
            // `ABUSE_MSN_FUTURE_BOUND` (live edge + 2) — a larger value is
            // rejected outright as abuse and never reaches the Await path
            // this block is demonstrating.
            query: BlockingQuery {
                hls_msn: Some(3),
                hls_part: None,
            },
        },
    );
    assert_eq!(outcome, EgressResponse::NotFound);
    println!("resolve(Playlist, _HLS_msn=3)   -> NotFound (Await's patience already expired)");

    // A `_HLS_msn` unreasonably far beyond the live edge is rejected outright
    // (RFC 8216bis §6.2.5.2 abuse prevention) rather than ever Await-ing.
    let outcome = resolve(
        &origin,
        HlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery {
                hls_msn: Some(999),
                hls_part: None,
            },
        },
    );
    assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
    println!("resolve(Playlist, _HLS_msn=999) -> BadRequest (abuse bound)");

    // `Resource`: the init segment and the closed segment are Ready...
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "init-1.mp4".to_string(),
        },
    ) {
        EgressResponse::Ready { .. } => println!("resolve(Resource, init-1.mp4)     -> Ready"),
        other => panic!("expected Ready, got {other:?}"),
    }
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "seg-1-1.m4s".to_string(),
        },
    ) {
        EgressResponse::Ready { .. } => println!("resolve(Resource, seg-1-1.m4s)    -> Ready"),
        other => panic!("expected Ready, got {other:?}"),
    }
    // ...a live part of the still-open segment is Ready too...
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "part-1-2.0.m4s".to_string(),
        },
    ) {
        EgressResponse::Ready { .. } => println!("resolve(Resource, part-1-2.0.m4s) -> Ready"),
        other => panic!("expected Ready, got {other:?}"),
    }
    // ...a preload-hinted part not yet produced reports NotFound once this
    // call's `await_policy` has already expired (a real HTTP adapter would
    // instead give it a real deadline and block on `Trunk::listen()`)...
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "part-1-2.1.m4s".to_string(),
        },
    ) {
        EgressResponse::NotFound => {
            println!(
                "resolve(Resource, part-1-2.1.m4s) -> NotFound (Await's patience already expired)"
            )
        }
        other => panic!("expected NotFound, got {other:?}"),
    }
    // ...and an unrecognised filename is a plain 404.
    match resolve(
        &origin,
        HlsRequest::Resource {
            name: "nope.txt".to_string(),
        },
    ) {
        EgressResponse::NotFound => println!("resolve(Resource, nope.txt)       -> NotFound"),
        other => panic!("expected NotFound, got {other:?}"),
    }
}

pub fn init_bytes(&self) -> Option<Bytes>

The fMP4 init segment bytes, if set.

pub fn closed_segments(&self) -> Vec<ClosedSegment>

A snapshot of this origin’s currently-advertised closed segments (drains the cursor first, same as render_playlist) — ascending by sequence number.

Exists for a caller that needs to merge this origin’s live window with a different source of segments over the same numbering (multimux’s DVR archive, issue #900: catch-up serving must present one continuous playlist spanning the archive and the still-live tail that hasn’t been archived yet). Reuses the one cursor drain already maintains rather than making the caller open a second cursor on the same Trunk just to learn the same window render_playlist itself renders — media_plane’s own module doc: writer cost is O(N) in cursor count, so a cursor is per distinct consumer, never per peer, and never duplicated for data another cursor already tracks.

Trait Implementations

impl ServedEgress for HlsOrigin

type Request = HlsRequest

The protocol-specific request shape (e.g. LL-HLS’s Media Sequence Number + part index; a DASH segment/manifest request; a catch-up time range) — owned by the implementing crate, not this one.

type Body = HlsBody

The protocol-specific resolved body (a playlist string, manifest XML, segment bytes, …).

fn resolve( &self, request: HlsRequest, now: Timestamp, await_policy: AwaitPolicy, ) -> EgressResponse<HlsBody>

Resolve request against this implementation’s current state at time now, bounded by await_policy — see EgressResponse::Await’s contract.