media-pp 0.1.3

A small, GStreamer-flavored media pipeline library built on FFmpeg.
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

use std::{
    fs,
    time::{SystemTime, UNIX_EPOCH},
};

use media_pp::{
    buffer::MediaBuffer,
    elements::{AppSink, AppSource},
    log::{self, Level},
    pipeline::Pipeline,
};

#[test]
fn pipeline_logs_topology_eos_and_control_at_each_boundary() {
    let unique = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system clock must be after the Unix epoch")
        .as_nanos();
    let log_dir = std::env::temp_dir().join(format!(
        "media_pp_flow_log_{}_{}",
        std::process::id(),
        unique
    ));
    let log_path = log_dir.to_string_lossy();
    let guard =
        log::init("flow", &log_path, Level::Trace, 1).expect("private file logger must initialize");

    let (source, handle) = AppSource::new("source", 4);
    let pipeline = Pipeline::new("flow-test", source, |source, context| {
        let branch = context
            .branch()
            .queue("queue", 4)
            .to(Box::new(AppSink::new("sink", |_buf| Ok(()))))?;
        context.attach(source, 0, branch)?;
        Ok(())
    })
    .expect("test pipeline must build");

    pipeline.run();
    pipeline.pause();
    pipeline.resume();
    handle
        .push(MediaBuffer::Eos)
        .expect("source must accept EOS");
    drop(handle);
    let _events: Vec<_> = pipeline.bus().iter().collect();

    assert_eq!(guard.dropped_lines(), 0);
    drop(guard);

    let log_file = fs::read_dir(&log_dir)
        .expect("temporary log directory must remain readable")
        .map(|entry| entry.expect("log directory entry must be readable").path())
        .find(|path| {
            path.file_name()
                .is_some_and(|name| name.to_string_lossy().starts_with("flow."))
        })
        .expect("the rolling appender must create a log file");
    let contents = fs::read_to_string(log_file).expect("log file must contain UTF-8 text");

    // One record, not two: the `Queue` worker started by this very `run()`
    // is already free to log, so a separate diagram record would only
    // usually follow the `run` that produced it.
    assert!(contents.contains(concat!(
        "[element=Pipeline] [name=flow-test] run\n",
        "AppSource(source)#1\n",
        "└── [source_src] → Queue(queue)#2\n",
        "                   └── [queue_src] → AppSink(sink)#3",
    )));
    assert!(
        contents.contains(
            "[element=AppSource] [name=source] event=control control=Pause phase=received"
        )
    );
    assert!(
        contents
            .contains("[element=Queue] [name=queue] event=control control=Pause phase=forwarding")
    );
    assert!(contents.contains(
        "[element=AppSink] [name=sink] event=control control=Pause phase=completed outcome=ok"
    ));
    assert!(contents.contains(
        "[element=AppSource] [name=source] event=eos phase=sent pad=source_src outcome=ok"
    ));
    assert!(contents.contains("[element=Queue] [name=queue] event=eos phase=completed outcome=ok"));
    assert!(contents.contains("[element=AppSink] [name=sink] event=eos phase=received"));

    // EOS crosses a `Queue`, so the record the source emits sending it and
    // the record the sink emits receiving it must name different threads —
    // that difference is the only reason the field exists.
    let sending_thread = thread_of(
        &contents,
        "[element=AppSource] [name=source] event=eos phase=sending",
    );
    let receiving_thread = thread_of(
        &contents,
        "[element=AppSink] [name=sink] event=eos phase=received",
    );
    assert!(sending_thread.starts_with("pipeline:source#"));
    assert!(receiving_thread.starts_with("queue:queue#"));

    fs::remove_dir_all(log_dir).expect("temporary log directory must be removable");
}

/// Reads the `[thread=…]` value off the one record containing `marker`.
fn thread_of(contents: &str, marker: &str) -> String {
    let line = contents
        .lines()
        .find(|line| line.contains(marker))
        .unwrap_or_else(|| panic!("no record matched `{marker}`"));
    let (_, after) = line
        .split_once("[thread=")
        .expect("every record carries a thread field");
    let (tag, _) = after
        .split_once(']')
        .expect("the thread field must be closed");
    tag.to_owned()
}