3 Commits
Author SHA1 Message Date
lumither a123ff9f3f fix(scheduler) fn type 2024-11-17 21:06:36 -08:00
lumither d98aa3f745 fix(scheduler) remove unnecessary field 2024-10-30 23:07:22 -07:00
lumither 6f45a140ba ref(scheduler) move module 2024-10-30 22:54:39 -07:00
15 changed files with 124 additions and 30 deletions

No files matched your search

+1
View File
@@ -20,3 +20,4 @@ uuid = { version = "1.10.0", features = ["v4", "fast-rng", "serde"] }
futures = { version = "0.3.30", features = ["executor"] }
sha2 = "0.10.8"
cron = "0.12.1"
git2 = "0.19.0"
@@ -0,0 +1,24 @@
use dirs::home_dir;
use std::env;
pub fn task() {
// get dir
let git_work_dir_path = &env::var("GIT_WORK_DIR").unwrap_or_else(|e| {
if let Ok(default_log_path) = env::var("WORK_DIR") {
if default_log_path.starts_with('~') {
let home_path = home_dir().unwrap_or_else(|| panic!("Cannot find home directory"));
home_path
.join(default_log_path.strip_prefix("~/").unwrap())
.into_os_string()
.into_string()
.expect(
"[fatal] failed to parse env `WORK_DIR`, try to use absolute path instead",
)
} else {
default_log_path
}
} else {
panic!("[fatal] failed to parse env `GIT_WORK_DIR`: {}", e);
}
});
}
+1
View File
@@ -0,0 +1 @@
pub mod check_update;
+1 -1
View File
@@ -1,5 +1,5 @@
mod api;
mod scheduler;
pub mod cron;
use crate::blog::api::{
get_page_count::get_page_count, get_post::get_post, get_post_list::get_post_list,
+35 -3
View File
@@ -1,11 +1,17 @@
use std::env;
use crate::logger;
use std::{
backtrace::{Backtrace, BacktraceStatus},
env, panic,
panic::PanicHookInfo,
};
use tracing::{error, event, span, Level};
use tracing_appender::non_blocking::WorkerGuard;
pub fn logger_init() -> (WorkerGuard, WorkerGuard) {
logger::init()
let ret = logger::init();
panic::set_hook(Box::new(panic_hook));
ret
}
pub fn load_env() {
@@ -22,3 +28,29 @@ pub fn load_env() {
println!("[info] no .env file referred, starting up")
}
}
fn panic_hook(panic_info: &PanicHookInfo) {
let payload = panic_info.payload();
// panic_info.
let payload = if let Some(s) = payload.downcast_ref::<&str>() {
Some(&**s)
} else {
payload.downcast_ref::<String>().map(|s| s.as_str())
};
// let location = panic_info.location().map(|l| l.to_string());
// let backtrace = Backtrace::capture();
// let note = (backtrace.status() == BacktraceStatus::Disabled)
// .then_some("run with RUST_BACKTRACE=1 environment variable to display a backtrace");
// (Some(backtrace), note)
// };
//
// panic_info.
if let Some(location) = panic_info.location() {
eprintln!("err filename: {}", location.file())
}
event!(target: "test", Level::ERROR, "panic hook: {:?}", panic_info);
}
+9 -1
View File
@@ -1,6 +1,7 @@
mod blog;
mod init;
mod logger;
mod scheduler;
mod server;
#[tokio::main]
@@ -8,5 +9,12 @@ async fn main() {
init::load_env();
let _guards = init::logger_init();
server::start().await;
// panic!("test");
match server::start().await {
Ok(_) => {}
Err(e) => {
panic!("server start failed: {}", e);
}
};
}
@@ -1,4 +1,4 @@
use crate::blog::{
use crate::{
scheduler::scheduler_error::SchedulerError,
scheduler::scheduler_error::SchedulerError::IdAlreadyRunning,
scheduler::scheduler_error::SchedulerError::IdNotFound, scheduler::sleeper::SkippableSleeper,
@@ -7,6 +7,7 @@ use crate::blog::{
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::RwLock;
use std::{collections::HashMap, sync::Arc, thread};
use tracing::log::error;
use uuid::Uuid;
pub mod scheduler_error;
@@ -23,7 +24,6 @@ struct TaskStatus {
pub struct Scheduler {
tasks: Arc<RwLock<HashMap<Uuid, Arc<Box<dyn Task>>>>>,
running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
stopped: Arc<RwLock<Vec<Uuid>>>,
}
impl Scheduler {
@@ -40,7 +40,6 @@ impl Scheduler {
fn run_task(
running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
stopped: Arc<RwLock<Vec<Uuid>>>,
id: Uuid,
task: Arc<Box<dyn Task>>,
) -> Result<(), SchedulerError> {
@@ -69,7 +68,6 @@ impl Scheduler {
let mut running_lock_guard = running.write()?;
running_lock_guard.remove(&id);
}
stopped.write()?.push(id);
return Ok(());
} else {
let sleeper = Arc::from(SkippableSleeper::from(next_execution));
@@ -89,7 +87,7 @@ impl Scheduler {
}
}
} else {
stopped.write()?.push(id);
error!("unexpected error ");
break;
}
}
@@ -100,7 +98,7 @@ impl Scheduler {
pub fn run(&self, id: &Uuid) -> Result<(), SchedulerError> {
let id = id.to_owned();
if let Some(task) = self.tasks.read()?.get(&id) {
Self::run_task(self.running.clone(), self.stopped.clone(), id, task.clone())
Self::run_task(self.running.clone(), id, task.clone())
} else {
Err(IdNotFound)
}
@@ -110,9 +108,8 @@ impl Scheduler {
let id = id.to_owned();
if let Some(task) = self.tasks.read()?.get(&id) {
let running_clone = self.running.clone();
let stopped_clone = self.stopped.clone();
let task_clone = task.clone();
thread::spawn(move || Self::run_task(running_clone, stopped_clone, id, task_clone));
thread::spawn(move || Self::run_task(running_clone, id, task_clone));
Ok(())
} else {
Err(IdNotFound)
@@ -159,12 +156,16 @@ impl Scheduler {
self.tasks.write()?.insert(id, Arc::new(task));
Ok(id)
}
pub fn new() -> Self {
Self::default()
}
}
#[cfg(test)]
mod tests {
use crate::blog::scheduler::tasks::basic::BasicTask;
use crate::blog::scheduler::Scheduler;
use crate::scheduler::tasks::basic::BasicTask;
use crate::scheduler::Scheduler;
use std::error::Error;
use std::thread;
use std::time::Duration;
@@ -174,11 +175,16 @@ mod tests {
let scheduler = Scheduler::default();
let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
let id = scheduler.insert(Box::new(task))?;
scheduler.run_non_blocking(&id)?;
thread::sleep(Duration::from_secs(4));
scheduler.stop(&id)?;
// test termination
thread::sleep(Duration::from_secs(5));
for i in 1..=2 {
println!("Loop {} start", i);
println!("Run {}", i);
scheduler.run_non_blocking(&id)?;
thread::sleep(Duration::from_secs(4));
scheduler.stop(&id)?;
println!("Stop {}", i);
thread::sleep(Duration::from_secs(2));
println!("Loop {} finished", i);
}
Ok(())
}
}
@@ -1,4 +1,4 @@
use crate::blog::scheduler::scheduler_error::SchedulerError::MutexPoison;
use crate::scheduler::scheduler_error::SchedulerError::MutexPoison;
use std::error::Error;
use std::fmt::{Debug, Display, Formatter};
use std::sync::PoisonError;
@@ -54,7 +54,7 @@ impl From<DateTime<Utc>> for SkippableSleeper {
#[cfg(test)]
mod test {
use crate::blog::scheduler::sleeper::SkippableSleeper;
use crate::scheduler::sleeper::SkippableSleeper;
use std::thread;
use std::time::Duration;
@@ -1,5 +1,5 @@
use crate::blog::scheduler::task_func::TaskFunc;
use crate::blog::scheduler::tasks::Task;
use crate::scheduler::task_func::TaskFunc;
use crate::scheduler::tasks::Task;
use chrono::{DateTime, Utc};
use cron::Schedule;
use std::error::Error;
@@ -11,12 +11,19 @@ pub struct BasicTask {
}
impl BasicTask {
pub fn new(func: fn(), cron_expr: &str) -> Result<Self, Box<dyn Error>> {
pub fn new<F>(func: F, cron_expr: &str) -> Result<Self, Box<dyn Error>>
where
F: TaskFunc,
{
Ok(BasicTask {
schedule: Schedule::from_str(cron_expr)?,
func: Box::new(func),
})
}
pub fn to_task(self) -> Box<dyn Task> {
Box::new(self)
}
}
impl Task for BasicTask {
@@ -1,4 +1,3 @@
use crate::blog::scheduler::task_func::TaskFunc;
use chrono::{DateTime, Utc};
pub mod basic;
+17 -1
View File
@@ -1,9 +1,12 @@
use std::env;
use std::error::Error;
use std::num::ParseIntError;
use std::process::exit;
use std::str::FromStr;
use crate::blog;
use crate::scheduler::tasks::basic::BasicTask;
use crate::scheduler::Scheduler;
use axum::Router;
use sqlx::{
postgres::{PgConnectOptions, PgPoolOptions},
@@ -11,7 +14,7 @@ use sqlx::{
};
use tracing::{error, info, warn};
pub async fn start() {
pub async fn start() -> Result<(), Box<dyn Error>> {
let port: u32 = match env::var("BACKEND_PORT") {
Ok(value) => value.parse().unwrap_or_else(|e: ParseIntError| {
warn!("failed to parse BACKEND_PORT: {}", e.to_string());
@@ -35,6 +38,18 @@ pub async fn start() {
}
};
// cron task
let scheduler = Scheduler::new();
let test_str = "hello world".to_string();
// scheduler.insert(
// BasicTask::new(
// move || blog::cron::check_update::task(test_str.clone()),
// "* * * * * *",
// )?
// .to_task(),
// )?;
// http server
let app = Router::new()
.nest("blog", blog::get_router())
.with_state(db_pool);
@@ -50,6 +65,7 @@ pub async fn start() {
error!("failed to start axum: {}", e.to_string());
exit(1);
});
Ok(())
}
fn parse_db_uri() -> String {
+1 -1
View File
@@ -9,6 +9,6 @@ chrono = { version = "0.4.38", features = ["serde", "alloc"] }
serde = { version = "1.0.209", features = ["derive"] }
serde_json = "1.0.128"
regex = "1.10.4"
yaml-rust2 = "0.8.1"
yaml-rust2 = "0.9.0"
sha2 = "0.10.8"
base64 = "0.22.1"
+2 -2
View File
@@ -60,7 +60,7 @@ tags: [ "test1", "test2" ]
#[test]
fn test_front_matter_parser() {
parse_front_matter(
dbg!(parse_front_matter(
r##"---
title: test_title
date: 0000-00-00+0000
@@ -71,7 +71,7 @@ summary: summary
tags: [ "test1", "test2" ]
---
"##,
)
))
.expect("Failed to parse front matter");
}
}