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> { 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(()) } }