1use std::cell::RefCell;
2use std::collections::{BTreeMap, HashMap, HashSet};
3use std::rc::Rc;
4use std::sync::mpsc::{Receiver, Sender, channel};
5use std::time::{Duration, Instant};
6
7use anyhow::{Result, anyhow};
8use rquickjs::{
9 ArrayBuffer, CatchResultExt, Context, Ctx, Function, Object, Runtime, Value,
10 context::EvalOptions, function::IntoArgs,
11};
12use serde_json::{Value as J, json};
13
14use crate::EngineConfig;
15use crate::audio::{AudioSink, PulseSink};
16use crate::buffer::{MediaSourceNative, SOURCE_BUFFER_QUOTA_BYTES, SourceBufferNative};
17use crate::cdm::{CdmBackend, CdmEvent, Widevine};
18use crate::eme::{audio_type_supported, can_play_type, eme_check};
19use crate::media::{MediaEngine, SinkFactory};
20use crate::mp4;
21use crate::net::{self, Completion, HttpRequest};
22
23const PLATFORM: &str = include_str!("../js/platform.js");
24
25pub type CdmFactory = Box<dyn Fn(bool) -> Result<Box<dyn CdmBackend>>>;
26
27struct Native {
28 start: Instant,
29 next_id: u32,
30 mses: HashMap<u32, MediaSourceNative>,
31 sbs: HashMap<u32, SourceBufferNative>,
32 medias: HashMap<u32, MediaEngine>,
33 cdms: HashMap<u32, Box<dyn CdmBackend>>,
34 cdm_factory: CdmFactory,
35 sink_factory: SinkFactory,
36 cdm_error: Option<String>,
37 net_tx: Sender<Completion>,
38 aborted: HashSet<u32>,
39 op_counts: BTreeMap<&'static str, u64>,
40 eme_requests: Vec<J>,
41 script_log: Vec<J>,
42 http_count: u64,
43 http_bytes: u64,
44 redact: Vec<String>,
45 truncate_scripts: bool,
46 retired_output: OutputCounts,
47 in_flight: usize,
48}
49
50impl Native {
51 fn id(&mut self) -> u32 {
52 self.next_id += 1;
53 self.next_id
54 }
55 fn count(&mut self, op: &'static str) {
56 *self.op_counts.entry(op).or_default() += 1;
57 }
58 fn primary_media(&self) -> Option<&MediaEngine> {
59 self.medias
60 .iter()
61 .filter(|(_, m)| m.keys_id.is_some())
62 .min_by_key(|(id, _)| **id)
63 .map(|(_, m)| m)
64 }
65 fn position_for_buffer(&self, sb_id: u32) -> i64 {
66 self.medias
67 .values()
68 .find(|m| {
69 m.ms_id
70 .and_then(|id| self.mses.get(&id))
71 .is_some_and(|ms| ms.sbs.contains(&sb_id))
72 })
73 .map(|m| m.current_time_us())
74 .unwrap_or(0)
75 }
76}
77
78type N = Rc<RefCell<Native>>;
79
80pub struct Engine {
81 ctx: Context,
82 rt: Runtime,
83 n: N,
84 net_rx: Receiver<Completion>,
85 last_gc: Instant,
86}
87
88const GC_INTERVAL: Duration = Duration::from_secs(2);
89const ACTIVE_WAIT_MS: f64 = 5.0;
90const IDLE_WAIT: Duration = Duration::from_secs(60);
91
92fn opts(name: &str) -> EvalOptions {
93 let mut o = EvalOptions::default();
94 o.strict = false;
95 o.global = true;
96 o.filename = Some(name.into());
97 o
98}
99
100fn bytes_of(v: &Value) -> Option<Vec<u8>> {
101 let ab = ArrayBuffer::from_value(v.clone())?;
102 unsafe { ab.as_bytes() }.map(|b| b.to_vec())
103}
104
105fn media_state(m: &MediaEngine) -> String {
106 json!({"currentTime": m.current_time(), "readyState": m.ready_state, "paused": m.paused, "seeking": m.seeking, "ended": m.ended}).to_string()
107}
108
109fn install_ops<'js>(ctx: &Ctx<'js>, n: &N) -> rquickjs::Result<()> {
110 let g = ctx.globals();
111 macro_rules! op {
112 ($name:literal, $f:expr) => {
113 g.set($name, Function::new(ctx.clone(), $f)?)?;
114 };
115 }
116 macro_rules! with_media {
117 ($name:literal, $count:literal, |$m:ident $(, $arg:ident : $ty:ty)*| $body:expr) => {
118 let s = n.clone();
119 op!($name, move |id: u32 $(, $arg: $ty)*| {
120 let mut st = s.borrow_mut();
121 st.count($count);
122 if let Some($m) = st.medias.get_mut(&id) {
123 $body;
124 }
125 });
126 };
127 }
128 macro_rules! with_cdm {
129 ($name:literal, $count:literal, |$c:ident, $p:ident $(, $arg:ident : $ty:ty)*| $body:expr) => {
130 let s = n.clone();
131 op!($name, move |keys_id: u32, $p: u32 $(, $arg: $ty)*| {
132 let mut st = s.borrow_mut();
133 st.count($count);
134 if let Some($c) = st.cdms.get_mut(&keys_id) {
135 $body;
136 }
137 });
138 };
139 }
140 let s = n.clone();
141 op!("__op_now", move || -> f64 {
142 s.borrow().start.elapsed().as_secs_f64() * 1000.0
143 });
144 let s = n.clone();
145 op!("__op_print", move |level: String, msg: String| {
146 eprintln!("JS[{level}] {}", crate::redact(&msg, &s.borrow().redact));
147 });
148 op!("__op_user_agent", crate::user_agent);
149 op!("__op_random", move |ctx: Ctx<'js>,
150 len: u32|
151 -> rquickjs::Result<
152 ArrayBuffer<'js>,
153 > {
154 let mut buf = vec![0u8; len as usize];
155 if let Err(e) = getrandom::getrandom(&mut buf) {
156 return Err(rquickjs::Exception::throw_internal(
157 &ctx,
158 &format!("crypto.getRandomValues: no system randomness: {e}"),
159 ));
160 }
161 ArrayBuffer::new(ctx, buf)
162 });
163 let s = n.clone();
164 op!("__op_http", move |id: u32,
165 method: String,
166 url: String,
167 headers: String,
168 body: Value<'js>,
169 timeout: f64,
170 cache: String| {
171 let mut st = s.borrow_mut();
172 st.count("http");
173 st.http_count += 1;
174 st.in_flight += 1;
175 let headers: Vec<(String, String)> = serde_json::from_str(&headers).unwrap_or_default();
176 let req = HttpRequest {
177 id,
178 method,
179 url,
180 headers,
181 body: bytes_of(&body),
182 timeout_ms: if timeout > 0.0 {
183 timeout as u64
184 } else {
185 60_000
186 },
187 use_cache: cache == "default",
188 };
189 net::spawn(req, st.net_tx.clone());
190 });
191 let s = n.clone();
192 op!("__op_http_abort", move |id: u32| {
193 let mut st = s.borrow_mut();
194 st.count("http_abort");
195 st.aborted.insert(id);
196 });
197 let s = n.clone();
198 op!("__op_load_script", move |id: u32, url: String| {
199 let mut st = s.borrow_mut();
200 st.count("load_script");
201 st.in_flight += 1;
202 st.script_log
203 .push(json!({"id": id, "url": url, "phase": "fetching"}));
204 net::spawn_script(id, url, st.net_tx.clone());
205 });
206 let s = n.clone();
207 op!("__op_ms_create", move || -> u32 {
208 let mut st = s.borrow_mut();
209 st.count("ms_create");
210 let id = st.id();
211 st.mses.insert(
212 id,
213 MediaSourceNative {
214 duration: f64::NAN,
215 ended: false,
216 sbs: Vec::new(),
217 },
218 );
219 id
220 });
221 let s = n.clone();
222 op!("__op_sb_add", move |ms_id: u32, mime: String| -> i32 {
223 let mut st = s.borrow_mut();
224 st.count("sb_add");
225 if !audio_type_supported(&mime) {
226 return -1;
227 }
228 let id = st.id();
229 st.sbs
230 .insert(id, SourceBufferNative::new(mime, SOURCE_BUFFER_QUOTA_BYTES));
231 if let Some(ms) = st.mses.get_mut(&ms_id) {
232 ms.sbs.push(id);
233 }
234 id as i32
235 });
236 let s = n.clone();
237 op!("__op_sb_remove_buffer", move |ms_id: u32, sb_id: u32| {
238 let mut st = s.borrow_mut();
239 st.count("sb_remove_buffer");
240 st.sbs.remove(&sb_id);
241 if let Some(ms) = st.mses.get_mut(&ms_id) {
242 ms.sbs.retain(|x| *x != sb_id);
243 }
244 });
245 let s = n.clone();
246 op!("__op_sb_prepare_append", move |sb_id: u32,
247 len: u32|
248 -> bool {
249 let mut st = s.borrow_mut();
250 st.count("sb_prepare_append");
251 let position = st.position_for_buffer(sb_id);
252 st.sbs
253 .get_mut(&sb_id)
254 .is_some_and(|sb| sb.make_room(len as usize, position))
255 });
256 let s = n.clone();
257 op!("__op_sb_append", move |sb_id: u32,
258 data: Value<'js>,
259 ts_offset: f64|
260 -> String {
261 let mut st = s.borrow_mut();
262 st.count("sb_append");
263 let Some(bytes) = bytes_of(&data) else {
264 return json!({"error": "not an ArrayBuffer"}).to_string();
265 };
266 let Some(sb) = st.sbs.get_mut(&sb_id) else {
267 return json!({"error": "no such SourceBuffer"}).to_string();
268 };
269 match sb.append(&bytes, if ts_offset.is_finite() { ts_offset } else { 0.0 }) {
270 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(),
271 Err(e) => json!({"error": format!("{e:#}")}).to_string(),
272 }
273 });
274 let s = n.clone();
275 op!(
276 "__op_sb_set_append_window",
277 move |sb_id: u32, start: f64, end: f64| {
278 if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
279 let us = |t: f64| (t * 1e6).round() as i64;
280 sb.append_window_us = (us(start), us(end));
281 }
282 }
283 );
284 let s = n.clone();
285 op!("__op_sb_init_data", move |ctx: Ctx<'js>,
286 sb_id: u32|
287 -> rquickjs::Result<
288 Value<'js>,
289 > {
290 match s
291 .borrow()
292 .sbs
293 .get(&sb_id)
294 .and_then(|sb| sb.init_data.clone())
295 {
296 Some(b) => Ok(ArrayBuffer::new(ctx, b)?.into_value()),
297 None => Ok(Value::new_null(ctx)),
298 }
299 });
300 let s = n.clone();
301 op!("__op_sb_remove", move |sb_id: u32, start: f64, end: f64| {
302 let mut st = s.borrow_mut();
303 st.count("sb_remove");
304 if let Some(sb) = st.sbs.get_mut(&sb_id) {
305 sb.remove(start, end);
306 }
307 });
308 let s = n.clone();
309 op!("__op_sb_abort", move |sb_id: u32| {
310 if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
311 sb.abort();
312 }
313 });
314 let s = n.clone();
315 op!("__op_sb_change_type", move |sb_id: u32,
316 mime: String|
317 -> bool {
318 if !audio_type_supported(&mime) {
319 return false;
320 }
321 if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
322 sb.mime = mime;
323 }
324 true
325 });
326 let s = n.clone();
327 op!("__op_sb_buffered", move |sb_id: u32| -> String {
328 let r: Vec<[f64; 2]> = s
329 .borrow()
330 .sbs
331 .get(&sb_id)
332 .map(|sb| sb.buffered().into_iter().map(|(a, b)| [a, b]).collect())
333 .unwrap_or_default();
334 serde_json::to_string(&r).unwrap()
335 });
336 let s = n.clone();
337 op!("__op_ms_set_duration", move |ms_id: u32, d: f64| {
338 if let Some(ms) = s.borrow_mut().mses.get_mut(&ms_id) {
339 ms.duration = d;
340 }
341 });
342 let s = n.clone();
343 op!("__op_ms_end", move |ms_id: u32, ended: bool| {
344 if let Some(ms) = s.borrow_mut().mses.get_mut(&ms_id) {
345 ms.ended = ended;
346 }
347 });
348 let s = n.clone();
349 op!("__op_media_create", move || -> u32 {
350 let mut st = s.borrow_mut();
351 st.count("media_create");
352 let id = st.id();
353 let m = MediaEngine::new(st.sink_factory.clone());
354 st.medias.insert(id, m);
355 id
356 });
357 with_media!("__op_media_attach", "media_attach", |m, ms_id: u32| m
358 .attach(ms_id));
359 with_media!("__op_media_detach", "media_detach", |m| m.detach());
360 with_media!("__op_media_play", "media_play", |m| m.play());
361 with_media!("__op_media_pause", "media_pause", |m| m.pause());
362 with_media!("__op_media_seek", "media_seek", |m, t: f64| m
363 .seek((t * 1e6).round() as i64));
364 with_media!(
365 "__op_media_set_loop",
366 "media_set_loop",
367 |m, looping: bool| m.looping = looping
368 );
369 with_media!("__op_media_set_gain", "media_set_gain", |m, gain: f64| m
370 .set_gain(gain as f32));
371 with_media!(
372 "__op_media_set_keys",
373 "media_set_keys",
374 |m, keys_id: u32| m.set_keys(Some(keys_id).filter(|k| *k != 0))
375 );
376 let s = n.clone();
377 op!("__op_media_state", move |id: u32| -> String {
378 s.borrow()
379 .medias
380 .get(&id)
381 .map(media_state)
382 .unwrap_or_else(|| "{}".into())
383 });
384 let s = n.clone();
385 op!("__op_destroy", move |kind: String, id: u32| {
386 let mut st = s.borrow_mut();
387 st.count("destroy");
388 match kind.as_str() {
389 "ms" => {
390 for sb in st.mses.remove(&id).map(|ms| ms.sbs).unwrap_or_default() {
391 st.sbs.remove(&sb);
392 }
393 }
394 "media" => {
395 if let Some(mut m) = st.medias.remove(&id) {
396 m.collect_underflows();
397 let counts = output_counts(&m.counters);
398 st.retired_output = add(st.retired_output, counts);
399 }
400 }
401 "cdm" => drop(st.cdms.remove(&id)),
402 _ => {}
403 }
404 });
405 op!("__op_can_play_type", |mime: String| can_play_type(&mime));
406 op!("__op_type_supported", |mime: String| audio_type_supported(
407 &mime
408 ));
409 let s = n.clone();
410 op!("__op_eme_check", move |key_system: String,
411 configs: String|
412 -> String {
413 let mut st = s.borrow_mut();
414 st.count("eme_check");
415 let r = eme_check(&key_system, &configs);
416 st.eme_requests.push(json!({"keySystem": key_system, "requested": serde_json::from_str::<J>(&configs).unwrap_or(J::Null), "granted": r.clone()}));
417 r.map(|x| x.to_string()).unwrap_or_default()
418 });
419 let s = n.clone();
420 op!(
421 "__op_cdm_create",
422 move |allow_persistent_state: bool| -> i32 {
423 let mut st = s.borrow_mut();
424 st.count("cdm_create");
425 match (st.cdm_factory)(allow_persistent_state) {
426 Ok(c) => {
427 let id = st.id();
428 st.cdms.insert(id, c);
429 id as i32
430 }
431 Err(e) => {
432 eprintln!("ENGINE cdm create failed: {e:#}");
433 st.cdm_error = Some(format!("{e:#}"));
434 -1
435 }
436 }
437 }
438 );
439 with_cdm!(
440 "__op_cdm_set_cert",
441 "cdm_set_cert",
442 |c, promise, data: Value<'js>| c
443 .set_server_certificate(promise, &bytes_of(&data).unwrap_or_default())
444 );
445 with_cdm!(
446 "__op_cdm_generate",
447 "cdm_generate",
448 |c, promise, init_type: u32, data: Value<'js>| c.create_session(
449 promise,
450 init_type,
451 &bytes_of(&data).unwrap_or_default()
452 )
453 );
454 with_cdm!(
455 "__op_cdm_update",
456 "cdm_update",
457 |c, promise, sid: String, data: Value<'js>| c.update_session(
458 promise,
459 &sid,
460 &bytes_of(&data).unwrap_or_default()
461 )
462 );
463 with_cdm!("__op_cdm_close", "cdm_close", |c, promise, sid: String| c
464 .close_session(promise, &sid));
465 with_cdm!(
466 "__op_cdm_remove",
467 "cdm_remove",
468 |c, promise, sid: String| c.remove_session(promise, &sid)
469 );
470 Ok(())
471}
472
473fn call<'js, A: IntoArgs<'js>>(
474 ctx: &Ctx<'js>,
475 n: &N,
476 obj: &str,
477 name: &str,
478 args: A,
479) -> Option<Value<'js>> {
480 let r = (|| -> rquickjs::Result<Value<'js>> {
481 let p: Object = ctx.globals().get(obj)?;
482 let f: Function = p.get(name)?;
483 f.call::<_, Value>(args)
484 })();
485 match r.catch(ctx) {
486 Ok(v) => Some(v),
487 Err(e) => {
488 let msg = crate::redact(&e.to_string(), &n.borrow().redact);
489 eprintln!("ENGINE call {obj}.{name} failed: {msg}");
490 None
491 }
492 }
493}
494
495fn buffer_or_null<'js>(ctx: &Ctx<'js>, body: Option<Vec<u8>>) -> Value<'js> {
496 body.and_then(|b| ArrayBuffer::new(ctx.clone(), b).ok())
497 .map(|b| b.into_value())
498 .unwrap_or_else(|| Value::new_null(ctx.clone()))
499}
500
501fn sha256_hex(bytes: &[u8]) -> String {
502 use sha2::{Digest, Sha256};
503 mp4::hex(&Sha256::digest(bytes))
504}
505
506fn handle_completion(ctx: &Ctx, n: &N, c: Completion) {
507 n.borrow_mut().in_flight -= 1;
508 match c {
509 Completion::Http(r) => {
510 {
511 let mut st = n.borrow_mut();
512 if st.aborted.remove(&r.id) {
513 return;
514 }
515 st.http_bytes += r.body.len() as u64;
516 }
517 let meta = json!({"status": r.status, "statusText": r.status_text, "url": r.url, "headers": r.headers, "error": r.error}).to_string();
518 call(
519 ctx,
520 n,
521 "__platform",
522 "dispatchHttp",
523 (r.id, meta, buffer_or_null(ctx, Some(r.body))),
524 );
525 }
526 Completion::Script { id, url, result } => {
527 let mut entry = json!({"id": id, "url": url});
528 let outcome = result.and_then(|mut bytes| {
529 entry["fetchedBytes"] = json!(bytes.len());
530 entry["fetchedSha256"] = json!(sha256_hex(&bytes));
531 if n.borrow().truncate_scripts {
532 bytes.truncate(bytes.len() / 2);
533 }
534 entry["evaluatedBytes"] = json!(bytes.len());
535 entry["evaluatedSha256"] = json!(sha256_hex(&bytes));
536 let t = Instant::now();
537 ctx.eval_with_options::<(), _>(bytes, opts(&url))
538 .catch(ctx)
539 .map(|_| t.elapsed())
540 .map_err(|e| e.to_string())
541 });
542 let (ok, err) = match outcome {
543 Ok(took) => {
544 entry["phase"] = json!("evaluated");
545 entry["evalMs"] = json!(took.as_secs_f64() * 1000.0);
546 (true, String::new())
547 }
548 Err(e) => {
549 entry["phase"] = json!("error");
550 entry["error"] = json!(e);
551 (false, e)
552 }
553 };
554 n.borrow_mut().script_log.push(entry);
555 call(ctx, n, "__platform", "dispatchScript", (id, ok, err));
556 }
557 }
558}
559
560#[cfg(feature = "trace")]
561pub(crate) fn trace_meta(meta: &J) -> J {
562 let mut shown = meta.clone();
563 if let Some(count) = shown["keys"].as_array().map(Vec::len) {
564 shown["keys"] = json!(count);
565 }
566 if let Some(len) = shown["message"].as_str().map(str::len) {
567 shown["message"] = json!(len);
568 }
569 shown
570}
571
572fn dispatch_cdm(ctx: &Ctx, n: &N, keys_id: u32, e: CdmEvent) {
573 #[cfg(feature = "trace")]
574 let traced = format!("{:.3}", n.borrow().start.elapsed().as_secs_f64() * 1000.0);
575 let (meta, body) = match e {
576 CdmEvent::Initialized(ok) => (json!({"type": "initialized", "ok": ok}), None),
577 CdmEvent::NewSession { promise, sid } => (
578 json!({"type": "newSession", "promise": promise, "sid": sid}),
579 None,
580 ),
581 CdmEvent::Resolve(p) => (json!({"type": "resolve", "promise": p}), None),
582 CdmEvent::Reject {
583 promise,
584 exception,
585 system_code,
586 message,
587 } => (
588 json!({"type": "reject", "promise": promise, "exception": exception, "code": system_code, "message": message}),
589 None,
590 ),
591 CdmEvent::Message { sid, kind, body } => (
592 json!({"type": "message", "sid": sid, "kind": kind}),
593 Some(body),
594 ),
595 CdmEvent::Keys { sid, keys } => (
596 json!({"type": "keys", "sid": sid, "keys": keys.iter().map(|(k, s)| json!({"kid": mp4::hex(k), "status": s})).collect::<Vec<_>>()}),
597 None,
598 ),
599 CdmEvent::Expiration { sid, expiry } => (
600 json!({"type": "expiration", "sid": sid, "expiry": expiry}),
601 None,
602 ),
603 CdmEvent::Closed { sid } => (json!({"type": "closed", "sid": sid}), None),
604 };
605 call(
606 ctx,
607 n,
608 "__platform",
609 "dispatchCdm",
610 (
611 keys_id,
612 {
613 #[cfg(feature = "trace")]
614 eprintln!(
615 "TRACE {traced} cdm-event keys={keys_id} {}",
616 trace_meta(&meta)
617 );
618 meta.to_string()
619 },
620 buffer_or_null(ctx, body),
621 ),
622 );
623}
624
625impl Engine {
626 pub fn new(config: EngineConfig) -> Result<Engine> {
627 let EngineConfig {
628 cdm_path,
629 audio_sink_name,
630 output_device,
631 redact,
632 } = config;
633 let cdm: CdmFactory = Box::new(move |persistent| Widevine::create(&cdm_path, persistent));
634 let sink: SinkFactory = Rc::new(move || {
635 Box::new(PulseSink::start(
636 audio_sink_name.clone(),
637 output_device.clone(),
638 )) as Box<dyn AudioSink>
639 });
640 crate::net::clear_cache();
641 let engine = Engine::with_backends(cdm, sink)?;
642 {
643 let mut st = engine.n.borrow_mut();
644 st.redact = redact;
645 }
646 Ok(engine)
647 }
648
649 pub(crate) fn with_backends(
650 cdm_factory: CdmFactory,
651 sink_factory: SinkFactory,
652 ) -> Result<Engine> {
653 let (net_tx, net_rx) = channel();
654 let n: N = Rc::new(RefCell::new(Native {
655 start: Instant::now(),
656 next_id: 0,
657 mses: HashMap::new(),
658 sbs: HashMap::new(),
659 medias: HashMap::new(),
660 cdms: HashMap::new(),
661 cdm_factory,
662 sink_factory,
663 cdm_error: None,
664 net_tx,
665 aborted: HashSet::new(),
666 op_counts: BTreeMap::new(),
667 eme_requests: Vec::new(),
668 script_log: Vec::new(),
669 http_count: 0,
670 http_bytes: 0,
671 redact: Vec::new(),
672 truncate_scripts: false,
673 retired_output: [0; OUTPUT_COUNTS],
674 in_flight: 0,
675 }));
676 let rt = Runtime::new()?;
677 rt.set_max_stack_size(4 * 1024 * 1024);
678 let ctx = Context::full(&rt)?;
679 ctx.with(|ctx| install_ops(&ctx, &n))
680 .map_err(|e| anyhow!("install ops: {e}"))?;
681 let engine = Engine {
682 ctx,
683 rt,
684 n,
685 net_rx,
686 last_gc: Instant::now(),
687 };
688 engine.eval_script("platform.js", PLATFORM)?;
689 #[cfg(feature = "trace")]
690 engine.eval_script("trace.js", include_str!("../js/trace.js"))?;
691 Ok(engine)
692 }
693
694 fn drain(&self) {
695 for _ in 0..100_000 {
696 match self.rt.execute_pending_job() {
697 Ok(true) => continue,
698 Ok(false) => break,
699 Err(e) => eprintln!("ENGINE job error: {e:?}"),
700 }
701 }
702 }
703
704 #[cfg(any(test, feature = "sabotage"))]
705 pub fn sabotage_truncate_scripts(&self) {
706 self.n.borrow_mut().truncate_scripts = true;
707 }
708
709 #[cfg(any(test, feature = "diagnostics"))]
710 pub fn enable_diagnostics(&self) -> Result<()> {
711 self.eval_script("diagnostics.js", include_str!("../js/diagnostics.js"))
712 }
713
714 pub fn eval_script(&self, name: &str, source: &str) -> Result<()> {
715 let r = self.ctx.with(|ctx| {
716 ctx.eval_with_options::<(), _>(source, opts(name))
717 .catch(&ctx)
718 .map_err(|e| anyhow!("{name} failed: {e}"))
719 });
720 self.drain();
721 r
722 }
723
724 pub fn eval_json(&self, expression: &str) -> Result<J> {
725 let src = format!("JSON.stringify(({expression}))");
726 let r = self.ctx.with(|ctx| {
727 ctx.eval_with_options::<Option<String>, _>(src, opts("eval"))
728 .catch(&ctx)
729 .map_err(|e| anyhow!("eval failed: {e}"))
730 });
731 self.drain();
732 Ok(r?
733 .map(|s| serde_json::from_str(&s))
734 .transpose()?
735 .unwrap_or(J::Null))
736 }
737
738 pub(crate) fn global_number(&self, object: &str, key: &str) -> Option<f64> {
739 self.ctx
740 .with(|ctx| ctx.globals().get::<_, Object>(object)?.get::<_, f64>(key))
741 .ok()
742 }
743
744 pub fn run_gc(&mut self) {
745 self.rt.run_gc();
746 self.last_gc = Instant::now();
747 self.drain();
748 }
749
750 fn run_tasks(&self) -> (bool, Option<f64>) {
751 for ran in 0..500 {
752 let r: f64 = self.ctx.with(|ctx| {
753 call(&ctx, &self.n, "__platform", "runTimers", ())
754 .and_then(|v| v.as_number())
755 .unwrap_or(-1.0)
756 });
757 self.drain();
758 if r != -2.0 {
759 return (ran > 0, (r >= 0.0).then_some(r));
760 }
761 }
762 (true, None)
763 }
764
765 pub fn tick(&mut self) -> Duration {
766 if self.last_gc.elapsed() >= GC_INTERVAL {
767 self.run_gc();
768 }
769 self.drain();
770 let mut busy = false;
771 while let Ok(c) = self.net_rx.try_recv() {
772 busy = true;
773 self.ctx.with(|ctx| handle_completion(&ctx, &self.n, c));
774 self.drain();
775 self.run_tasks();
776 }
777 let cdm_events: Vec<(u32, CdmEvent)> = self
778 .n
779 .borrow_mut()
780 .cdms
781 .iter_mut()
782 .flat_map(|(id, c)| c.poll().into_iter().map(move |e| (*id, e)))
783 .collect();
784 busy |= !cdm_events.is_empty();
785 for (id, e) in cdm_events {
786 self.ctx.with(|ctx| dispatch_cdm(&ctx, &self.n, id, e));
787 self.drain();
788 }
789 let (ran, next_ms) = self.run_tasks();
790 busy |= ran;
791 let now = Instant::now();
792 let media_events: Vec<(u32, String)> = {
793 let mut guard = self.n.borrow_mut();
794 let st = &mut *guard;
795 let mut out = Vec::new();
796 for (id, m) in st.medias.iter_mut() {
797 m.pump(now, &st.sbs, &st.mses, &mut st.cdms);
798 if !m.events.is_empty() {
799 let events = std::mem::take(&mut m.events);
800 out.push((
801 *id,
802 format!(
803 "{{\"events\":{},\"state\":{}}}",
804 json!(events),
805 media_state(m)
806 ),
807 ));
808 }
809 }
810 out
811 };
812 busy |= !media_events.is_empty();
813 for (id, payload) in media_events {
814 self.ctx.with(|ctx| {
815 call(&ctx, &self.n, "__platform", "dispatchMedia", (id, payload));
816 });
817 self.drain();
818 self.run_tasks();
819 }
820 let st = self.n.borrow();
821 let active = busy
822 || st.in_flight > st.aborted.len()
823 || self.rt.is_job_pending()
824 || st.medias.values().any(|m| !m.idle());
825 if active {
826 let ms = next_ms.unwrap_or(ACTIVE_WAIT_MS).clamp(1.0, ACTIVE_WAIT_MS);
827 return Duration::from_micros((ms * 1000.0) as u64);
828 }
829 let js = next_ms.map(|ms| Duration::from_micros((ms.max(1.0) * 1000.0) as u64));
830 let cdm = st.cdms.values().filter_map(|c| c.next_timer()).min();
831 js.into_iter().chain(cdm).min().unwrap_or(IDLE_WAIT)
832 }
833
834 pub fn run_for(&mut self, d: Duration) {
835 let end = Instant::now() + d;
836 while Instant::now() < end {
837 let wait = self.tick();
838 std::thread::sleep(wait.min(end.saturating_duration_since(Instant::now())));
839 }
840 }
841
842 pub fn current_time(&self) -> Option<f64> {
843 self.n.borrow().primary_media().map(|m| m.current_time())
844 }
845
846 pub fn item_time(&self) -> Option<f64> {
847 let st = self.n.borrow();
848 let m = st.primary_media()?;
849 let t = m.current_time_us();
850 let offset = m
851 .ms_id
852 .and_then(|id| st.mses.get(&id))
853 .and_then(|ms| ms.sbs.first())
854 .and_then(|id| st.sbs.get(id))
855 .and_then(|sb| sb.offset_us_at(t))
856 .unwrap_or(0);
857 Some((t - offset).max(0) as f64 / 1e6)
858 }
859
860 pub fn output_failed(&self) -> bool {
861 self.n.borrow().medias.values().any(|m| m.output_failed())
862 }
863
864 pub fn output_counts(&self) -> OutputCounts {
865 let st = self.n.borrow();
866 let live = st.medias.values().map(|m| output_counts(&m.counters));
867 live.fold(st.retired_output, add)
868 }
869
870 pub fn status(&self) -> J {
871 let st = self.n.borrow();
872 let medias: serde_json::Map<String, J> = st
873 .medias
874 .iter()
875 .map(|(id, m)| {
876 let c = &m.counters;
877 let rate = if m.rate > 0 { m.rate as f64 } else { 48000.0 };
878 let v = json!({
879 "currentTime": m.current_time(), "maxCurrentTime": m.max_current_time_us as f64 / 1e6, "outputPosition": m.output_position(),
880 "readyState": m.ready_state, "paused": m.paused, "seeking": m.seeking, "ended": m.ended, "waitingForKey": m.waiting_for_key, "keysId": m.keys_id,
881 "rate": m.rate, "channels": m.channels, "clearSamples": c.clear_samples, "encryptedSamples": c.encrypted_samples, "decryptOk": c.decrypt_ok,
882 "decryptFail": c.decrypt_fail.iter().map(|(k, v)| (k.to_string(), json!(v))).collect::<serde_json::Map<_, _>>(),
883 "decodedClearSeconds": c.decoded_clear_frames as f64 / rate, "decodedDecryptedSeconds": c.decoded_decrypted_frames as f64 / rate,
884 "decodeErr": c.decode_err, "keyWaits": c.key_waits, "underruns": c.underruns, "seeks": c.seeks, "loops": c.loops,
885 "drainsAcked": c.drains_acked, "drainTimeouts": c.drain_timeouts, "sinkRestarts": c.sink_restarts,
886 });
887 (id.to_string(), v)
888 })
889 .collect();
890 let sbs: serde_json::Map<String, J> = st
891 .sbs
892 .iter()
893 .map(|(id, sb)| {
894 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()}));
895 (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}))
896 })
897 .collect();
898 let platform = self
899 .ctx
900 .with(|ctx| {
901 call(&ctx, &self.n, "__platform", "statusJson", ())
902 .and_then(|v| v.as_string().and_then(|s| s.to_string().ok()))
903 })
904 .and_then(|s| serde_json::from_str(&s).ok())
905 .unwrap_or(J::Null);
906 json!({
907 "pid": std::process::id(),
908 "elapsedSec": st.start.elapsed().as_secs_f64(),
909 "opCounts": st.op_counts,
910 "live": {"medias": st.medias.len(), "mediaSources": st.mses.len(), "sourceBuffers": st.sbs.len(), "cdms": st.cdms.len()},
911 "medias": medias,
912 "sourceBuffers": sbs,
913 "cdm": st.cdm_error.as_ref().map(|e| json!({"error": e})).or_else(|| st.cdms.values().next().map(|c| c.describe())),
914 "emeRequests": st.eme_requests,
915 "scripts": st.script_log,
916 "http": {"count": st.http_count, "bytes": st.http_bytes},
917 "platform": platform,
918 })
919 }
920}
921
922pub const OUTPUT_COUNTS: usize = 8;
923pub type OutputCounts = [u64; OUTPUT_COUNTS];
924
925pub(crate) fn output_counts(c: &crate::media::Counters) -> OutputCounts {
926 let decrypt_fails = c.decrypt_fail.values().sum();
927 [
928 c.drains_acked,
929 c.drain_timeouts,
930 c.sent_us,
931 c.skipped_us,
932 c.decrypted_us,
933 c.underruns,
934 decrypt_fails,
935 c.output_underflows,
936 ]
937}
938
939fn add(a: OutputCounts, b: OutputCounts) -> OutputCounts {
940 std::array::from_fn(|i| a[i] + b[i])
941}