ishikari 0.1.0

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

//! Ishikari Stager
//!
//! The Stager is responsible for moving tasks from scheduled/retryable to available.

use crate::engine::Storage;
use std::pin::pin;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, info, instrument};

#[derive(Debug)]
pub struct Stager<S>
where
    S: Storage + 'static,
{
    storage: Arc<S>,
    interval: Duration,
    limit: i32,
}

impl<S> Stager<S>
where
    S: Storage + 'static,
{
    pub fn new(storage: Arc<S>, interval: Duration, limit: i32) -> Self {
        Self {
            storage,
            interval,
            limit,
        }
    }

    #[instrument(skip(self))]
    pub fn start(self) -> tokio::task::JoinHandle<Result<(), S::Error>> {
        tokio::spawn(async move {
            info!("starting stager");
            let mut interval = pin!(tokio::time::interval(self.interval));

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        let count = self.storage.stage_jobs(self.limit).await?;
                        debug!(count = count, "staging jobs");

                        if count > 0 {
                            info!(count = count, "staged jobs");
                        }
                    }
                }
            }
        })
    }
}