kojin 0.4.0

Async distributed task queue for Rust — like Celery, BullMQ, Sidekiq, but for Rust
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

//! Observability example — MetricsMiddleware + TracingMiddleware + RateLimitMiddleware.
//!
//! Run with:
//!   cargo run --example observability --features rate-limit

use std::num::NonZeroU32;
use std::time::Duration;

use async_trait::async_trait;
use kojin::{
    Broker, KojinBuilder, MemoryBroker, MetricsMiddleware, RateLimitMiddleware, Task, TaskContext,
    TaskMessage, TaskResult, TracingMiddleware,
};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct PingTask {
    seq: u32,
}

#[async_trait]
impl Task for PingTask {
    const NAME: &'static str = "ping";
    type Output = String;

    async fn run(&self, _ctx: &TaskContext) -> TaskResult<Self::Output> {
        println!("  ping #{}", self.seq);
        Ok(format!("pong #{}", self.seq))
    }
}

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();

    let broker = MemoryBroker::new();

    // Enqueue a batch of tasks
    for i in 1..=10 {
        let msg = TaskMessage::new(
            "ping",
            "default",
            serde_json::to_value(&PingTask { seq: i }).unwrap(),
        );
        broker.enqueue(msg).await.unwrap();
    }

    // MetricsMiddleware exposes counters you can query after processing
    let metrics = MetricsMiddleware::new();

    let worker = KojinBuilder::new(broker)
        .register_task::<PingTask>()
        // Middleware stack — order matters (outermost first)
        .middleware(TracingMiddleware)
        .middleware(metrics.clone())
        .middleware(RateLimitMiddleware::per_second(NonZeroU32::new(5).unwrap()))
        .concurrency(4)
        .queues(vec!["default".to_string()])
        .build();

    let cancel = worker.cancel_token();

    // Auto-shutdown after tasks drain
    let m = metrics.clone();
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_millis(500)).await;
            if m.tasks_started() >= 10
                && m.tasks_started() == m.tasks_succeeded() + m.tasks_failed()
            {
                break;
            }
        }
        cancel.cancel();
    });

    println!("Starting worker with observability middleware...\n");
    worker.run().await;

    println!("\n=== Metrics ===");
    println!("  started:   {}", metrics.tasks_started());
    println!("  succeeded: {}", metrics.tasks_succeeded());
    println!("  failed:    {}", metrics.tasks_failed());
}