chainmq 1.3.2

A Redis-backed, type-safe job queue for Rust. Provides job registration and execution, delayed jobs, retries with backoff, and scalable workers.
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

use chainmq::{
    AppContext, Job, JobContext, JobOptions, JobRegistry, Result, WorkerBuilder, async_trait,
};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

#[derive(Serialize, Deserialize)]
struct FlakyJob {
    fail_times: u32,
}

#[async_trait]
impl Job for FlakyJob {
    async fn perform(&self, _ctx: &JobContext) -> Result<()> {
        static ATTEMPTS: AtomicU32 = AtomicU32::new(0);
        let n = ATTEMPTS.fetch_add(1, Ordering::SeqCst) + 1;
        println!("[FlakyJob] attempt {}", n);
        if n <= self.fail_times {
            Err(anyhow::anyhow!("simulated failure").into())
        } else {
            Ok(())
        }
    }
    fn name() -> &'static str {
        "FlakyJob"
    }
}

#[derive(Clone, Default)]
struct AppState;
impl AppContext for AppState {
    fn clone_context(&self) -> Arc<dyn AppContext> {
        Arc::new(self.clone())
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::try_init().ok();
    let app = Arc::new(AppState::default());
    let mut registry = JobRegistry::new();
    registry.register::<FlakyJob>();

    let mut worker = WorkerBuilder::new_with_redis_uri("redis://localhost:6370", registry)
        .with_app_context(app)
        .with_queue_name("default")
        .with_concurrency(1)
        .spawn()
        .await?;
    worker.start().await?;

    // Enqueue a job that fails twice, then succeeds. Backoff and attempts configured.
    let queue = chainmq::Queue::new(chainmq::QueueOptions::default()).await?;
    let job = FlakyJob { fail_times: 2 };
    let opts = JobOptions {
        attempts: 3,
        ..Default::default()
    };
    let id = queue.enqueue_with_options(job, opts).await?;
    println!("[FlakyJob] enqueued id={}", id);

    Ok(())
}