1
use serde::Deserialize;
2
use std::{
3
    io::{self, BufRead, Write},
4
    sync::mpsc,
5
    thread,
6
    time::{Duration, Instant},
7
};
8
use str0m::crypto::dtls::DtlsCert;
9
use str0m_probe::driver::{Driver, Result};
10
 
11
#[derive(Deserialize)]
12
#[serde(tag = "command", rename_all = "snake_case")]
13
enum Command {
14
    Answer { sdp: String },
15
    Channels { robot: u16, spectrum: u16 },
16
    Run { frames: u32, drop_every: u32 },
17
    Warmup { frames: u32 },
18
    Stats,
19
    Stop,
20
}
21
 
22
fn emit(value: &impl serde::Serialize) -> Result<()> {
23
    let mut out = io::stdout().lock();
24
    serde_json::to_writer(&mut out, value)?;
25
    writeln!(out)?;
26
    out.flush()?;
27
    Ok(())
28
}
29
 
30
fn run() -> Result<()> {
31
    let args: Vec<String> = std::env::args().collect();
32
    if args.len() != 4 {
33
        return Err("expected IP, certificate and key paths".into());
34
    }
35
    let cert = DtlsCert {
36
        certificate: std::fs::read(&args[2])?,
37
        private_key: std::fs::read(&args[3])?,
38
    };
39
    let (mut driver, offer) = Driver::new(&args[1], cert)?;
40
    // The parent process consumes this private signaling IPC; it is not a log.
41
    emit(&serde_json::json!({"event": "offer", "sdp": offer, "mid": driver.mid()}))?;
42
    let (sender, commands) = mpsc::sync_channel(8);
43
    thread::spawn(move || {
44
        for line in io::stdin().lock().lines() {
45
            let Ok(line) = line else {
46
                break;
47
            };
48
            let Ok(command) = serde_json::from_str::<Command>(&line) else {
49
                break;
50
            };
51
            if sender.send(command).is_err() {
52
                break;
53
            }
54
        }
55
    });
56
    let started = Instant::now();
57
    let mut workload: Option<(Instant, u32, u32, bool)> = None;
58
    while started.elapsed() < Duration::from_secs(60) {
59
        if let Ok(command) = commands.try_recv() {
60
            match command {
61
                Command::Answer { sdp } => driver.answer(&sdp)?,
62
                Command::Channels { robot, spectrum } => {
63
                    driver.add_channel("robot", robot, true)?;
64
                    driver.add_channel("spectrum", spectrum, false)?;
65
                }
66
                Command::Run { frames, drop_every } => {
67
                    driver.drop_every = drop_every;
68
                    workload = Some((Instant::now(), 0, frames, true));
69
                }
70
                Command::Warmup { frames } => workload = Some((Instant::now(), 0, frames, false)),
71
                Command::Stats => {
72
                    emit(&serde_json::json!({"event": "stats", "stats": driver.stats}))?
73
                }
74
                Command::Stop => {
75
                    emit(&serde_json::json!({"event": "stopped", "stats": driver.stats}))?;
76
                    return Ok(());
77
                }
78
            }
79
        }
80
        driver.pump()?;
81
        for event in driver.events.drain(..) {
82
            emit(&event)?;
83
        }
84
        if let Some((begin, frame, count, data)) = &mut workload {
85
            if begin.elapsed() >= Duration::from_millis(u64::from(*frame) * 20) {
86
                driver.audio(*frame)?;
87
                if *data && *frame % 5 == 0 {
88
                    driver.data("robot", *frame / 5)?;
89
                }
90
                if *data && *frame % 2 == 0 {
91
                    driver.data("spectrum", *frame / 2)?;
92
                }
93
                *frame += 1;
94
                if *frame == *count {
95
                    emit(
96
                        &serde_json::json!({"event": if *data {"run_done"} else {"warmup_done"}, "stats": driver.stats}),
97
                    )?;
98
                    workload = None;
99
                }
100
            }
101
        }
102
        thread::sleep(Duration::from_millis(1));
103
    }
104
    Err("probe deadline exceeded".into())
105
}
106
 
107
fn main() {
108
    if run().is_err() {
109
        // Error payloads can contain SDP. Keep stdout/stderr free of diagnostics.
110
        let _ = emit(&serde_json::json!({"event": "failed"}));
111
        std::process::exit(1);
112
    }
113
}