use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use chrono::{DateTime, Datelike, Duration as ChronoDuration, Local, Timelike, Utc, Weekday};
use scc::HashMap as SccHashMap;
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;
use crate::agent_os::AgentOs;
use crate::config::{ScheduleDriver, ScheduleEntry, ScheduleHandle};
use crate::error::ClientError;
use crate::session::CreateSessionOptions;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum CronOverlap {
#[default]
Allow,
Skip,
Queue,
}
#[derive(Clone)]
pub enum CronAction {
Session {
agent_type: String,
prompt: String,
options: Option<CreateSessionOptions>,
},
Exec { command: String, args: Vec<String> },
Callback {
#[allow(clippy::type_complexity)]
callback: Arc<dyn Fn() -> futures::future::BoxFuture<'static, ()> + Send + Sync>,
},
}
impl std::fmt::Debug for CronAction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
CronAction::Session {
agent_type, prompt, ..
} => f
.debug_struct("Session")
.field("agent_type", agent_type)
.field("prompt", prompt)
.finish_non_exhaustive(),
CronAction::Exec { command, args } => f
.debug_struct("Exec")
.field("command", command)
.field("args", args)
.finish(),
CronAction::Callback { .. } => f.debug_struct("Callback").finish_non_exhaustive(),
}
}
}
#[derive(Clone)]
pub struct CronJobOptions {
pub id: Option<String>,
pub schedule: String,
pub action: CronAction,
pub overlap: Option<CronOverlap>,
}
#[derive(Debug, Clone)]
pub struct CronJobInfo {
pub id: String,
pub schedule: String,
pub action: CronAction,
pub overlap: CronOverlap,
pub last_run: Option<DateTime<Utc>>,
pub next_run: Option<DateTime<Utc>>,
pub run_count: u64,
pub running: bool,
}
#[derive(Debug, Clone)]
pub enum CronEvent {
Fire {
job_id: String,
time: DateTime<Utc>,
},
Complete {
job_id: String,
time: DateTime<Utc>,
duration_ms: f64,
},
Error {
job_id: String,
time: DateTime<Utc>,
error: String,
},
}
#[derive(Clone)]
pub struct CronJobHandle {
pub id: String,
pub(crate) manager: Arc<CronManager>,
}
impl CronJobHandle {
pub fn cancel(&self) {
self.manager.cancel_job(&self.id);
}
}
pub(crate) struct CronJobState {
pub schedule: String,
pub action: CronAction,
pub overlap: CronOverlap,
pub last_run: parking_lot::Mutex<Option<DateTime<Utc>>>,
pub next_run: parking_lot::Mutex<Option<DateTime<Utc>>>,
pub run_count: std::sync::atomic::AtomicU64,
pub running: AtomicBool,
pub queued: AtomicBool,
pub handle: ScheduleHandle,
}
pub struct CronManager {
pub(crate) jobs: SccHashMap<String, CronJobState>,
pub(crate) driver: Arc<dyn ScheduleDriver>,
pub(crate) event_tx: broadcast::Sender<CronEvent>,
}
impl CronManager {
pub(crate) fn new(driver: Arc<dyn ScheduleDriver>) -> Self {
let (event_tx, _rx) = broadcast::channel(256);
Self {
jobs: SccHashMap::new(),
driver,
event_tx,
}
}
pub(crate) fn cancel_job(&self, id: &str) {
if let Some((_, state)) = self.jobs.remove(id) {
self.driver.cancel(&state.handle);
}
}
pub(crate) fn dispose(&self) {
self.jobs.scan(|_, state| {
self.driver.cancel(&state.handle);
});
self.jobs.clear();
self.driver.dispose();
}
}
fn execute_job(
manager: Arc<CronManager>,
vm: AgentOs,
id: String,
) -> futures::future::BoxFuture<'static, ()> {
Box::pin(execute_job_inner(manager, vm, id))
}
async fn execute_job_inner(manager: Arc<CronManager>, vm: AgentOs, id: String) {
let manager = &manager;
let vm = &vm;
let id = id.as_str();
{
let mut should_return = false;
let mut should_queue = false;
manager.jobs.read(id, |_, state| {
if state.running.load(Ordering::SeqCst) {
match state.overlap {
CronOverlap::Allow => {}
CronOverlap::Skip => should_return = true,
CronOverlap::Queue => should_queue = true,
}
}
});
if should_return {
return;
}
if should_queue {
manager.jobs.read(id, |_, state| {
state.queued.store(true, Ordering::SeqCst);
});
return;
}
}
let action = match manager.jobs.read(id, |_, state| {
state.running.store(true, Ordering::SeqCst);
*state.last_run.lock() = Some(Utc::now());
state.run_count.fetch_add(1, Ordering::SeqCst);
state.action.clone()
}) {
Some(action) => action,
None => return,
};
let _ = manager.event_tx.send(CronEvent::Fire {
job_id: id.to_string(),
time: Utc::now(),
});
let start = Utc::now();
let result = run_action(vm, &action).await;
let duration_ms = (Utc::now() - start).num_milliseconds() as f64;
match result {
Ok(()) => {
let _ = manager.event_tx.send(CronEvent::Complete {
job_id: id.to_string(),
time: Utc::now(),
duration_ms,
});
}
Err(error) => {
let _ = manager.event_tx.send(CronEvent::Error {
job_id: id.to_string(),
time: Utc::now(),
error: error.to_string(),
});
}
}
let mut run_queued = false;
manager.jobs.read(id, |_, state| {
state.running.store(false, Ordering::SeqCst);
*state.next_run.lock() = compute_next_time(&state.schedule, Utc::now());
if state.queued.swap(false, Ordering::SeqCst) {
run_queued = true;
}
});
if run_queued {
let manager = Arc::clone(manager);
let vm = vm.clone();
let id = id.to_string();
tokio::spawn(execute_job(manager, vm, id));
}
}
async fn run_action(vm: &AgentOs, action: &CronAction) -> Result<(), ClientError> {
match action {
CronAction::Session {
agent_type,
prompt,
options,
} => {
let session = vm
.create_session(agent_type, options.clone().unwrap_or_default())
.await
.map_err(|err| ClientError::Sidecar(err.to_string()))?;
let prompt_result = vm.prompt(&session.session_id, prompt).await;
let _ = vm.close_session(&session.session_id);
prompt_result.map_err(|err| ClientError::Sidecar(err.to_string()))?;
Ok(())
}
CronAction::Exec { command, args } => {
let cmd = if args.is_empty() {
command.clone()
} else {
format!("{} {}", command, args.join(" "))
};
vm.exec(&cmd, crate::process::ExecOptions::default())
.await
.map_err(|err| ClientError::Sidecar(err.to_string()))?;
Ok(())
}
CronAction::Callback { callback } => {
callback().await;
Ok(())
}
}
}
pub(crate) enum ParsedSchedule {
Date(DateTime<Utc>),
Cron(CronExpr),
}
impl ParsedSchedule {
pub(crate) fn is_cron(&self) -> bool {
matches!(self, ParsedSchedule::Cron(_))
}
}
pub(crate) fn resolve_next_run(
parsed: &ParsedSchedule,
now: DateTime<Utc>,
) -> Option<DateTime<Utc>> {
match parsed {
ParsedSchedule::Cron(cron) => cron.next_after(now),
ParsedSchedule::Date(date) => {
if date.timestamp_millis() > now.timestamp_millis() {
Some(*date)
} else {
None
}
}
}
}
fn looks_like_one_shot(schedule: &str) -> bool {
let bytes = schedule.as_bytes();
let mut i = 0usize;
let is_digit = |b: u8| b.is_ascii_digit();
let take_digits = |bytes: &[u8], i: &mut usize, n: usize| -> bool {
for _ in 0..n {
match bytes.get(*i) {
Some(&b) if is_digit(b) => *i += 1,
_ => return false,
}
}
true
};
let take_lit = |bytes: &[u8], i: &mut usize, lit: u8| -> bool {
match bytes.get(*i) {
Some(&b) if b == lit => {
*i += 1;
true
}
_ => false,
}
};
if !take_digits(bytes, &mut i, 4) {
return false;
}
if !take_lit(bytes, &mut i, b'-') {
return false;
}
if !take_digits(bytes, &mut i, 2) {
return false;
}
if !take_lit(bytes, &mut i, b'-') {
return false;
}
if !take_digits(bytes, &mut i, 2) {
return false;
}
if i == bytes.len() {
return true;
}
match bytes.get(i) {
Some(b'T') | Some(b' ') => i += 1,
_ => return false,
}
if !take_digits(bytes, &mut i, 2) {
return false;
}
if !take_lit(bytes, &mut i, b':') {
return false;
}
if !take_digits(bytes, &mut i, 2) {
return false;
}
if take_lit(bytes, &mut i, b':') {
if !take_digits(bytes, &mut i, 2) {
return false;
}
if take_lit(bytes, &mut i, b'.') {
let mut frac = 0;
while matches!(bytes.get(i), Some(&b) if is_digit(b)) {
i += 1;
frac += 1;
}
if frac == 0 {
return false;
}
}
}
match bytes.get(i) {
None => return true,
Some(b'Z') => {
i += 1;
}
Some(b'+') | Some(b'-') => {
i += 1;
if !take_digits(bytes, &mut i, 2) {
return false;
}
if !take_lit(bytes, &mut i, b':') {
return false;
}
if !take_digits(bytes, &mut i, 2) {
return false;
}
}
_ => return false,
}
i == bytes.len()
}
fn parse_one_shot(schedule: &str) -> Option<DateTime<Utc>> {
use chrono::TimeZone;
if let Ok(dt) = DateTime::parse_from_rfc3339(schedule) {
return Some(dt.with_timezone(&Utc));
}
let normalized = schedule.replacen(' ', "T", 1);
for fmt in ["%Y-%m-%dT%H:%M:%S%.f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"] {
if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(&normalized, fmt) {
return match Local.from_local_datetime(&naive) {
chrono::LocalResult::Single(dt) => Some(dt.with_timezone(&Utc)),
chrono::LocalResult::Ambiguous(dt, _) => Some(dt.with_timezone(&Utc)),
chrono::LocalResult::None => None,
};
}
}
if let Ok(date) = chrono::NaiveDate::parse_from_str(schedule, "%Y-%m-%d") {
let naive = date.and_hms_opt(0, 0, 0)?;
return Some(DateTime::<Utc>::from_naive_utc_and_offset(naive, Utc));
}
None
}
pub(crate) fn parse_schedule(schedule: &str) -> std::result::Result<ParsedSchedule, ClientError> {
let normalized = schedule.trim();
if looks_like_one_shot(normalized) {
return match parse_one_shot(normalized) {
Some(date) => Ok(ParsedSchedule::Date(date)),
None => Err(ClientError::InvalidSchedule(schedule.to_string())),
};
}
match CronExpr::parse(normalized) {
Ok(cron) => Ok(ParsedSchedule::Cron(cron)),
Err(_) => Err(ClientError::InvalidSchedule(schedule.to_string())),
}
}
pub(crate) fn compute_next_time(schedule: &str, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
let parsed = parse_schedule(schedule).ok()?;
resolve_next_run(&parsed, now)
}
pub(crate) fn validate_schedule(
schedule: &str,
now: DateTime<Utc>,
) -> std::result::Result<Option<DateTime<Utc>>, ClientError> {
let parsed = parse_schedule(schedule)?;
match parsed {
ParsedSchedule::Cron(cron) => Ok(cron.next_after(now)),
ParsedSchedule::Date(date) => {
if date.timestamp_millis() > now.timestamp_millis() {
Ok(Some(date))
} else {
Err(ClientError::PastSchedule(schedule.to_string()))
}
}
}
}
pub(crate) struct CronExpr {
seconds: Vec<u32>,
minutes: Vec<u32>,
hours: Vec<u32>,
days_of_month: Vec<u32>,
months: Vec<u32>,
days_of_week: Vec<u32>,
years: Option<Vec<u32>>,
dom_restricted: bool,
dow_restricted: bool,
dom_last: bool,
dom_last_weekday: bool,
dom_nearest_weekday: Option<u32>,
dow_last: Option<u32>,
dow_nth: Option<(u32, u32)>,
}
const MONTH_NAMES: [&str; 12] = [
"JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC",
];
const WEEKDAY_NAMES: [&str; 7] = ["SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT"];
impl CronExpr {
fn parse(expr: &str) -> std::result::Result<Self, ()> {
let fields: Vec<&str> = expr.split_whitespace().collect();
let (sec, min, hour, dom, month, dow, year): (
&str,
&str,
&str,
&str,
&str,
&str,
Option<&str>,
) = match fields.len() {
5 => ("0", fields[0], fields[1], fields[2], fields[3], fields[4], None),
6 => (
fields[0], fields[1], fields[2], fields[3], fields[4], fields[5], None,
),
7 => (
fields[0],
fields[1],
fields[2],
fields[3],
fields[4],
fields[5],
Some(fields[6]),
),
_ => return Err(()),
};
let seconds = parse_field(sec, 0, 59, FieldKind::Plain)?;
let minutes = parse_field(min, 0, 59, FieldKind::Plain)?;
let hours = parse_field(hour, 0, 23, FieldKind::Plain)?;
let mut dom_last = false;
let mut dom_last_weekday = false;
let mut dom_nearest_weekday = None;
let days_of_month = parse_dom_field(
dom,
&mut dom_last,
&mut dom_last_weekday,
&mut dom_nearest_weekday,
)?;
let months = parse_field(month, 1, 12, FieldKind::Month)?;
let mut dow_last = None;
let mut dow_nth = None;
let days_of_week = parse_dow_field(dow, &mut dow_last, &mut dow_nth)?;
let years = match year {
Some(y) => Some(parse_field(y, 1970, 2099, FieldKind::Plain)?),
None => None,
};
let dom_restricted = dom != "*" && dom != "?";
let dow_restricted = dow != "*" && dow != "?";
Ok(Self {
seconds,
minutes,
hours,
days_of_month,
months,
days_of_week,
years,
dom_restricted,
dow_restricted,
dom_last,
dom_last_weekday,
dom_nearest_weekday,
dow_last,
dow_nth,
})
}
fn next_after(&self, after: DateTime<Utc>) -> Option<DateTime<Utc>> {
let local_after = after.with_timezone(&Local);
let by_seconds = self.seconds != vec![0];
let step = if by_seconds {
ChronoDuration::seconds(1)
} else {
ChronoDuration::minutes(1)
};
let mut candidate = if by_seconds {
local_after.with_nanosecond(0)? + ChronoDuration::seconds(1)
} else {
local_after.with_second(0)?.with_nanosecond(0)? + ChronoDuration::minutes(1)
};
let max_iterations: u64 = if by_seconds {
2u64 * 366 * 24 * 60 * 60
} else {
6u64 * 366 * 24 * 60
};
for _ in 0..max_iterations {
if self.matches_local(&candidate) {
return Some(candidate.with_timezone(&Utc));
}
candidate += step;
}
None
}
fn matches_local(&self, dt: &DateTime<Local>) -> bool {
if !self.seconds.contains(&dt.second()) {
return false;
}
if !self.minutes.contains(&dt.minute()) {
return false;
}
if !self.hours.contains(&dt.hour()) {
return false;
}
if !self.months.contains(&dt.month()) {
return false;
}
if let Some(years) = &self.years {
let year = dt.year();
if year < 0 || !years.contains(&(year as u32)) {
return false;
}
}
let dom_match = self.dom_matches(dt);
let dow_match = self.dow_matches(dt);
match (self.dom_restricted, self.dow_restricted) {
(true, true) => dom_match || dow_match,
(true, false) => dom_match,
(false, true) => dow_match,
(false, false) => true,
}
}
fn dom_matches(&self, dt: &DateTime<Local>) -> bool {
let dom = dt.day();
if self.dom_last && dom == last_day_of_month(dt.year(), dt.month()) {
return true;
}
if self.dom_last_weekday {
if is_nearest_weekday(dt, last_day_of_month(dt.year(), dt.month())) {
return true;
}
}
if let Some(target) = self.dom_nearest_weekday {
if is_nearest_weekday(dt, target) {
return true;
}
}
self.days_of_month.contains(&dom)
}
fn dow_matches(&self, dt: &DateTime<Local>) -> bool {
let dow = weekday_sun0(dt.weekday());
if let Some(target) = self.dow_last {
if dow == target {
let next_week = *dt + ChronoDuration::days(7);
if next_week.month() != dt.month() {
return true;
}
}
}
if let Some((target, n)) = self.dow_nth {
if dow == target {
let occurrence = (dt.day() - 1) / 7 + 1;
if occurrence == n {
return true;
}
}
}
self.days_of_week.contains(&dow)
}
}
fn weekday_sun0(weekday: Weekday) -> u32 {
weekday.num_days_from_sunday()
}
fn last_day_of_month(year: i32, month: u32) -> u32 {
let (ny, nm) = if month == 12 {
(year + 1, 1)
} else {
(year, month + 1)
};
let first_next = chrono::NaiveDate::from_ymd_opt(ny, nm, 1).expect("valid first-of-month");
(first_next - ChronoDuration::days(1)).day()
}
fn is_nearest_weekday(dt: &DateTime<Local>, target: u32) -> bool {
let last = last_day_of_month(dt.year(), dt.month());
let target = target.min(last);
let target_date = chrono::NaiveDate::from_ymd_opt(dt.year(), dt.month(), target);
let target_date = match target_date {
Some(d) => d,
None => return false,
};
let target_weekday = target_date.weekday();
let resolved_day = match target_weekday {
Weekday::Sat => {
if target > 1 {
target - 1
} else {
target + 2
}
}
Weekday::Sun => {
if target < last {
target + 1
} else {
target - 2
}
}
Weekday::Mon | Weekday::Tue | Weekday::Wed | Weekday::Thu | Weekday::Fri => target,
};
dt.day() == resolved_day
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum FieldKind {
Plain,
Month,
Weekday,
}
fn parse_field(
field: &str,
min: u32,
max: u32,
kind: FieldKind,
) -> std::result::Result<Vec<u32>, ()> {
if field == "?" {
return Ok((min..=max).collect());
}
let mut values: Vec<u32> = Vec::new();
for part in field.split(',') {
if part.is_empty() {
return Err(());
}
parse_field_part(part, min, max, kind, &mut values)?;
}
if values.is_empty() {
return Err(());
}
values.sort_unstable();
values.dedup();
Ok(values)
}
fn parse_dom_field(
field: &str,
dom_last: &mut bool,
dom_last_weekday: &mut bool,
dom_nearest_weekday: &mut Option<u32>,
) -> std::result::Result<Vec<u32>, ()> {
let upper = field.to_ascii_uppercase();
if upper == "L" {
*dom_last = true;
return Ok(Vec::new());
}
if upper == "LW" {
*dom_last_weekday = true;
return Ok(Vec::new());
}
if let Some(stripped) = upper.strip_suffix('W') {
let day: u32 = stripped.parse().map_err(|_| ())?;
if !(1..=31).contains(&day) {
return Err(());
}
*dom_nearest_weekday = Some(day);
return Ok(Vec::new());
}
parse_field(field, 1, 31, FieldKind::Plain)
}
fn parse_dow_field(
field: &str,
dow_last: &mut Option<u32>,
dow_nth: &mut Option<(u32, u32)>,
) -> std::result::Result<Vec<u32>, ()> {
let upper = field.to_ascii_uppercase();
if let Some((wd, nth)) = upper.split_once('#') {
let weekday = parse_weekday_token(wd)?;
let n: u32 = nth.parse().map_err(|_| ())?;
if !(1..=5).contains(&n) {
return Err(());
}
*dow_nth = Some((weekday, n));
return Ok(Vec::new());
}
if let Some(stripped) = upper.strip_suffix('L') {
let weekday = parse_weekday_token(stripped)?;
*dow_last = Some(weekday);
return Ok(Vec::new());
}
if upper == "?" || upper == "*" {
let mut v = parse_field(field, 0, 7, FieldKind::Plain)?;
fold_sunday(&mut v);
return Ok(v);
}
let mut values = parse_field(field, 0, 7, FieldKind::Weekday)?;
fold_sunday(&mut values);
Ok(values)
}
fn fold_sunday(values: &mut Vec<u32>) {
for v in values.iter_mut() {
if *v == 7 {
*v = 0;
}
}
values.sort_unstable();
values.dedup();
}
fn parse_weekday_token(token: &str) -> std::result::Result<u32, ()> {
let upper = token.to_ascii_uppercase();
if let Some(idx) = WEEKDAY_NAMES.iter().position(|name| *name == upper) {
return Ok(idx as u32);
}
let v: u32 = upper.parse().map_err(|_| ())?;
match v {
0..=6 => Ok(v),
7 => Ok(0),
_ => Err(()),
}
}
impl FieldKind {
fn resolve_name(self, token: &str) -> Option<u32> {
let upper = token.to_ascii_uppercase();
match self {
FieldKind::Plain => None,
FieldKind::Month => MONTH_NAMES
.iter()
.position(|name| *name == upper)
.map(|i| (i + 1) as u32),
FieldKind::Weekday => WEEKDAY_NAMES
.iter()
.position(|name| *name == upper)
.map(|i| i as u32),
}
}
}
fn parse_field_part(
part: &str,
min: u32,
max: u32,
kind: FieldKind,
out: &mut Vec<u32>,
) -> std::result::Result<(), ()> {
let (range_spec, step) = match part.split_once('/') {
Some((range_spec, step_str)) => {
let step: u32 = step_str.parse().map_err(|_| ())?;
if step == 0 {
return Err(());
}
(range_spec, Some(step))
}
None => (part, None),
};
let (start, end) = if range_spec == "*" {
(min, max)
} else if let Some((lo, hi)) = range_spec.split_once('-') {
let lo = parse_value_token(lo, kind)?;
let hi = parse_value_token(hi, kind)?;
(lo, hi)
} else {
if step.is_some() {
return Err(());
}
let v = parse_value_token(range_spec, kind)?;
(v, v)
};
if start < min || end > max || start > end {
return Err(());
}
let step = step.unwrap_or(1);
let mut v = start;
while v <= end {
out.push(v);
v += step;
}
Ok(())
}
fn parse_value_token(token: &str, kind: FieldKind) -> std::result::Result<u32, ()> {
match kind {
FieldKind::Weekday => parse_weekday_token(token),
FieldKind::Month => {
if let Some(v) = kind.resolve_name(token) {
return Ok(v);
}
token.parse().map_err(|_| ())
}
FieldKind::Plain => token.parse().map_err(|_| ()),
}
}
impl AgentOs {
pub fn schedule_cron(
&self,
options: CronJobOptions,
) -> std::result::Result<CronJobHandle, ClientError> {
let cron = self.cron();
let now = Utc::now();
let next_run = validate_schedule(&options.schedule, now)?;
let id = options.id.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let overlap = options.overlap.unwrap_or_default();
let manager = Arc::clone(cron);
let vm = self.clone();
let callback_id = id.clone();
let callback: crate::config::ScheduleCallback = Arc::new(move || {
let manager = Arc::clone(&manager);
let vm = vm.clone();
let id = callback_id.clone();
Box::pin(async move {
execute_job(manager, vm, id).await;
})
});
let handle = cron.driver.schedule(ScheduleEntry {
id: id.clone(),
schedule: options.schedule.clone(),
callback,
});
let state = CronJobState {
schedule: options.schedule.clone(),
action: options.action,
overlap,
last_run: parking_lot::Mutex::new(None),
next_run: parking_lot::Mutex::new(next_run),
run_count: std::sync::atomic::AtomicU64::new(0),
running: AtomicBool::new(false),
queued: AtomicBool::new(false),
handle,
};
if let Some((_, old)) = cron.jobs.remove(&id) {
cron.driver.cancel(&old.handle);
}
let _ = cron.jobs.insert(id.clone(), state);
Ok(CronJobHandle {
id,
manager: Arc::clone(cron),
})
}
pub fn list_cron_jobs(&self) -> Vec<CronJobInfo> {
let mut result = Vec::new();
self.cron().jobs.scan(|id, state| {
result.push(CronJobInfo {
id: id.clone(),
schedule: state.schedule.clone(),
action: state.action.clone(),
overlap: state.overlap,
last_run: *state.last_run.lock(),
next_run: *state.next_run.lock(),
run_count: state.run_count.load(Ordering::SeqCst),
running: state.running.load(Ordering::SeqCst),
});
});
result
}
pub fn cancel_cron_job(&self, id: &str) {
self.cron().cancel_job(id);
}
pub fn cron_events(&self) -> broadcast::Receiver<CronEvent> {
self.cron().event_tx.subscribe()
}
}