aide-de-camp-sqlite

A SQLite backed implementation of the job Queue.

NOTE: It is possible that a single job gets sent to two runners. This is due to SQLite lacking row locking and BEGIN EXCLUSIVE TRANSACTION not working well (it was very slow) for this use case. This is only an issue at high concurrency, in which case you probably don't want to use SQLite in the first place. In other words, this isn't “Exactly Once” kind of queue.

Schema

```sql CREATE TABLE IF NOT EXISTS adcqueue ( jid TEXT PRIMARY KEY, queue TEXT NOT NULL default 'default', jobtype TEXT not null, payload blob not null, retries int not null default 0, scheduledat INTEGER not null, startedat INTEGER, enqueued_at INTEGER not null default (strftime('%s', 'now')), priority TINYINT not null default 0, );

CREATE TABLE IF NOT EXISTS adcdeadqueue ( jid TEXT PRIMARY KEY, queue TEXT NOT NULL, jobtype TEXT not null, payload blob not null, retries int not null, scheduledat INTEGER not null, startedat INTEGER not null, enqueuedat INTEGER not null, died_at INTEGER not null default (strftime('%s', 'now')), priority TINYINT not null default 0, );

CREATE INDEX IF NOT EXISTS adcqueuejobs ON adcqueue ( scheduledat asc, startedat asc, queue, jobtype ); ```

Crate includes SQLx MIGRATOR that could be used to manage schema.

NOTE: SQLx doesn't support multiple migrators in the same database. That means ADC should either have dedicated schema/database or it's your responsibility to apply migrations.

Example

```rust

use aidedecampsqlite::{SqliteQueue, MIGRATOR}; use aidedecamp::prelude::{Queue, JobProcessor, JobRunner, RunnerOptions, RunnerRouter, Duration, Xid, CancellationToken}; use asynctrait::async_trait; use sqlx::SqlitePool;

struct MyJob;

[asynctrait::asynctrait]

impl JobProcessor for MyJob { type Payload = Vec; type Error = anyhow::Error;

async fn handle(&self, jid: Xid, payload: Self::Payload, cancellation_token: CancellationToken) -> Result<(), Self::Error> {
    // Do work here
    Ok(())
}

fn name() -> &'static str {
    "my_job"
}

}

[tokio::main]

async fn main() -> Result<(), Box> {

let pool = SqlitePool::connect(":memory:").await?;
// Setup schema, alternatively you can add schema to your migrations.
MIGRATOR.run(&pool).await?;
let queue = SqliteQueue::with_pool(pool);

// Add job the queue to run next
let _jid = queue.schedule::<MyJob>(vec![1,2,3], 0).await?;

// First create a job processor and router
let router = {
    let mut r = RunnerRouter::default();
    r.add_job_handler(MyJob);
    r
};
// Setup runner to at most 10 jobs concurrently
let mut runner = JobRunner::new(queue, router, 10, RunnerOptions::default());
// Poll the queue every second, this will block unless something went really wrong.
// The future supplied as the second parameter will tell the server to shut down when it completes.
runner.run_with_shutdown(Duration::seconds(1), async move {
    // To avoid blocking this doctest, run for 10 milliseconds, then initiate shutdown.
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;

    // In a real application, you may want to wait for a CTRL+C event or something similar.
    // You could do this with tokio using the signal module: tokio::signal::ctrl_c().await.expect("failed to install CTRL+C signal handler");
}).await?;
Ok(())

} ```