use std::cell::RefCell;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::rc::Rc;
use std::sync::mpsc::{Receiver, Sender, channel};
use std::time::{Duration, Instant};
use anyhow::{Result, anyhow};
use rquickjs::{
ArrayBuffer, CatchResultExt, Context, Ctx, Function, Object, Runtime, Value,
context::EvalOptions, function::IntoArgs,
};
use serde_json::{Value as J, json};
use crate::EngineConfig;
use crate::audio::{AudioSink, PulseSink};
use crate::buffer::{MediaSourceNative, SOURCE_BUFFER_QUOTA_BYTES, SourceBufferNative};
use crate::cdm::{CdmBackend, CdmEvent, Widevine};
use crate::eme::{audio_type_supported, can_play_type, eme_check};
use crate::media::{MediaEngine, SinkFactory};
use crate::mp4;
use crate::net::{self, Completion, HttpRequest};
const PLATFORM: &str = include_str!("../js/platform.js");
pub type CdmFactory = Box<dyn Fn(bool) -> Result<Box<dyn CdmBackend>>>;
struct Native {
start: Instant,
next_id: u32,
mses: HashMap<u32, MediaSourceNative>,
sbs: HashMap<u32, SourceBufferNative>,
medias: HashMap<u32, MediaEngine>,
cdms: HashMap<u32, Box<dyn CdmBackend>>,
cdm_factory: CdmFactory,
sink_factory: SinkFactory,
cdm_error: Option<String>,
net_tx: Sender<Completion>,
aborted: HashSet<u32>,
op_counts: BTreeMap<&'static str, u64>,
eme_requests: Vec<J>,
script_log: Vec<J>,
http_count: u64,
http_bytes: u64,
redact: Vec<String>,
truncate_scripts: bool,
retired_output: OutputCounts,
in_flight: usize,
}
impl Native {
fn id(&mut self) -> u32 {
self.next_id += 1;
self.next_id
}
fn count(&mut self, op: &'static str) {
*self.op_counts.entry(op).or_default() += 1;
}
fn primary_media(&self) -> Option<&MediaEngine> {
self.medias
.iter()
.filter(|(_, m)| m.keys_id.is_some())
.min_by_key(|(id, _)| **id)
.map(|(_, m)| m)
}
fn position_for_buffer(&self, sb_id: u32) -> i64 {
self.medias
.values()
.find(|m| {
m.ms_id
.and_then(|id| self.mses.get(&id))
.is_some_and(|ms| ms.sbs.contains(&sb_id))
})
.map(|m| m.current_time_us())
.unwrap_or(0)
}
}
type N = Rc<RefCell<Native>>;
pub struct Engine {
ctx: Context,
rt: Runtime,
n: N,
net_rx: Receiver<Completion>,
last_gc: Instant,
}
const GC_INTERVAL: Duration = Duration::from_secs(2);
const ACTIVE_WAIT_MS: f64 = 5.0;
const IDLE_WAIT: Duration = Duration::from_secs(60);
fn opts(name: &str) -> EvalOptions {
let mut o = EvalOptions::default();
o.strict = false;
o.global = true;
o.filename = Some(name.into());
o
}
fn bytes_of(v: &Value) -> Option<Vec<u8>> {
let ab = ArrayBuffer::from_value(v.clone())?;
unsafe { ab.as_bytes() }.map(|b| b.to_vec())
}
fn media_state(m: &MediaEngine) -> String {
json!({"currentTime": m.current_time(), "readyState": m.ready_state, "paused": m.paused, "seeking": m.seeking, "ended": m.ended}).to_string()
}
fn install_ops<'js>(ctx: &Ctx<'js>, n: &N) -> rquickjs::Result<()> {
let g = ctx.globals();
macro_rules! op {
($name:literal, $f:expr) => {
g.set($name, Function::new(ctx.clone(), $f)?)?;
};
}
macro_rules! with_media {
($name:literal, $count:literal, |$m:ident $(, $arg:ident : $ty:ty)*| $body:expr) => {
let s = n.clone();
op!($name, move |id: u32 $(, $arg: $ty)*| {
let mut st = s.borrow_mut();
st.count($count);
if let Some($m) = st.medias.get_mut(&id) {
$body;
}
});
};
}
macro_rules! with_cdm {
($name:literal, $count:literal, |$c:ident, $p:ident $(, $arg:ident : $ty:ty)*| $body:expr) => {
let s = n.clone();
op!($name, move |keys_id: u32, $p: u32 $(, $arg: $ty)*| {
let mut st = s.borrow_mut();
st.count($count);
if let Some($c) = st.cdms.get_mut(&keys_id) {
$body;
}
});
};
}
let s = n.clone();
op!("__op_now", move || -> f64 {
s.borrow().start.elapsed().as_secs_f64() * 1000.0
});
let s = n.clone();
op!("__op_print", move |level: String, msg: String| {
eprintln!("JS[{level}] {}", crate::redact(&msg, &s.borrow().redact));
});
op!("__op_user_agent", crate::user_agent);
op!("__op_random", move |ctx: Ctx<'js>,
len: u32|
-> rquickjs::Result<
ArrayBuffer<'js>,
> {
let mut buf = vec![0u8; len as usize];
if let Err(e) = getrandom::getrandom(&mut buf) {
return Err(rquickjs::Exception::throw_internal(
&ctx,
&format!("crypto.getRandomValues: no system randomness: {e}"),
));
}
ArrayBuffer::new(ctx, buf)
});
let s = n.clone();
op!("__op_http", move |id: u32,
method: String,
url: String,
headers: String,
body: Value<'js>,
timeout: f64,
cache: String| {
let mut st = s.borrow_mut();
st.count("http");
st.http_count += 1;
st.in_flight += 1;
let headers: Vec<(String, String)> = serde_json::from_str(&headers).unwrap_or_default();
let req = HttpRequest {
id,
method,
url,
headers,
body: bytes_of(&body),
timeout_ms: if timeout > 0.0 {
timeout as u64
} else {
60_000
},
use_cache: cache == "default",
};
net::spawn(req, st.net_tx.clone());
});
let s = n.clone();
op!("__op_http_abort", move |id: u32| {
let mut st = s.borrow_mut();
st.count("http_abort");
st.aborted.insert(id);
});
let s = n.clone();
op!("__op_load_script", move |id: u32, url: String| {
let mut st = s.borrow_mut();
st.count("load_script");
st.in_flight += 1;
st.script_log
.push(json!({"id": id, "url": url, "phase": "fetching"}));
net::spawn_script(id, url, st.net_tx.clone());
});
let s = n.clone();
op!("__op_ms_create", move || -> u32 {
let mut st = s.borrow_mut();
st.count("ms_create");
let id = st.id();
st.mses.insert(
id,
MediaSourceNative {
duration: f64::NAN,
ended: false,
sbs: Vec::new(),
},
);
id
});
let s = n.clone();
op!("__op_sb_add", move |ms_id: u32, mime: String| -> i32 {
let mut st = s.borrow_mut();
st.count("sb_add");
if !audio_type_supported(&mime) {
return -1;
}
let id = st.id();
st.sbs
.insert(id, SourceBufferNative::new(mime, SOURCE_BUFFER_QUOTA_BYTES));
if let Some(ms) = st.mses.get_mut(&ms_id) {
ms.sbs.push(id);
}
id as i32
});
let s = n.clone();
op!("__op_sb_remove_buffer", move |ms_id: u32, sb_id: u32| {
let mut st = s.borrow_mut();
st.count("sb_remove_buffer");
st.sbs.remove(&sb_id);
if let Some(ms) = st.mses.get_mut(&ms_id) {
ms.sbs.retain(|x| *x != sb_id);
}
});
let s = n.clone();
op!("__op_sb_prepare_append", move |sb_id: u32,
len: u32|
-> bool {
let mut st = s.borrow_mut();
st.count("sb_prepare_append");
let position = st.position_for_buffer(sb_id);
st.sbs
.get_mut(&sb_id)
.is_some_and(|sb| sb.make_room(len as usize, position))
});
let s = n.clone();
op!("__op_sb_append", move |sb_id: u32,
data: Value<'js>,
ts_offset: f64|
-> String {
let mut st = s.borrow_mut();
st.count("sb_append");
let Some(bytes) = bytes_of(&data) else {
return json!({"error": "not an ArrayBuffer"}).to_string();
};
let Some(sb) = st.sbs.get_mut(&sb_id) else {
return json!({"error": "no such SourceBuffer"}).to_string();
};
match sb.append(&bytes, if ts_offset.is_finite() { ts_offset } else { 0.0 }) {
Ok(r) => json!({"init": r.init_segment, "frames": r.media_frames, "hasInitData": r.init_segment && sb.init_data.is_some(), "initDuration": sb.init.as_ref().and_then(|i| i.duration_secs), "end": sb.end_us().map(|e| e as f64 / 1e6)}).to_string(),
Err(e) => json!({"error": format!("{e:#}")}).to_string(),
}
});
let s = n.clone();
op!(
"__op_sb_set_append_window",
move |sb_id: u32, start: f64, end: f64| {
if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
let us = |t: f64| (t * 1e6).round() as i64;
sb.append_window_us = (us(start), us(end));
}
}
);
let s = n.clone();
op!("__op_sb_init_data", move |ctx: Ctx<'js>,
sb_id: u32|
-> rquickjs::Result<
Value<'js>,
> {
match s
.borrow()
.sbs
.get(&sb_id)
.and_then(|sb| sb.init_data.clone())
{
Some(b) => Ok(ArrayBuffer::new(ctx, b)?.into_value()),
None => Ok(Value::new_null(ctx)),
}
});
let s = n.clone();
op!("__op_sb_remove", move |sb_id: u32, start: f64, end: f64| {
let mut st = s.borrow_mut();
st.count("sb_remove");
if let Some(sb) = st.sbs.get_mut(&sb_id) {
sb.remove(start, end);
}
});
let s = n.clone();
op!("__op_sb_abort", move |sb_id: u32| {
if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
sb.abort();
}
});
let s = n.clone();
op!("__op_sb_change_type", move |sb_id: u32,
mime: String|
-> bool {
if !audio_type_supported(&mime) {
return false;
}
if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
sb.mime = mime;
}
true
});
let s = n.clone();
op!("__op_sb_buffered", move |sb_id: u32| -> String {
let r: Vec<[f64; 2]> = s
.borrow()
.sbs
.get(&sb_id)
.map(|sb| sb.buffered().into_iter().map(|(a, b)| [a, b]).collect())
.unwrap_or_default();
serde_json::to_string(&r).unwrap()
});
let s = n.clone();
op!("__op_ms_set_duration", move |ms_id: u32, d: f64| {
if let Some(ms) = s.borrow_mut().mses.get_mut(&ms_id) {
ms.duration = d;
}
});
let s = n.clone();
op!("__op_ms_end", move |ms_id: u32, ended: bool| {
if let Some(ms) = s.borrow_mut().mses.get_mut(&ms_id) {
ms.ended = ended;
}
});
let s = n.clone();
op!("__op_media_create", move || -> u32 {
let mut st = s.borrow_mut();
st.count("media_create");
let id = st.id();
let m = MediaEngine::new(st.sink_factory.clone());
st.medias.insert(id, m);
id
});
with_media!("__op_media_attach", "media_attach", |m, ms_id: u32| m
.attach(ms_id));
with_media!("__op_media_detach", "media_detach", |m| m.detach());
with_media!("__op_media_play", "media_play", |m| m.play());
with_media!("__op_media_pause", "media_pause", |m| m.pause());
with_media!("__op_media_seek", "media_seek", |m, t: f64| m
.seek((t * 1e6).round() as i64));
with_media!(
"__op_media_set_loop",
"media_set_loop",
|m, looping: bool| m.looping = looping
);
with_media!("__op_media_set_gain", "media_set_gain", |m, gain: f64| m
.set_gain(gain as f32));
with_media!(
"__op_media_set_keys",
"media_set_keys",
|m, keys_id: u32| m.set_keys(Some(keys_id).filter(|k| *k != 0))
);
let s = n.clone();
op!("__op_media_state", move |id: u32| -> String {
s.borrow()
.medias
.get(&id)
.map(media_state)
.unwrap_or_else(|| "{}".into())
});
let s = n.clone();
op!("__op_destroy", move |kind: String, id: u32| {
let mut st = s.borrow_mut();
st.count("destroy");
match kind.as_str() {
"ms" => {
for sb in st.mses.remove(&id).map(|ms| ms.sbs).unwrap_or_default() {
st.sbs.remove(&sb);
}
}
"media" => {
if let Some(mut m) = st.medias.remove(&id) {
m.collect_underflows();
let counts = output_counts(&m.counters);
st.retired_output = add(st.retired_output, counts);
}
}
"cdm" => drop(st.cdms.remove(&id)),
_ => {}
}
});
op!("__op_can_play_type", |mime: String| can_play_type(&mime));
op!("__op_type_supported", |mime: String| audio_type_supported(
&mime
));
let s = n.clone();
op!("__op_eme_check", move |key_system: String,
configs: String|
-> String {
let mut st = s.borrow_mut();
st.count("eme_check");
let r = eme_check(&key_system, &configs);
st.eme_requests.push(json!({"keySystem": key_system, "requested": serde_json::from_str::<J>(&configs).unwrap_or(J::Null), "granted": r.clone()}));
r.map(|x| x.to_string()).unwrap_or_default()
});
let s = n.clone();
op!(
"__op_cdm_create",
move |allow_persistent_state: bool| -> i32 {
let mut st = s.borrow_mut();
st.count("cdm_create");
match (st.cdm_factory)(allow_persistent_state) {
Ok(c) => {
let id = st.id();
st.cdms.insert(id, c);
id as i32
}
Err(e) => {
eprintln!("ENGINE cdm create failed: {e:#}");
st.cdm_error = Some(format!("{e:#}"));
-1
}
}
}
);
with_cdm!(
"__op_cdm_set_cert",
"cdm_set_cert",
|c, promise, data: Value<'js>| c
.set_server_certificate(promise, &bytes_of(&data).unwrap_or_default())
);
with_cdm!(
"__op_cdm_generate",
"cdm_generate",
|c, promise, init_type: u32, data: Value<'js>| c.create_session(
promise,
init_type,
&bytes_of(&data).unwrap_or_default()
)
);
with_cdm!(
"__op_cdm_update",
"cdm_update",
|c, promise, sid: String, data: Value<'js>| c.update_session(
promise,
&sid,
&bytes_of(&data).unwrap_or_default()
)
);
with_cdm!("__op_cdm_close", "cdm_close", |c, promise, sid: String| c
.close_session(promise, &sid));
with_cdm!(
"__op_cdm_remove",
"cdm_remove",
|c, promise, sid: String| c.remove_session(promise, &sid)
);
Ok(())
}
fn call<'js, A: IntoArgs<'js>>(
ctx: &Ctx<'js>,
n: &N,
obj: &str,
name: &str,
args: A,
) -> Option<Value<'js>> {
let r = (|| -> rquickjs::Result<Value<'js>> {
let p: Object = ctx.globals().get(obj)?;
let f: Function = p.get(name)?;
f.call::<_, Value>(args)
})();
match r.catch(ctx) {
Ok(v) => Some(v),
Err(e) => {
let msg = crate::redact(&e.to_string(), &n.borrow().redact);
eprintln!("ENGINE call {obj}.{name} failed: {msg}");
None
}
}
}
fn buffer_or_null<'js>(ctx: &Ctx<'js>, body: Option<Vec<u8>>) -> Value<'js> {
body.and_then(|b| ArrayBuffer::new(ctx.clone(), b).ok())
.map(|b| b.into_value())
.unwrap_or_else(|| Value::new_null(ctx.clone()))
}
fn sha256_hex(bytes: &[u8]) -> String {
use sha2::{Digest, Sha256};
mp4::hex(&Sha256::digest(bytes))
}
fn handle_completion(ctx: &Ctx, n: &N, c: Completion) {
n.borrow_mut().in_flight -= 1;
match c {
Completion::Http(r) => {
{
let mut st = n.borrow_mut();
if st.aborted.remove(&r.id) {
return;
}
st.http_bytes += r.body.len() as u64;
}
let meta = json!({"status": r.status, "statusText": r.status_text, "url": r.url, "headers": r.headers, "error": r.error}).to_string();
call(
ctx,
n,
"__platform",
"dispatchHttp",
(r.id, meta, buffer_or_null(ctx, Some(r.body))),
);
}
Completion::Script { id, url, result } => {
let mut entry = json!({"id": id, "url": url});
let outcome = result.and_then(|mut bytes| {
entry["fetchedBytes"] = json!(bytes.len());
entry["fetchedSha256"] = json!(sha256_hex(&bytes));
if n.borrow().truncate_scripts {
bytes.truncate(bytes.len() / 2);
}
entry["evaluatedBytes"] = json!(bytes.len());
entry["evaluatedSha256"] = json!(sha256_hex(&bytes));
let t = Instant::now();
ctx.eval_with_options::<(), _>(bytes, opts(&url))
.catch(ctx)
.map(|_| t.elapsed())
.map_err(|e| e.to_string())
});
let (ok, err) = match outcome {
Ok(took) => {
entry["phase"] = json!("evaluated");
entry["evalMs"] = json!(took.as_secs_f64() * 1000.0);
(true, String::new())
}
Err(e) => {
entry["phase"] = json!("error");
entry["error"] = json!(e);
(false, e)
}
};
n.borrow_mut().script_log.push(entry);
call(ctx, n, "__platform", "dispatchScript", (id, ok, err));
}
}
}
#[cfg(feature = "trace")]
pub(crate) fn trace_meta(meta: &J) -> J {
let mut shown = meta.clone();
if let Some(count) = shown["keys"].as_array().map(Vec::len) {
shown["keys"] = json!(count);
}
if let Some(len) = shown["message"].as_str().map(str::len) {
shown["message"] = json!(len);
}
shown
}
fn dispatch_cdm(ctx: &Ctx, n: &N, keys_id: u32, e: CdmEvent) {
#[cfg(feature = "trace")]
let traced = format!("{:.3}", n.borrow().start.elapsed().as_secs_f64() * 1000.0);
let (meta, body) = match e {
CdmEvent::Initialized(ok) => (json!({"type": "initialized", "ok": ok}), None),
CdmEvent::NewSession { promise, sid } => (
json!({"type": "newSession", "promise": promise, "sid": sid}),
None,
),
CdmEvent::Resolve(p) => (json!({"type": "resolve", "promise": p}), None),
CdmEvent::Reject {
promise,
exception,
system_code,
message,
} => (
json!({"type": "reject", "promise": promise, "exception": exception, "code": system_code, "message": message}),
None,
),
CdmEvent::Message { sid, kind, body } => (
json!({"type": "message", "sid": sid, "kind": kind}),
Some(body),
),
CdmEvent::Keys { sid, keys } => (
json!({"type": "keys", "sid": sid, "keys": keys.iter().map(|(k, s)| json!({"kid": mp4::hex(k), "status": s})).collect::<Vec<_>>()}),
None,
),
CdmEvent::Expiration { sid, expiry } => (
json!({"type": "expiration", "sid": sid, "expiry": expiry}),
None,
),
CdmEvent::Closed { sid } => (json!({"type": "closed", "sid": sid}), None),
};
call(
ctx,
n,
"__platform",
"dispatchCdm",
(
keys_id,
{
#[cfg(feature = "trace")]
eprintln!(
"TRACE {traced} cdm-event keys={keys_id} {}",
trace_meta(&meta)
);
meta.to_string()
},
buffer_or_null(ctx, body),
),
);
}
impl Engine {
pub fn new(config: EngineConfig) -> Result<Engine> {
let EngineConfig {
cdm_path,
audio_sink_name,
output_device,
redact,
} = config;
let cdm: CdmFactory = Box::new(move |persistent| Widevine::create(&cdm_path, persistent));
let sink: SinkFactory = Rc::new(move || {
Box::new(PulseSink::start(
audio_sink_name.clone(),
output_device.clone(),
)) as Box<dyn AudioSink>
});
crate::net::clear_cache();
let engine = Engine::with_backends(cdm, sink)?;
{
let mut st = engine.n.borrow_mut();
st.redact = redact;
}
Ok(engine)
}
pub(crate) fn with_backends(
cdm_factory: CdmFactory,
sink_factory: SinkFactory,
) -> Result<Engine> {
let (net_tx, net_rx) = channel();
let n: N = Rc::new(RefCell::new(Native {
start: Instant::now(),
next_id: 0,
mses: HashMap::new(),
sbs: HashMap::new(),
medias: HashMap::new(),
cdms: HashMap::new(),
cdm_factory,
sink_factory,
cdm_error: None,
net_tx,
aborted: HashSet::new(),
op_counts: BTreeMap::new(),
eme_requests: Vec::new(),
script_log: Vec::new(),
http_count: 0,
http_bytes: 0,
redact: Vec::new(),
truncate_scripts: false,
retired_output: [0; OUTPUT_COUNTS],
in_flight: 0,
}));
let rt = Runtime::new()?;
rt.set_max_stack_size(4 * 1024 * 1024);
let ctx = Context::full(&rt)?;
ctx.with(|ctx| install_ops(&ctx, &n))
.map_err(|e| anyhow!("install ops: {e}"))?;
let engine = Engine {
ctx,
rt,
n,
net_rx,
last_gc: Instant::now(),
};
engine.eval_script("platform.js", PLATFORM)?;
#[cfg(feature = "trace")]
engine.eval_script("trace.js", include_str!("../js/trace.js"))?;
Ok(engine)
}
fn drain(&self) {
for _ in 0..100_000 {
match self.rt.execute_pending_job() {
Ok(true) => continue,
Ok(false) => break,
Err(e) => eprintln!("ENGINE job error: {e:?}"),
}
}
}
#[cfg(any(test, feature = "sabotage"))]
pub fn sabotage_truncate_scripts(&self) {
self.n.borrow_mut().truncate_scripts = true;
}
#[cfg(any(test, feature = "diagnostics"))]
pub fn enable_diagnostics(&self) -> Result<()> {
self.eval_script("diagnostics.js", include_str!("../js/diagnostics.js"))
}
pub fn eval_script(&self, name: &str, source: &str) -> Result<()> {
let r = self.ctx.with(|ctx| {
ctx.eval_with_options::<(), _>(source, opts(name))
.catch(&ctx)
.map_err(|e| anyhow!("{name} failed: {e}"))
});
self.drain();
r
}
pub fn eval_json(&self, expression: &str) -> Result<J> {
let src = format!("JSON.stringify(({expression}))");
let r = self.ctx.with(|ctx| {
ctx.eval_with_options::<Option<String>, _>(src, opts("eval"))
.catch(&ctx)
.map_err(|e| anyhow!("eval failed: {e}"))
});
self.drain();
Ok(r?
.map(|s| serde_json::from_str(&s))
.transpose()?
.unwrap_or(J::Null))
}
pub(crate) fn global_number(&self, object: &str, key: &str) -> Option<f64> {
self.ctx
.with(|ctx| ctx.globals().get::<_, Object>(object)?.get::<_, f64>(key))
.ok()
}
pub fn run_gc(&mut self) {
self.rt.run_gc();
self.last_gc = Instant::now();
self.drain();
}
fn run_tasks(&self) -> (bool, Option<f64>) {
for ran in 0..500 {
let r: f64 = self.ctx.with(|ctx| {
call(&ctx, &self.n, "__platform", "runTimers", ())
.and_then(|v| v.as_number())
.unwrap_or(-1.0)
});
self.drain();
if r != -2.0 {
return (ran > 0, (r >= 0.0).then_some(r));
}
}
(true, None)
}
pub fn tick(&mut self) -> Duration {
if self.last_gc.elapsed() >= GC_INTERVAL {
self.run_gc();
}
self.drain();
let mut busy = false;
while let Ok(c) = self.net_rx.try_recv() {
busy = true;
self.ctx.with(|ctx| handle_completion(&ctx, &self.n, c));
self.drain();
self.run_tasks();
}
let cdm_events: Vec<(u32, CdmEvent)> = self
.n
.borrow_mut()
.cdms
.iter_mut()
.flat_map(|(id, c)| c.poll().into_iter().map(move |e| (*id, e)))
.collect();
busy |= !cdm_events.is_empty();
for (id, e) in cdm_events {
self.ctx.with(|ctx| dispatch_cdm(&ctx, &self.n, id, e));
self.drain();
}
let (ran, next_ms) = self.run_tasks();
busy |= ran;
let now = Instant::now();
let media_events: Vec<(u32, String)> = {
let mut guard = self.n.borrow_mut();
let st = &mut *guard;
let mut out = Vec::new();
for (id, m) in st.medias.iter_mut() {
m.pump(now, &st.sbs, &st.mses, &mut st.cdms);
if !m.events.is_empty() {
let events = std::mem::take(&mut m.events);
out.push((
*id,
format!(
"{{\"events\":{},\"state\":{}}}",
json!(events),
media_state(m)
),
));
}
}
out
};
busy |= !media_events.is_empty();
for (id, payload) in media_events {
self.ctx.with(|ctx| {
call(&ctx, &self.n, "__platform", "dispatchMedia", (id, payload));
});
self.drain();
self.run_tasks();
}
let st = self.n.borrow();
let active = busy
|| st.in_flight > st.aborted.len()
|| self.rt.is_job_pending()
|| st.medias.values().any(|m| !m.idle());
if active {
let ms = next_ms.unwrap_or(ACTIVE_WAIT_MS).clamp(1.0, ACTIVE_WAIT_MS);
return Duration::from_micros((ms * 1000.0) as u64);
}
let js = next_ms.map(|ms| Duration::from_micros((ms.max(1.0) * 1000.0) as u64));
let cdm = st.cdms.values().filter_map(|c| c.next_timer()).min();
js.into_iter().chain(cdm).min().unwrap_or(IDLE_WAIT)
}
pub fn run_for(&mut self, d: Duration) {
let end = Instant::now() + d;
while Instant::now() < end {
let wait = self.tick();
std::thread::sleep(wait.min(end.saturating_duration_since(Instant::now())));
}
}
pub fn current_time(&self) -> Option<f64> {
self.n.borrow().primary_media().map(|m| m.current_time())
}
pub fn item_time(&self) -> Option<f64> {
let st = self.n.borrow();
let m = st.primary_media()?;
let t = m.current_time_us();
let offset = m
.ms_id
.and_then(|id| st.mses.get(&id))
.and_then(|ms| ms.sbs.first())
.and_then(|id| st.sbs.get(id))
.and_then(|sb| sb.offset_us_at(t))
.unwrap_or(0);
Some((t - offset).max(0) as f64 / 1e6)
}
pub fn output_failed(&self) -> bool {
self.n.borrow().medias.values().any(|m| m.output_failed())
}
pub fn output_counts(&self) -> OutputCounts {
let st = self.n.borrow();
let live = st.medias.values().map(|m| output_counts(&m.counters));
live.fold(st.retired_output, add)
}
pub fn status(&self) -> J {
let st = self.n.borrow();
let medias: serde_json::Map<String, J> = st
.medias
.iter()
.map(|(id, m)| {
let c = &m.counters;
let rate = if m.rate > 0 { m.rate as f64 } else { 48000.0 };
let v = json!({
"currentTime": m.current_time(), "maxCurrentTime": m.max_current_time_us as f64 / 1e6, "outputPosition": m.output_position(),
"readyState": m.ready_state, "paused": m.paused, "seeking": m.seeking, "ended": m.ended, "waitingForKey": m.waiting_for_key, "keysId": m.keys_id,
"rate": m.rate, "channels": m.channels, "clearSamples": c.clear_samples, "encryptedSamples": c.encrypted_samples, "decryptOk": c.decrypt_ok,
"decryptFail": c.decrypt_fail.iter().map(|(k, v)| (k.to_string(), json!(v))).collect::<serde_json::Map<_, _>>(),
"decodedClearSeconds": c.decoded_clear_frames as f64 / rate, "decodedDecryptedSeconds": c.decoded_decrypted_frames as f64 / rate,
"decodeErr": c.decode_err, "keyWaits": c.key_waits, "underruns": c.underruns, "seeks": c.seeks, "loops": c.loops,
"drainsAcked": c.drains_acked, "drainTimeouts": c.drain_timeouts, "sinkRestarts": c.sink_restarts,
});
(id.to_string(), v)
})
.collect();
let sbs: serde_json::Map<String, J> = st
.sbs
.iter()
.map(|(id, sb)| {
let init = sb.init.as_ref().map(|i| json!({"codec": i.codec, "timescale": i.timescale, "channels": i.channels, "sampleRate": i.sample_rate, "protected": i.protection.as_ref().map(|p| json!({"scheme": p.scheme, "kid": mp4::hex(&p.kid), "ivSize": p.iv_size})), "psshBytes": i.pssh.len()}));
(id.to_string(), json!({"mime": sb.mime, "appendedBytes": sb.appended_bytes, "bytes": sb.bytes, "evictedFrames": sb.evicted_frames, "frames": sb.frames.len(), "buffered": sb.buffered(), "init": init}))
})
.collect();
let platform = self
.ctx
.with(|ctx| {
call(&ctx, &self.n, "__platform", "statusJson", ())
.and_then(|v| v.as_string().and_then(|s| s.to_string().ok()))
})
.and_then(|s| serde_json::from_str(&s).ok())
.unwrap_or(J::Null);
json!({
"pid": std::process::id(),
"elapsedSec": st.start.elapsed().as_secs_f64(),
"opCounts": st.op_counts,
"live": {"medias": st.medias.len(), "mediaSources": st.mses.len(), "sourceBuffers": st.sbs.len(), "cdms": st.cdms.len()},
"medias": medias,
"sourceBuffers": sbs,
"cdm": st.cdm_error.as_ref().map(|e| json!({"error": e})).or_else(|| st.cdms.values().next().map(|c| c.describe())),
"emeRequests": st.eme_requests,
"scripts": st.script_log,
"http": {"count": st.http_count, "bytes": st.http_bytes},
"platform": platform,
})
}
}
pub const OUTPUT_COUNTS: usize = 8;
pub type OutputCounts = [u64; OUTPUT_COUNTS];
pub(crate) fn output_counts(c: &crate::media::Counters) -> OutputCounts {
let decrypt_fails = c.decrypt_fail.values().sum();
[
c.drains_acked,
c.drain_timeouts,
c.sent_us,
c.skipped_us,
c.decrypted_us,
c.underruns,
decrypt_fails,
c.output_underflows,
]
}
fn add(a: OutputCounts, b: OutputCounts) -> OutputCounts {
std::array::from_fn(|i| a[i] + b[i])
}