effectum 0.7.0

An embeddable task queue based on SQLite
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

//! This is a simple example of using the queue to schedule a number of jobs that use a JSON
//! payload and access the context.

use std::{
    path::Path,
    sync::{Arc, Mutex},
    time::Duration,
};

use ahash::HashMap;
use effectum::{Job, JobRunner, Queue, RecurringJobSchedule, RunningJob};
use time::OffsetDateTime;
use tracing_subscriber::{layer::SubscriberExt, EnvFilter};

#[derive(Debug)]
struct JobContext {
    start: OffsetDateTime,
    job_times: Mutex<HashMap<String, Vec<i64>>>,
}

async fn print_message_job(job: RunningJob, context: Arc<JobContext>) -> Result<(), eyre::Report> {
    let message: String = job.json_payload()?;
    let seconds = OffsetDateTime::now_utc() - context.start;
    println!(
        "Running job: {message} after {}s",
        seconds.as_seconds_f32().round()
    );

    context
        .job_times
        .lock()
        .unwrap()
        .entry(message)
        .or_default()
        .push(seconds.as_seconds_f32() as i64);

    Ok(())
}

const PRINT_MESSAGE_JOB: &str = "print_message";

#[tokio::main(flavor = "current_thread")]
pub async fn main() -> Result<(), eyre::Report> {
    color_eyre::install()?;

    let tree = tracing_tree::HierarchicalLayer::new(2)
        .with_targets(true)
        .with_bracketed_fields(true);
    let env_filter = EnvFilter::try_from_env("LOG").unwrap_or_else(|_| EnvFilter::new("info"));

    let subscriber = tracing_subscriber::Registry::default()
        .with(env_filter)
        .with(tree);
    tracing::subscriber::set_global_default(subscriber).unwrap();

    let queue = Queue::new(Path::new("effectum-examples.db")).await?;

    let context = Arc::new(JobContext {
        start: OffsetDateTime::now_utc(),
        job_times: Mutex::new(HashMap::default()),
    });

    let worker = effectum::Worker::builder(&queue, context.clone())
        .jobs(vec![JobRunner::builder(
            PRINT_MESSAGE_JOB,
            print_message_job,
        )
        .build()])
        .min_concurrency(3)
        .max_concurrency(5)
        .build()
        .await?;

    let times = [2, 3, 5];
    for time in &times {
        let job_name = format!("job_{time}");
        let schedule = RecurringJobSchedule::RepeatEvery {
            interval: Duration::from_secs(*time),
        };

        let job = Job::builder(PRINT_MESSAGE_JOB)
            .json_payload(&serde_json::json!(job_name.clone()))?
            .build();
        queue
            .upsert_recurring_job(job_name, schedule, job, false)
            .await?;
    }

    // Wait for the jobs to run.
    tokio::time::sleep(Duration::from_secs(15)).await;

    drop(worker);

    for time in &times {
        let key = format!("job_{time}");
        let times = context.job_times.lock().unwrap().remove(&key).unwrap();
        println!("{key} ran at: {times:?}");
    }

    Ok(())
}