dbq 0.1.0

extern crate dbq;
extern crate postgres;
extern crate serde_json;

use std::error::Error;
use std::result::Result;
use std::thread;
use std::time::Duration;

#[derive(Clone)]
struct TxHandler {}

fn main() -> Result<(), Box<Error>> {
    let db_conn_params = "postgres://postgres:password@localhost/dbq";
    let conn = postgres::Connection::connect(db_conn_params, postgres::TlsMode::None)?;

    // Schema config allows for changing the database schema and table names
    // Defaults are no schema (default is used) and tables are prefixed with "dbq_"
    let schema_config = dbq::SchemaConfig::default();
    // Run the migrations on start. Migrations are idempotent and should be run
    // on startup
    dbq::run_migrations(&schema_config, &conn, Some(646_271)).unwrap();

    let queue = dbq::Queue::new(schema_config, "example_tx".to_string());

    // Enqueue a job
    let conn = postgres::Connection::connect(db_conn_params, postgres::TlsMode::None)?;
    queue.enqueue("job_a", serde_json::Value::Null, 3, &conn)?;

    // Start a worker pool
    let workers_config = dbq::WorkerPoolConfig::new(queue, db_conn_params, TxHandler {})?;
    let workers = dbq::WorkerPool::start(workers_config);

    // Give a worker time to find and start the job
    thread::sleep(Duration::new(1, 0));
    // Shutdown the worker pool waiting for all currently executing jobs to finish
    workers.join();
    Ok(())
}

impl dbq::Handler for TxHandler {
    type Error = postgres::Error;

    fn handle(&self, ctx: dbq::JobContext) -> Result<(), Self::Error> {
        // This SQL statement runs inside the same transaction that runs the
        // job. It will only commit if the job succeeds. If the job fails,
        // this statement will be rolled back and the database will not be
        // left in an inconsistent state.
        ctx.tx.execute("insert into foo (a) values ('bar')", &[])?;
        Ok(())
    }
}