erikn/jobs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: erikn:02339e1842e162b0299839a76bb721ed8313578f
Branches Tags
erikn:main
..
compare: erikn:1498ce6c15ddc580646aa824acc1869c200602d8
Branches Tags
erikn:main

No commits in common. "02339e1842e162b0299839a76bb721ed8313578f" and "1498ce6c15ddc580646aa824acc1869c200602d8" have entirely different histories.
02339e1842 ... 1498ce6c15

20 changed files with 17 additions and 2093 deletions
Show stats Expand all files Collapse all files

1505
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

8
Cargo.toml
View file

@ -4,10 +4,7 @@ version = "0.1.0"
edition = "2021"
[dependencies]
anyhow = { version = "1.0.96", features = ["backtrace"] }
apalis = { version = "0.6.4", features = ["catch-panic", "docsrs", "document-features", "filter", "layers", "limit", "prometheus", "retry", "sentry", "timeout", "ulid", "uuid"] }
apalis-amqp = "0.4.1"
apalis-sql = { version = "0.6.4", features = ["async-std", "postgres", "sqlite"] }
chrono = { version = "0.4.39", features = ["rkyv-64", "serde"] }
chrono-tz = { version = "0.10.1", features = ["arbitrary", "case-insensitive", "filter-by-regex", "serde"] }
clokwerk = "0.4.0"
@ -16,12 +13,7 @@ cron-job = "0.2.0"
cron_tab = { version = "0.2.8", features = ["all"] }
croner = "2.1.0"
delay_timer = { version = "0.11.6", features = ["full"] }
email_address = "0.2.9"
english-to-cron = "0.1.2"
log = { version = "0.4.26", features = ["kv", "kv_serde", "max_level_debug", "release_max_level_debug", "serde", "std", "sval", "value-bag"] }
serde = { version = "1.0.218", features = ["alloc", "derive", "rc", "serde_derive"] }
smol = "2.0.2"
sqlx = { version = "0.8.3", features = ["runtime-async-std"] }
tokio = { version = "1.43.0", features = ["full", "mio", "test-util", "tracing"] }
tokio-cron-scheduler = { version = "0.13.0", features = ["english", "log", "postgres_storage", "prost", "signal", "tokio-postgres", "tracing-subscriber"] }
tracing = { version = "0.1.41", features = ["async-await", "log", "valuable", "max_level_debug", "release_max_level_debug"] }

143
README.md
View file

@ -1,143 +0,0 @@
# jobs
Trying out different job scheduling and job parsing crates.
## Apalis
In addition to the below, a lot of other `apalis` examples can be found at
<https://github.com/geofmureithi/apalis/tree/main/examples>.
### PostgreSQL
Create database.
```zsh
createdb foobar
```
Run sample program.
```zsh
DATABASE_URL="postgres://${USER}:passwordifany@localhost/foobar" cargo run --release --bin using-crate-apalis-postgres
```
### SQLite
Run sample program.
```zsh
cargo run --release --bin using-crate-apalis-sqlite
```
## Clokwerk
Run sample program.
```zsh
cargo run --release --bin using-crate-clokwerk
```
The `clokwerk` repo contains only this single example
in its readme and does *not* contain any other
direct examples of use at the time of writing this.
## Cron
Run sample program.
```zsh
cargo run --release --bin using-crate-cron
```
The `cron` repo contains only this single example
in its readme and does *not* contain any other
direct examples of use at the time of writing this.
## Cron-job
Run sample program.
```zsh
cargo run --release --bin using-crate-cron-job
```
The `cron-job` repo contains a couple of other
direct examples of use in the readme, but nothing
beyond that at the time of writing this.
## Cron\_tab
Run sample programs.
```zsh
cargo run --release --bin using-crate-cron_tab-sync
```
```zsh
cargo run --release --bin using-crate-cron_tab-async
```
The `cron_tab` repo contains only these two examples
in its readme, and a copy of the two same sample programs
at <https://github.com/tuyentv96/rust-crontab/tree/master/examples>,
and does *not* contain any other direct examples of use
at the time of writing this.
## Croner
Run sample program.
```zsh
cargo run --release --bin using-crate-croner
```
Additional examples at <https://github.com/Hexagon/croner-rust/tree/main/examples>.
## Delay\_timer
Run sample programs.
```zsh
cargo run --release --bin using-crate-delay_timer-internal
```
```zsh
cargo run --release --bin using-crate-delay_timer-in-async-context
```
Additional examples at: <https://github.com/BinChengZhao/delay-timer/tree/master/examples>
## English-to-cron
Run sample program.
```zsh
cargo run --release --bin using-crate-english-to-cron
```
The `english-to-cron` repo contains only this example
in its readme, and a copy of the same sample program
at <https://github.com/kaplanelad/english-to-cron/tree/main/examples>,
and does *not* contain any other direct examples of use
at the time of writing this.
## Tokio-cron-scheduler
Run sample programs.
```zsh
cargo run --release --bin using-crate-tokio-cron-scheduler-simple_job_tokio_in_a_thread
```
```zsh
cargo run --release --bin using-crate-tokio-cron-scheduler-simple_job
```
```zsh
cargo run --release --bin using-crate-tokio-cron-scheduler-postgres_job
```
One additional example at <https://github.com/mvniekerk/tokio-cron-scheduler/tree/main/examples>
although that one (`nats_job.rs`) is not buildable for me I assume, as enabling the
nats related features did not work for me. Enabling nats features in `Cargo.toml`
for me makes Rust unable to build anything.

116
src/apalis_email_service.rs
View file

@ -1,116 +0,0 @@
//! Code from <https://github.com/geofmureithi/apalis/blob/060a7d260cc66714afe9ddc20012a569b00103a2/examples/email-service/src/lib.rs>
use std::{str::FromStr, sync::Arc};
use apalis::prelude::*;
use email_address::EmailAddress;
use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct Email {
pub to: String,
pub subject: String,
pub text: String,
}
pub async fn send_email(job: Email) -> Result<(), Error> {
let validation = EmailAddress::from_str(&job.to);
match validation {
Ok(email) => {
log::info!("Attempting to send email to {}", email.as_str());
Ok(())
}
Err(email_address::Error::InvalidCharacter) => {
log::error!("Killed send email job. Invalid character {}", job.to);
Err(Error::Abort(Arc::new(Box::new(
email_address::Error::InvalidCharacter,
))))
}
Err(e) => Err(Error::Failed(Arc::new(Box::new(e)))),
}
}
pub fn example_good_email() -> Email {
Email {
subject: "Test Subject".to_string(),
to: "example@gmail.com".to_string(),
text: "Some Text".to_string(),
}
}
pub fn example_killed_email() -> Email {
Email {
subject: "Test Subject".to_string(),
to: "example@©.com".to_string(), // killed because it has © which is invalid
text: "Some Text".to_string(),
}
}
pub fn example_retry_able_email() -> Email {
Email {
subject: "Test Subject".to_string(),
to: "example".to_string(),
text: "Some Text".to_string(),
}
}
pub const FORM_HTML: &str = r#"
<!doctype html>
<html>
<head>
<link href="https://unpkg.com/tailwindcss@1.2.0/dist/tailwind.min.css" rel="stylesheet">
<meta credits="https://tailwindcomponents.com/component/basic-contact-form" />
</head>
<body>
<form style="margin: 0 auto;" class="w-full max-w-lg pt-20" action="/" method="post">
<div class="flex flex-wrap -mx-3 mb-6">
<div class="w-full md:w-2/3 px-3 mb-6 md:mb-0">
<label class="block uppercase tracking-wide text-gray-700 text-xs font-bold mb-2" for="to">
To
</label>
<input class="appearance-none block w-full bg-gray-200 text-gray-700 border border-red-500 rounded py-3 px-4 mb-3 leading-tight focus:outline-none focus:bg-white" id="to" type="email" name="to" placeholder="test@example.com">
<p class="text-red-500 text-xs italic">Please fill out this field.</p>
</div>
</div>
<div class="flex flex-wrap -mx-3 mb-6">
<div class="w-full px-3">
<label class="block uppercase tracking-wide text-gray-700 text-xs font-bold mb-2" for="subject">
Subject
</label>
<input class="appearance-none block w-full bg-gray-200 text-gray-700 border border-gray-200 rounded py-3 px-4 mb-3 leading-tight focus:outline-none focus:bg-white focus:border-gray-500" id="subject" type="text" name="subject">
<p class="text-gray-600 text-xs italic">Some tips - as long as needed</p>
</div>
</div>
<div class="flex flex-wrap -mx-3 mb-6">
<div class="w-full px-3">
<label class="block uppercase tracking-wide text-gray-700 text-xs font-bold mb-2" for="text">
Message
</label>
<textarea class=" no-resize appearance-none block w-full bg-gray-200 text-gray-700 border border-gray-200 rounded py-3 px-4 mb-3 leading-tight focus:outline-none focus:bg-white focus:border-gray-500 h-48 resize-none" id="text" name="text" ></textarea>
</div>
</div>
<div class="md:flex md:items-center">
<div class="md:w-1/3">
<button class="shadow bg-teal-400 hover:bg-teal-400 focus:shadow-outline focus:outline-none text-white font-bold py-2 px-4 rounded" type="submit">
Send
</button>
</div>
<div class="md:w-2/3"></div>
</div>
</form>
</body>
</html>
"#;
#[derive(Debug)]
pub enum EmailError {
NoStorage,
SomeError(&'static str),
}
impl std::fmt::Display for EmailError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{self:?}")
}
}

66
src/bin/using-crate-apalis-postgres.rs
View file

@ -1,66 +0,0 @@
//! Code from <https://github.com/geofmureithi/apalis/blob/060a7d260cc66714afe9ddc20012a569b00103a2/examples/postgres/src/main.rs>
use anyhow::Result;
use apalis::layers::retry::RetryPolicy;
use apalis::prelude::*;
use apalis_sql::{
postgres::{PgListen, PgPool, PostgresStorage},
Config,
};
use jobs::apalis_email_service::{send_email, Email};
use tracing::{debug, info};
async fn produce_jobs(storage: &mut PostgresStorage<Email>) -> Result<()> {
for index in 0..10 {
storage
.push(Email {
to: format!("test{}@example.com", index),
text: "Test background job from apalis".to_string(),
subject: "Background email job".to_string(),
})
.await?;
}
// The sql way
tracing::info!("You can also add jobs via sql query, run this: \n Select apalis.push_job('apalis::Email', json_build_object('subject', 'Test apalis', 'to', 'test1@example.com', 'text', 'Lorem Ipsum'));");
Ok(())
}
#[tokio::main]
async fn main() -> Result<()> {
std::env::set_var("RUST_LOG", "debug,sqlx::query=error");
tracing_subscriber::fmt::init();
let database_url = std::env::var("DATABASE_URL").expect("Must specify path to db");
let pool = PgPool::connect(&database_url).await?;
PostgresStorage::setup(&pool)
.await
.expect("unable to run migrations for postgres");
let mut pg = PostgresStorage::new_with_config(pool.clone(), Config::new("apalis::Email"));
produce_jobs(&mut pg).await?;
let mut listener = PgListen::new(pool).await?;
listener.subscribe_with(&mut pg);
tokio::spawn(async move {
listener.listen().await.unwrap();
});
Monitor::new()
.register({
WorkerBuilder::new("tasty-orange")
.retry(RetryPolicy::retries(5))
.enable_tracing()
.backend(pg)
.build_fn(send_email)
})
.on_event(|e| debug!("{e}"))
.run_with_signal(async {
tokio::signal::ctrl_c().await?;
info!("Shutting down the system");
Ok(())
})
.await?;
Ok(())
}

88
src/bin/using-crate-apalis-sqlite.rs
View file

@ -1,88 +0,0 @@
//! Code from:
//! - <https://github.com/geofmureithi/apalis/blob/060a7d260cc66714afe9ddc20012a569b00103a2/examples/sqlite/src/job.rs>
//! - <https://github.com/geofmureithi/apalis/blob/060a7d260cc66714afe9ddc20012a569b00103a2/examples/sqlite/src/main.rs>
use anyhow::Result;
use apalis::prelude::*;
use apalis_sql::sqlite::SqliteStorage;
use chrono::Utc;
use jobs::apalis_email_service::{send_email, Email};
use serde::{Deserialize, Serialize};
use sqlx::SqlitePool;
#[derive(Debug, Deserialize, Serialize)]
pub struct Notification {
pub to: String,
pub text: String,
}
pub async fn notify(job: Notification) {
tracing::info!("Attempting to send notification to {}", job.to);
}
async fn produce_emails(storage: &SqliteStorage<Email>) -> Result<()> {
let mut storage = storage.clone();
for i in 0..1 {
storage
.schedule(
Email {
to: format!("test{i}@example.com"),
text: "Test background job from apalis".to_string(),
subject: "Background email job".to_string(),
},
(Utc::now() + chrono::Duration::seconds(4)).timestamp(),
)
.await?;
}
Ok(())
}
async fn produce_notifications(storage: &SqliteStorage<Notification>) -> Result<()> {
let mut storage = storage.clone();
for i in 0..20 {
storage
.push(Notification {
to: format!("notify:{i}@example.com"),
text: "Test background job from apalis".to_string(),
})
.await?;
}
Ok(())
}
#[tokio::main]
async fn main() -> Result<()> {
std::env::set_var("RUST_LOG", "debug,sqlx::query=info");
tracing_subscriber::fmt::init();
let pool = SqlitePool::connect("sqlite::memory:").await?;
// Do migrations: Mainly for "sqlite::memory:"
SqliteStorage::setup(&pool)
.await
.expect("unable to run migrations for sqlite");
let email_storage: SqliteStorage<Email> = SqliteStorage::new(pool.clone());
produce_emails(&email_storage).await?;
let notification_storage: SqliteStorage<Notification> = SqliteStorage::new(pool);
produce_notifications(&notification_storage).await?;
Monitor::new()
.register({
WorkerBuilder::new("tasty-banana")
.enable_tracing()
.backend(email_storage)
.build_fn(send_email)
})
.register({
WorkerBuilder::new("tasty-mango")
// .enable_tracing()
.backend(notification_storage)
.build_fn(notify)
})
.run()
.await?;
Ok(())
}

40
src/bin/using-crate-clokwerk.rs
View file

@ -1,40 +0,0 @@
//! Code from <https://github.com/mdsherry/clokwerk/blob/f8180dfe64a98d39a5cded998998a3df8809b92c/README.md>
// Scheduler, and trait for .seconds(), .minutes(), etc.
use clokwerk::{Job, Scheduler, TimeUnits};
// Import week days and WeekDay
use clokwerk::Interval::*;
use std::thread;
use std::time::Duration;
fn main() {
// Create a new scheduler
//let mut scheduler = Scheduler::new();
// or a scheduler with a given timezone
let mut scheduler = Scheduler::with_tz(chrono::Utc);
// Add some tasks to it
scheduler
.every(10.minutes())
.plus(30.seconds())
.run(|| println!("Periodic task"));
scheduler
.every(1.day())
.at("3:20 pm")
.run(|| println!("Daily task"));
scheduler
.every(Tuesday)
.at("14:20:17")
.and_every(Thursday)
.at("15:00")
.run(|| println!("Biweekly task"));
// Manually run the scheduler in an event loop
for _ in 1..10 {
scheduler.run_pending();
thread::sleep(Duration::from_millis(10));
}
// Or run it in a background thread
let thread_handle = scheduler.watch_thread(Duration::from_millis(100));
// The scheduler stops when `thread_handle` is dropped, or `stop` is called
thread_handle.stop();
}

2
src/bin/using-crate-cron.rs
View file

@ -1,5 +1,3 @@
//! Code from <https://github.com/zslayton/cron/blob/956beaf3cfe32091dc7a0b371340b59ae5e1a860/README.md>
use chrono::Utc;
use cron::Schedule;
use std::str::FromStr;

4
src/bin/using-crate-cron-job.rs → src/bin/using-crate-cron_job.rs
View file

@ -1,5 +1,3 @@
//! Based on code from <https://github.com/nambrosini/cron-job/blob/e51067cb2395994cb8643204152a1e7dfc161aa5/README.md>
use cron_job::{CronJob, Job};
fn main() {
@ -14,7 +12,7 @@ fn main() {
// Say hello every second
cron.new_job("* * * * * *", hello_job);
// Start jobs
cron.start().expect("Failed to start jobs");
cron.start().expect("Failed start jobs.");
}
// The function to be executed every second.
fn run_every_second() {

2
src/bin/using-crate-cron_tab-async.rs
View file

@ -1,5 +1,3 @@
//! Based on code from <https://github.com/tuyentv96/rust-crontab/blob/01a266e8e1c7f6ee86b3d1010b8818ffc4db6518/README.md>
use std::sync::Arc;
use chrono::{FixedOffset, Local, TimeZone};

2
src/bin/using-crate-cron_tab-sync.rs
View file

@ -1,5 +1,3 @@
//! Based on code from <https://github.com/tuyentv96/rust-crontab/blob/01a266e8e1c7f6ee86b3d1010b8818ffc4db6518/README.md>
use chrono::{FixedOffset, Local, TimeZone};
use cron_tab;

26
src/bin/using-crate-croner.rs
View file

@ -1,22 +1,18 @@
//! Code from <https://github.com/Hexagon/croner-rust/blob/4da136d3363ff7d99f9bdb211f7c8c852449de6d/README.md>
use chrono::Local;
use chrono::Utc;
use croner::Cron;
fn main() {
// Parse cron expression for Fridays in December
let cron = Cron::new("0 0 0 31 12 FRI")
// Include seconds in pattern
.with_seconds_optional()
// Ensure both day of month and day of week conditions are met
.with_dom_and_dow()
// Parse a cron expression to find the next occurrence at 00:00 on Friday
let cron = Cron::new("0 0 * * FRI")
.parse()
.expect("Couldn't parse cron string");
.expect("Successful parsing");
let time = Local::now();
// Get the next occurrence from the current time, excluding the current time
let next = cron.find_next_occurrence(&Utc::now(), false).unwrap();
println!("Finding the next 5 New Year's Eves on a Friday:");
for time in cron.iter_from(time).take(5) {
println!("{}", time);
}
println!(
"Pattern \"{}\" will match next at {}",
cron.pattern.to_string(),
next
);
}

50
src/bin/using-crate-delay_timer-in-async-context.rs
View file

@ -1,50 +0,0 @@
//! Based on code from <https://github.com/BinChengZhao/delay-timer/blob/560c92f9b12c56a6729bc46cdaab52c9557a84e7/README.md>
use delay_timer::prelude::*;
use anyhow::Result;
use smol::Timer;
use std::time::Duration;
#[tokio::main]
async fn main() -> Result<()> {
// In addition to the mixed (smol & tokio) runtime
// You can also share a tokio runtime with delayTimer, please see api `DelayTimerBuilder::tokio_runtime` for details.
// Build an DelayTimer that uses the default configuration of the Smol runtime internally.
let delay_timer = DelayTimerBuilder::default().build();
// Develop a print job that runs in an asynchronous cycle.
let task_instance_chain = delay_timer.insert_task(build_task_async_print()?)?;
// Get the running instance of task 1.
let task_instance = task_instance_chain.next_with_async_wait().await?;
// Cancel running task instances.
task_instance.cancel_with_async_wait().await?;
// Remove task which id is 1.
delay_timer.remove_task(1)?;
// No new tasks are accepted; running tasks are not affected.
Ok(delay_timer.stop_delay_timer()?)
}
fn build_task_async_print() -> Result<Task, TaskError> {
let mut task_builder = TaskBuilder::default();
let body = || async {
println!("create_async_fn_body!");
Timer::after(Duration::from_secs(3)).await;
println!("create_async_fn_body:i'success");
};
task_builder
.set_task_id(1)
.set_frequency_repeated_by_cron_str("@secondly")
.set_maximum_parallel_runnable_num(2)
.spawn_async_routine(body)
}

47
src/bin/using-crate-delay_timer-internal.rs
View file

@ -1,47 +0,0 @@
//! Code from <https://github.com/BinChengZhao/delay-timer/blob/560c92f9b12c56a6729bc46cdaab52c9557a84e7/README.md>
use anyhow::Result;
use delay_timer::prelude::*;
use smol::Timer;
use std::time::Duration;
fn main() -> Result<()> {
// Build an DelayTimer that uses the default configuration of the Smol runtime internally.
let delay_timer = DelayTimerBuilder::default().build();
// Develop a print job that runs in an asynchronous cycle.
// A chain of task instances.
let task_instance_chain = delay_timer.insert_task(build_task_async_print()?)?;
// Get the running instance of task 1.
let task_instance = task_instance_chain.next_with_wait()?;
// Cancel running task instances.
task_instance.cancel_with_wait()?;
// Remove task which id is 1.
delay_timer.remove_task(1)?;
// No new tasks are accepted; running tasks are not affected.
delay_timer.stop_delay_timer()?;
Ok(())
}
fn build_task_async_print() -> Result<Task, TaskError> {
let mut task_builder = TaskBuilder::default();
let body = || async {
println!("create_async_fn_body!");
Timer::after(Duration::from_secs(3)).await;
println!("create_async_fn_body:i'success");
};
task_builder
.set_task_id(1)
.set_frequency_repeated_by_cron_str("@secondly")
.set_maximum_parallel_runnable_num(2)
.spawn_async_routine(body)
}

2
src/bin/using-crate-english-to-cron.rs
View file

@ -1,5 +1,3 @@
//! Code from <https://github.com/kaplanelad/english-to-cron/blob/90fe66631e51e9ad7fb631cd55434b7ab8f990f7/examples/run.rs>
fn main() {
let texts = vec![
"every 15 seconds",

2
src/bin/using-crate-tokio-cron-scheduler-postgres_job.rs
View file

@ -1,5 +1,3 @@
//! Code based on <https://github.com/mvniekerk/tokio-cron-scheduler/blob/6c568541022317cc07905ffd25305f0e6e2cfc74/examples/postgres_job.rs>
use jobs::tcs_helpers::{run_example, stop_example};
use tokio_cron_scheduler::{
JobScheduler, PostgresMetadataStore, PostgresNotificationStore, SimpleJobCode,

2
src/bin/using-crate-tokio-cron-scheduler-simple_job.rs
View file

@ -1,5 +1,3 @@
//! Code based on <https://github.com/mvniekerk/tokio-cron-scheduler/blob/6c568541022317cc07905ffd25305f0e6e2cfc74/examples/simple_job.rs>
use jobs::tcs_helpers::{run_example, stop_example};
use tokio_cron_scheduler::JobScheduler;
use tracing::Level;

2
src/bin/using-crate-tokio-cron-scheduler-simple_job_tokio_in_a_thread.rs
View file

@ -1,5 +1,3 @@
//! Code based on <https://github.com/mvniekerk/tokio-cron-scheduler/blob/6c568541022317cc07905ffd25305f0e6e2cfc74/examples/simple_job_tokio_in_a_thread.rs>
use jobs::tcs_helpers::{run_example, stop_example};
use std::error::Error;
use tokio_cron_scheduler::JobScheduler;

1
src/lib.rs
View file

@ -1,2 +1 @@
pub mod apalis_email_service;
pub mod tcs_helpers;

2
src/tcs_helpers.rs
View file

@ -1,5 +1,3 @@
//! Code from <https://github.com/mvniekerk/tokio-cron-scheduler/blob/6c568541022317cc07905ffd25305f0e6e2cfc74/examples/lib.rs>
use chrono::Utc;
use std::time::Duration;
use tokio_cron_scheduler::{Job, JobBuilder, JobScheduler, JobSchedulerError};