multimux 0.10.0

Multi-input (RTSP/RTP/TS-UDP/TS-HTTP/SRT/HLS-pull/DASH-pull/Smooth-pull/RTMP/file), multi-output (LL-HLS/DASH/LL-DASH/Smooth/TS-HLS/catch-up + SRT/RTMP/RTSP push) just-in-time repackaging HTTP origin (library: tokio + axum), with shared output auth and an external scheme plugin registry.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340

//! Deterministic end-to-end integration gate for the LL-HLS origin (#663): a
//! real [`LlHlsSegmenter`] feeding [`RouteHandle`] directly -> the axum
//! [`router`], driven with `tower::ServiceExt::oneshot` (no real TCP socket,
//! no timing-dependent assertions) so the whole segmenter-to-HTTP path is
//! exercised without flakiness.
//!
//! Test 1 proves the served bytes are actually valid fMP4/CMAF (via
//! `transmux::validate`), not just "some bytes came back". Test 2 proves the
//! blocking-reload path (RFC 8216bis §6.2.5.2) really wakes on new data,
//! rather than serving a stale playlist or hanging to its timeout.
//!
//! Issue #805 task 5 deleted `crate::pipeline::run_pipeline`/`MockSource`
//! (the old `SourceConnector`-fed path's test scaffolding); test 1 below
//! drives the same real `LlHlsSegmenter` that module used to wrap, directly —
//! mirroring `tests/lldash_dashjs.rs`'s own `run_live_producer` — since
//! `RouteHandle`'s owned `Trunk` is a test-only placeholder now (see
//! `RouteHandle`'s own doc), and this test exists precisely to drive it
//! directly and publish it.

use std::collections::HashMap;
use std::sync::Arc;

use axum::body::Body;
use axum::http::{Request, StatusCode};
use tower::ServiceExt;

use multimux::origin::{AppState, router};
use multimux::output::Output;
use multimux::output::llhls::LlHlsOutput;
use multimux::route::RouteHandle;
use transmux::avc_config_from_sprop;
use transmux::ll_hls::{LlHlsSegmenter, PartInfo};
use transmux::pipeline::{CodecConfig, Sample, TrackSpec};
use transmux::validate::{Severity, validate_init_segment, validate_media_segment};

/// A real-ish sprop-parameter-sets pair (SPS+PPS) — same one used by
/// `multimux::source::rtsp`'s own tests — decoded into an `avcC` config so
/// the init segment carries a genuine AVC configuration record rather than a
/// fabricated one.
const SPROP: &str = "Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==";

/// 90 kHz video timescale (the CMAF movie timescale [`feed_via_segmenter`]
/// builds with) — 1/30 s per access unit at 30 fps.
const VIDEO_TIMESCALE: u32 = 90_000;
const FRAME_DUR: u32 = VIDEO_TIMESCALE / 30;
/// Target full-segment duration, seconds — matches the pre-#805-task-5
/// `run_pipeline`-driven version of this test.
const TARGET_DURATION_SECS: f64 = 1.0;
/// LL-HLS part target, milliseconds — matches the pre-#805-task-5 version.
const PART_TARGET_MS: u32 = 500;

fn video_track_spec() -> TrackSpec {
    let config = avc_config_from_sprop(SPROP).expect("valid sprop");
    TrackSpec::new(
        1,
        VIDEO_TIMESCALE,
        CodecConfig::Avc {
            config,
            width: 0,
            height: 0,
        },
    )
}

/// Only `Severity::Error` issues fail the gate — warnings (e.g. a missing
/// `mvex`/`trex` on a segment that isn't the init segment) are informational.
fn errors_only(
    issues: &[transmux::validate::ConformanceIssue],
) -> Vec<&transmux::validate::ConformanceIssue> {
    issues
        .iter()
        .filter(|i| i.severity == Severity::Error)
        .collect()
}

/// Extract the first `seg-{track}-{seq}.m4s` URI out of a rendered media
/// playlist body.
fn first_segment_uri(playlist: &str) -> Option<&str> {
    let start = playlist.find("seg-")?;
    let rest = &playlist[start..];
    let end = rest.find(".m4s")? + ".m4s".len();
    Some(&rest[..end])
}

async fn body_string(resp: axum::response::Response) -> String {
    let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
        .await
        .unwrap();
    String::from_utf8(bytes.to_vec()).unwrap()
}

async fn body_bytes(resp: axum::response::Response) -> Vec<u8> {
    axum::body::to_bytes(resp.into_body(), usize::MAX)
        .await
        .unwrap()
        .to_vec()
}

fn get(uri: &str) -> Request<Body> {
    Request::builder()
        .uri(uri)
        .body(Body::empty())
        .expect("well-formed GET request")
}

/// Drives `batches` through a real [`LlHlsSegmenter`] into `store` — the
/// direct replacement for the deleted `crate::pipeline::run_pipeline`
/// (issue #805 task 5): same segmenter, same track specs, same
/// `set_init`/`add_part`/`add_segment` calls, just inlined here rather than
/// wrapped behind a `SampleSource`/`MockSource` indirection this crate no
/// longer has. Flushes any trailing buffered partial segment at the end,
/// exactly like the deleted `run_pipeline` did. Publishes `SPTS_PROGRAM_ID`
/// first (issue #805 task 6: a `ProgramServing` bundle must exist before
/// `set_init`/`add_part`/`add_segment` can write into it).
///
/// MUTATION VERIFIED: commenting out the `store.set_init(...)` call below
/// makes `end_to_end_pipeline_serves_valid_llhls` fail:
/// `assert_eq!(resp.status(), StatusCode::OK)` (the init-segment `GET`) fails
/// at `multimux/tests/origin_llhls.rs:209`, comparing actual `404` against
/// expected `200` — with no init bytes ever set, the route's `init-1.mp4`
/// route has nothing to serve. Rebuilt and re-ran to confirm this exact
/// failure, then reverted.
fn feed_via_segmenter(
    store: &RouteHandle,
    specs: Vec<TrackSpec>,
    batches: Vec<Vec<(u32, Sample)>>,
) {
    let program = media_plane::ProgramId(0);
    store.publish_new_program(program);

    let mut seg = LlHlsSegmenter::with_part_target(
        specs,
        transmux::VIDEO_CLOCK_RATE,
        TARGET_DURATION_SECS,
        PART_TARGET_MS,
    )
    .expect("segmenter builds");
    store.set_init(program, seg.init_segment().expect("init segment builds"));

    for batch in batches {
        for (track_id, sample) in batch {
            seg.push(track_id, sample).expect("push succeeds");
        }
        for part in seg.take_ready_parts() {
            store.add_part(program, part);
        }
        for segment in seg.take_ready_segments() {
            store.add_segment(program, segment);
        }
    }

    seg.flush().expect("flush succeeds");
    for part in seg.take_ready_parts() {
        store.add_part(program, part);
    }
    for segment in seg.take_ready_segments() {
        store.add_segment(program, segment);
    }
}

#[tokio::test]
async fn end_to_end_pipeline_serves_valid_llhls() {
    let store = Arc::new(RouteHandle::new(TARGET_DURATION_SECS, PART_TARGET_MS, 8));
    let specs = vec![video_track_spec()];

    // 120 frames @ 30 fps = 4 s of video, with sync samples every 30 frames
    // (1 s, == the 1.0 s target duration) so at least 3 segments close via
    // the normal boundary path (not just the EOS flush tail), each carrying
    // several 500 ms parts (15 frames/part at 30 fps/90kHz).
    let mut batches = Vec::new();
    for i in 0..120u32 {
        let is_sync = i % 30 == 0;
        let data = vec![0xAAu8.wrapping_add((i % 251) as u8); 64];
        let sample = Sample::new(
            data,
            Some(i64::from(i) * i64::from(FRAME_DUR)),
            Some(i64::from(i) * i64::from(FRAME_DUR)),
            Some(FRAME_DUR),
            is_sync,
        );
        batches.push(vec![(1u32, sample)]);
    }

    // `feed_via_segmenter` publishes `SPTS_PROGRAM_ID` itself before writing
    // (issue #805 task 6) — no separate publish call needed here.
    feed_via_segmenter(&store, specs, batches);

    let mut streams = HashMap::new();
    streams.insert(
        "cam".to_string(),
        (
            store.clone(),
            vec![Arc::new(LlHlsOutput::default()) as Arc<dyn Output>],
        ),
    );
    let app = router(Arc::new(AppState::new(streams)));

    // 1. Media playlist: LL-HLS tags present.
    let resp = app
        .clone()
        .oneshot(get("/cam/media.m3u8"))
        .await
        .expect("router call");
    assert_eq!(resp.status(), StatusCode::OK);
    let playlist = body_string(resp).await;
    assert!(playlist.contains("#EXT-X-PART"), "playlist: {playlist}");
    assert!(playlist.contains("#EXT-X-PART-INF"), "playlist: {playlist}");
    assert!(
        playlist.contains("#EXT-X-SERVER-CONTROL"),
        "playlist: {playlist}"
    );

    // 2. Init segment: 200 + structurally conformant fMP4 init segment.
    let resp = app
        .clone()
        .oneshot(get("/cam/init-1.mp4"))
        .await
        .expect("router call");
    assert_eq!(resp.status(), StatusCode::OK);
    let init_bytes = body_bytes(resp).await;
    assert!(!init_bytes.is_empty(), "init segment body non-empty");
    let init_issues = validate_init_segment(&init_bytes);
    let init_errors = errors_only(&init_issues);
    assert!(
        init_errors.is_empty(),
        "init segment must have no conformance errors, got: {init_errors:?}"
    );

    // 3. A real media segment URI from the playlist: 200 + conformant media
    // segment.
    let seg_uri = first_segment_uri(&playlist)
        .unwrap_or_else(|| panic!("no seg-*.m4s URI in playlist: {playlist}"));
    let resp = app
        .clone()
        .oneshot(get(&format!("/cam/{seg_uri}")))
        .await
        .expect("router call");
    assert_eq!(resp.status(), StatusCode::OK, "segment URI: {seg_uri}");
    let seg_bytes = body_bytes(resp).await;
    assert!(!seg_bytes.is_empty(), "segment body non-empty");
    let seg_issues = validate_media_segment(&seg_bytes);
    let seg_errors = errors_only(&seg_issues);
    assert!(
        seg_errors.is_empty(),
        "media segment must have no conformance errors, got: {seg_errors:?}"
    );

    // 4. Unknown stream -> 404.
    let resp = app
        .oneshot(get("/ghost/media.m3u8"))
        .await
        .expect("router call");
    assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}

fn part(seq: u32, idx: u32) -> PartInfo {
    PartInfo {
        bytes: vec![0x10 + idx as u8; 4],
        duration: 0.5,
        independent: idx == 0,
        segment_seq: seq,
        part_index: idx,
    }
}

/// Bites the blocking-reload wakeup path (`output::llhls`'s internal `wait_for_progress`):
/// without the `watch`-based wakeup + re-render, a `_HLS_msn`/`_HLS_part`
/// request for a not-yet-available part would either serve the stale
/// playlist (missing the new part) or hang to `BLOCKING_RELOAD_TIMEOUT`
/// (5 s) rather than resolving as soon as the part lands.
#[tokio::test]
async fn blocking_reload_resolves_when_part_arrives() {
    let store = Arc::new(RouteHandle::new(4.0, 500, 8));
    let program = media_plane::ProgramId(0);
    store.publish_new_program(program);
    store.set_init(program, vec![0xAA; 8]);
    store.add_part(program, part(1, 0));

    let mut streams = HashMap::new();
    streams.insert(
        "cam".to_string(),
        (
            store.clone(),
            vec![Arc::new(LlHlsOutput::default()) as Arc<dyn Output>],
        ),
    );
    let app = router(Arc::new(AppState::new(streams)));

    // latest_progress() is currently (1, 1): in-progress segment 1 has one
    // part (index 0). Asking for msn=1/part=1 is NOT yet satisfied
    // (part_count(1) > part(1) is false), so this request must block.
    let app_for_task = app.clone();
    let handle = tokio::spawn(async move {
        app_for_task
            .oneshot(get("/cam/media.m3u8?_HLS_msn=1&_HLS_part=1"))
            .await
            .expect("router call")
    });

    // Let the spawned task actually reach the `watch` await point before we
    // publish the part it's waiting for; the correctness of the assertion
    // below rests on the spawn+add_part ordering and the handler's own
    // `changed().await`, not on this yield being "long enough".
    for _ in 0..8 {
        tokio::task::yield_now().await;
    }

    store.add_part(program, part(1, 1));

    // NOT a pure hang guard (issue #807): this must stay meaningfully below
    // the 5s internal `BLOCKING_RELOAD_TIMEOUT` fallback it distinguishes
    // from, or a broken (never-wakes) watch would still pass by falling
    // through to that same fallback within a too-generous window. Bound the
    // wait in *real* time (no `tokio::time::pause()`): a working watch wakeup
    // resolves within milliseconds of `add_part`, whereas a broken wakeup
    // only resolves at the 5s fallback. 500 ms comfortably separates the two
    // (10x below the 5s cap) without being flaky on a loaded CI box.
    let resp = tokio::time::timeout(std::time::Duration::from_millis(500), handle)
        .await
        .expect("blocking reload must resolve promptly on the watch wakeup, not the 5s timeout fallback")
        .expect("blocking request task did not panic");
    assert_eq!(
        resp.status(),
        StatusCode::OK,
        "blocking reload must resolve (not timeout/404) once the awaited part lands"
    );
    let playlist = body_string(resp).await;
    // Assert the *real* `#EXT-X-PART` line for the arrived part, not merely
    // the bare URI — that URI also appears in the always-emitted
    // `#EXT-X-PRELOAD-HINT:TYPE=PART,URI="part-1-1.1.m4s"` line for the
    // next-expected part, which is present even before `add_part(1, 1)` runs
    // (see `multimux::output::llhls::media_playlist_m3u8`). Only a genuine PART line
    // proves the new part was actually rendered.
    let real_part_line = "#EXT-X-PART:DURATION=0.5,URI=\"part-1-1.1.m4s\"";
    assert!(
        playlist.contains(real_part_line),
        "resolved playlist must include the real #EXT-X-PART line for the \
         newly-arrived part (not just the preload-hint URI): {playlist}"
    );
}