ishikari 0.1.1

Atomic, transaction-safe job queueing for Rust applications. Backed by PostgreSQL. Features include reliable background job execution, queue management, retry mechanisms, and flexible backoff strategies.
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

mod common;

use crate::common::AppState;
use ishikari::{Engine, Postgres, Queue};
use sqlx::PgPool;
use std::{sync::Arc, time::Duration};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Setup tracing
    tracing_subscriber::fmt::init();

    // Setup the database
    let database_url = std::env::var("DATABASE_URL").unwrap();
    let pool = PgPool::connect(&database_url).await.unwrap();

    // Run the migrations
    // TODO: migrations should be generated and placed in the user's application
    sqlx::migrate!("../migrations").run(&pool).await.unwrap();

    // Clear the jobs table. This is only for demonstration purposes
    sqlx::query("TRUNCATE jobs RESTART IDENTITY")
        .execute(&pool)
        .await?;

    // Set up some application state which will be available to all workers.
    // This is useful for sharing a database connection pool, configuration, etc.
    let state = AppState { pool: pool.clone() };

    // Setup and start the engine
    // Give the engine a name and setup some queues.
    //
    // The name can be anything. If you run multiple nodes, this should
    // be the node name to differentiate from each node in your cluster.
    //
    // For instance, you can use a Kubernetes Pod DNS name:
    //
    // 10-0-1-162.backend.pod.cluster.local
    let _engine = Engine::builder("ishikari-example")
        .stager_interval(Duration::from_secs(5))
        .with_queue(Queue::builder("default").concurrency(10))
        .with_queue(
            Queue::builder("low_latency")
                .concurrency(20)
                // Poll every 500ms (default is 1s)
                .interval(Duration::from_millis(500)),
        )
        .with_state(Arc::new(state))
        .start(Postgres::new(pool.clone()));

    // Wait for a signal to shutdown
    tokio::signal::ctrl_c().await.unwrap();

    Ok(())
}