kojin

Async distributed task queue for Rust — the equivalent of Celery / Dramatiq (Python), BullMQ (Node.js), Sidekiq (Ruby), and Machinery (Go).

Features

• Async-first — built on Tokio, designed for async/await from the ground up
• #[kojin::task] — proc-macro to define tasks from plain async functions
• Pluggable broker — trait-based broker abstraction (Redis included, bring your own)
• Middleware — composable pre/post-execution hooks (tracing, metrics, custom)
• Graceful shutdown — cooperative cancellation via CancellationToken
• Weighted queues — prioritize work across multiple queues
• Configurable retries — per-task retry limits with backoff strategies

Quick Start

Add to your Cargo.toml:

[dependencies]
kojin = "0.1"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
async-trait = "0.1"

Define a task, enqueue it, and run a worker:

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

#[derive(Debug, Serialize, Deserialize)]
struct SendEmail {
    to: String,
    subject: String,
}

#[async_trait]
impl Task for SendEmail {
    const NAME: &'static str = "send_email";
    const QUEUE: &'static str = "emails";
    const MAX_RETRIES: u32 = 3;

    type Output = String;

    async fn run(&self, _ctx: &TaskContext) -> TaskResult<Self::Output> {
        println!("Sending email to {}", self.to);
        Ok(format!("Email sent to {}", self.to))
    }
}

#[tokio::main]
async fn main() {
    let broker = MemoryBroker::new();

    // Enqueue a task
    let msg = TaskMessage::new(
        "send_email",
        "emails",
        serde_json::to_value(&SendEmail {
            to: "user@example.com".into(),
            subject: "Hello!".into(),
        }).unwrap(),
    );
    broker.enqueue(msg).await.unwrap();

    // Build and run worker
    let worker = KojinBuilder::new(broker)
        .register_task::<SendEmail>()
        .concurrency(4)
        .queues(vec!["emails".into()])
        .build();

    worker.run().await;
}

Crate Architecture

License

Licensed under either of Apache License, Version 2.0 or MIT License at your option.