Struct LlHlsOrigin 

pub struct LlHlsOrigin { /* 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 LlHlsOrigin

pub fn new( trunk: Arc<Trunk>, target_duration_secs: f64, part_target_ms: u32, window_segments: NonZeroUsize, ) -> Self

Build a fresh origin over trunk, subscribing its one SegmentCursor immediately (so the window starts empty but never misses a segment published from this point on).

window_segments bounds how many closed segments this origin advertises in a rendered Media Playlist — independent of media_plane::trunk::TrunkConfig::segment_capacity (the Trunk’s own retention bound): a caller may legitimately want a shorter advertised window than the Trunk retains for other consumers (e.g. a DVR SegmentEgress reading the same Trunk).

Examples found in repository

examples/client_stepping.rs (line 43)

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 = LlHlsOrigin::new(std::sync::Arc::clone(&trunk), 1.0, 500, nz(4));
    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(
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
        Timestamp::from_nanos(0),
        AwaitPolicy::new(Timestamp::from_nanos(0)),
    ) {
        EgressResponse::Ready {
            body: LlHlsBody::Playlist(m),
            ..
        } => m,
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }
}

examples/origin_playlist.rs (lines 57-62)

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 = LlHlsOrigin::new(
        std::sync::Arc::clone(&trunk),
        TARGET_DURATION_SECS,
        PART_TARGET_MS,
        nz(WINDOW_SEGMENTS),
    );
    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,
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    ) {
        EgressResponse::Ready {
            body: LlHlsBody::Playlist(m),
            ..
        } => println!("{m}"),
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }

    // A plain (non-blocking) request is Ready immediately.
    let outcome = resolve(
        &origin,
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    );
    assert!(matches!(
        outcome,
        EgressResponse::Ready {
            body: LlHlsBody::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,
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery {
                hls_msn: Some(5),
                hls_part: None,
            },
        },
    );
    assert_eq!(outcome, EgressResponse::NotFound);
    println!("resolve(Playlist, _HLS_msn=5)   -> 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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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 fMP4 init segment bytes — see this module’s doc for why an init segment is not something any Trunk ring holds.

Examples found in repository

examples/client_stepping.rs (line 44)

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 = LlHlsOrigin::new(std::sync::Arc::clone(&trunk), 1.0, 500, nz(4));
    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(
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
        Timestamp::from_nanos(0),
        AwaitPolicy::new(Timestamp::from_nanos(0)),
    ) {
        EgressResponse::Ready {
            body: LlHlsBody::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 = LlHlsOrigin::new(
        std::sync::Arc::clone(&trunk),
        TARGET_DURATION_SECS,
        PART_TARGET_MS,
        nz(WINDOW_SEGMENTS),
    );
    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,
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    ) {
        EgressResponse::Ready {
            body: LlHlsBody::Playlist(m),
            ..
        } => println!("{m}"),
        other => panic!("expected Ready(Playlist), got {other:?}"),
    }

    // A plain (non-blocking) request is Ready immediately.
    let outcome = resolve(
        &origin,
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery::default(),
        },
    );
    assert!(matches!(
        outcome,
        EgressResponse::Ready {
            body: LlHlsBody::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,
        LlHlsRequest::Playlist {
            track_id: DEFAULT_TRACK_ID,
            query: BlockingQuery {
                hls_msn: Some(5),
                hls_part: None,
            },
        },
    );
    assert_eq!(outcome, EgressResponse::NotFound);
    println!("resolve(Playlist, _HLS_msn=5)   -> 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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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,
        LlHlsRequest::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.

Trait Implementations

impl ServedEgress for LlHlsOrigin

type Request = LlHlsRequest

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 = LlHlsBody

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

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

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