use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use rquickjs::function::{Func, Rest};
use rquickjs::{Class, Ctx, Function, JsLifetime, Value, class::Trace};
use tokio::sync::Notify;
const MAX_FREE_NESTING: u32 = 5;
#[derive(Clone)]
struct Nesting(Arc<AtomicU32>);
#[allow(unsafe_code)]
unsafe impl JsLifetime<'_> for Nesting {
type Changed<'to> = Nesting;
}
fn nesting_of(ctx: &Ctx<'_>) -> Arc<AtomicU32> {
ctx
.userdata::<Nesting>()
.map_or_else(|| Arc::new(AtomicU32::new(0)), |n| Arc::clone(&n.0))
}
pub trait CallbackPolicy: Clone + 'static {
fn capture(ctx: &Ctx<'_>) -> Option<Self>
where
Self: Sized;
fn enter<R>(ctx: &Ctx<'_>, policy: Option<&Self>, f: impl FnOnce() -> R) -> R
where
Self: Sized;
}
#[derive(Clone, Copy)]
pub struct NoPolicy;
impl CallbackPolicy for NoPolicy {
fn capture(_ctx: &Ctx<'_>) -> Option<Self> {
None
}
fn enter<R>(_ctx: &Ctx<'_>, _policy: Option<&Self>, f: impl FnOnce() -> R) -> R {
f()
}
}
#[derive(Trace, JsLifetime)]
#[rquickjs::class]
pub struct Timeout {
#[qjs(skip_trace)]
abort: Arc<Notify>,
}
fn clear_timeout(value: Rest<Value<'_>>) {
if let Some(v) = value.0.first() {
if let Ok(timeout) = Class::<Timeout>::from_value(v) {
timeout.borrow().abort.notify_one();
}
}
}
fn requested_ms(msec: Option<f64>) -> u64 {
match msec {
Some(ms) if ms.is_finite() && ms >= 1.0 && ms < f64::from(i32::MAX) => ms as u64,
_ => 0,
}
}
fn clamped(requested: u64, level: u32) -> Duration {
if level > MAX_FREE_NESTING && requested < 4 {
Duration::from_millis(4)
} else {
Duration::from_millis(requested)
}
}
async fn wait(delay: Duration) {
if delay.is_zero() {
tokio::task::yield_now().await;
} else {
tokio::time::sleep(delay).await;
}
}
fn set_timeout_interval<'js, P: CallbackPolicy>(
ctx: Ctx<'js>,
cb: Function<'js>,
msec: Option<f64>,
args: Vec<Value<'js>>,
is_interval: bool,
) -> rquickjs::Result<Class<'js, Timeout>> {
let requested = requested_ms(msec);
let nesting = nesting_of(&ctx);
let level = nesting.load(Ordering::Relaxed).saturating_add(1);
let abort = Arc::new(Notify::new());
let abort_ref = abort.clone();
let policy = P::capture(&ctx);
ctx.spawn(async move {
let fire = |level: u32| {
let mut call_args = rquickjs::function::Args::new(cb.ctx().clone(), args.len());
if call_args.push_args(args.iter().cloned()).is_err() {
return false;
}
let outer = nesting.swap(level, Ordering::Relaxed);
let res: rquickjs::Result<()> = P::enter(cb.ctx(), policy.as_ref(), || cb.call_arg(call_args));
nesting.store(outer, Ordering::Relaxed);
res
.inspect_err(|err| tracing::warn!(target: "ferrijs::timers", "timer callback threw: {err}"))
.is_ok()
};
if !is_interval {
tokio::select! {
() = abort_ref.notified() => {},
() = wait(clamped(requested, level)) => { fire(level); },
}
return;
}
let mut level = level;
let mut next = tokio::time::Instant::now() + clamped(requested, level);
loop {
let delay = next.saturating_duration_since(tokio::time::Instant::now());
let aborted = tokio::select! {
() = abort_ref.notified() => true,
() = wait(delay) => false,
};
if aborted || !fire(level) {
break;
}
level = level.saturating_add(1);
let period = clamped(requested, level);
next += period;
let now = tokio::time::Instant::now();
if next <= now {
next = now + period;
}
}
});
Class::instance(ctx, Timeout { abort })
}
fn set_timeout<'js, P: CallbackPolicy>(
ctx: Ctx<'js>,
cb: Function<'js>,
rest: Rest<Value<'js>>,
) -> rquickjs::Result<Class<'js, Timeout>> {
let (msec, args) = split_delay_args(rest.0);
set_timeout_interval::<P>(ctx, cb, msec, args, false)
}
fn set_interval<'js, P: CallbackPolicy>(
ctx: Ctx<'js>,
cb: Function<'js>,
rest: Rest<Value<'js>>,
) -> rquickjs::Result<Class<'js, Timeout>> {
let (msec, args) = split_delay_args(rest.0);
set_timeout_interval::<P>(ctx, cb, msec, args, true)
}
fn split_delay_args(mut rest: Vec<Value<'_>>) -> (Option<f64>, Vec<Value<'_>>) {
if rest.is_empty() {
return (None, rest);
}
let delay = rest.remove(0);
(delay.as_number(), rest)
}
fn set_immediate<'js, P: CallbackPolicy>(
ctx: Ctx<'js>,
cb: Function<'js>,
rest: Rest<Value<'js>>,
) -> rquickjs::Result<()> {
match P::capture(&ctx) {
None => {
let mut args = rquickjs::function::Args::new(ctx, rest.0.len());
args.push_args(rest.0)?;
cb.defer_arg(args)
},
Some(policy) => {
let policy = Some(policy);
let wrapper = Function::new(ctx.clone(), move |args: Rest<Value<'_>>| {
deferred_call::<P>(policy.as_ref(), &args.0)
})?;
let mut args = rquickjs::function::Args::new(ctx, rest.0.len() + 1);
args.push_arg(cb)?;
args.push_args(rest.0)?;
wrapper.defer_arg(args)
},
}
}
fn deferred_call<P: CallbackPolicy>(policy: Option<&P>, args: &[Value<'_>]) -> rquickjs::Result<()> {
let inner = args.first().and_then(|v| v.as_function().cloned()).ok_or_else(|| {
rquickjs::Error::new_from_js_message("setImmediate", "Error", "deferred callback missing".to_string())
})?;
let ctx = inner.ctx().clone();
let mut call_args = rquickjs::function::Args::new(ctx.clone(), args.len().saturating_sub(1));
call_args.push_args(args.iter().skip(1).cloned())?;
P::enter(&ctx, policy, || inner.call_arg(call_args))
}
fn queue_microtask<'js, P: CallbackPolicy>(ctx: Ctx<'js>, cb: Function<'js>) -> rquickjs::Result<()> {
match P::capture(&ctx) {
None => cb.defer::<()>(()),
Some(policy) => {
let policy = Some(policy);
let wrapper = Function::new(ctx.clone(), move |args: Rest<Value<'_>>| {
deferred_call::<P>(policy.as_ref(), &args.0)
})?;
wrapper.defer((cb,))
},
}
}
pub fn install<P: CallbackPolicy>(ctx: &Ctx<'_>) -> rquickjs::Result<()> {
let globals = ctx.globals();
let _ = ctx.store_userdata(Nesting(Arc::new(AtomicU32::new(0))));
globals.set("setTimeout", Func::from(set_timeout::<P>))?;
globals.set("clearTimeout", Func::from(clear_timeout))?;
globals.set("setInterval", Func::from(set_interval::<P>))?;
globals.set("clearInterval", Func::from(clear_timeout))?;
globals.set("setImmediate", Func::from(set_immediate::<P>))?;
globals.set("queueMicrotask", Func::from(queue_microtask::<P>))?;
Ok(())
}