Crate background_jobs

Background Jobs

This crate provides tooling required to run some processes asynchronously from a usually synchronous application. The standard example of this is Web Services, where certain things need to be processed, but processing them while a user is waiting for their browser to respond might not be the best experience.

Usage

Add Background Jobs to your project

[dependencies]
actix = "0.8"
background-jobs = "0.6.0"
anyhow = "1.0"
futures = "0.1"
serde = { version = "1.0", features = ["derive"] }

To get started with Background Jobs, first you should define a job.

Jobs are a combination of the data required to perform an operation, and the logic of that operation. They implment the Job, serde::Serialize, and serde::DeserializeOwned.

use anyhow::Error;
use background_jobs::Job;
use futures::future::{ok, Ready};

#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
pub struct MyJob {
    some_usize: usize,
    other_usize: usize,
}

impl MyJob {
    pub fn new(some_usize: usize, other_usize: usize) -> Self {
        MyJob {
            some_usize,
            other_usize,
        }
    }
}

impl Job for MyJob {
    type State = ();
    type Future = Ready<Result<(), Error>>;

    const NAME: &'static str = "MyJob";

    fn run(self, _: Self::State) -> Self::Future {
        println!("args: {:?}", self);

        ok(())
    }
}

The run method for a job takes an additional argument, which is the state the job expects to use. The state for all jobs defined in an application must be the same. By default, the state is an empty tuple, but it's likely you'll want to pass in some Actix address, or something else.

Let's re-define the job to care about some application state.

use anyhow::Error;
use background_jobs::Job;
use futures::future::{ok, Ready};

#[derive(Clone, Debug)]
pub struct MyState {
    pub app_name: String,
}

impl MyState {
    pub fn new(app_name: &str) -> Self {
        MyState {
            app_name: app_name.to_owned(),
        }
    }
}

impl Job for MyJob {
    type State = MyState;
    type Future = Ready<Result<(), Error>>;

    const NAME: &'static str = "MyJob";

    fn run(self, state: Self::State) -> Self::Future {
        info!("{}: args, {:?}", state.app_name, self);

        ok(())
    }
}

Running jobs

By default, this crate ships with the background-jobs-actix feature enabled. This uses the background-jobs-actix crate to spin up a Server and Workers, and provides a mechanism for spawning new jobs.

background-jobs-actix on it's own doesn't have a mechanism for storing worker state. This can be implemented manually by implementing the Storage trait from background-jobs-core, or the provided in-memory store can be used.

With that out of the way, back to the examples:

Main

use anyhow::Error;
use background_jobs::{ServerConfig, memory_storage::Storage, WorkerConfig};

#[actix_rt::main]
async fn main() -> Result<(), Error> {
    // Set up our Storage
    let storage = Storage::new();

    // Start the application server. This guards access to to the jobs store
    let queue_handle = ServerConfig::new(storage).start();

    // Configure and start our workers
    WorkerConfig::new(move || MyState::new("My App"))
        .register::<MyJob>()
        .set_processor_count(DEFAULT_QUEUE, 16)
        .start(queue_handle.clone());

    // Queue our jobs
    queue_handle.queue(MyJob::new(1, 2))?;
    queue_handle.queue(MyJob::new(3, 4))?;
    queue_handle.queue(MyJob::new(5, 6))?;

    // Block on Actix
    actix_rt::signal::ctrl_c().await?;
    Ok(())
}

Complete Example

For the complete example project, see the examples folder

Bringing your own server/worker implementation

If you want to create your own jobs processor based on this idea, you can depend on the background-jobs-core crate, which provides the Job trait, as well as some other useful types for implementing a jobs processor and job store.

Modules

dev	Useful types and methods for developing Storage and Processor implementations.
memory_storage	A default, in-memory implementation of a storage mechanism

Structs

JobStat	A time-based overview of job completion and failures
QueueHandle	A handle to the job server, used for queuing new jobs
Stats	Statistics about the jobs processor
WorkerConfig	Worker Configuration

Enums

Backoff	Different styles for retrying jobs
MaxRetries	How many times a job should be retried before giving up

Traits

ActixJob	The ActixJob trait defines parameters pertaining to an instance of background job
Job	The Job trait defines parameters pertaining to an instance of background job

Functions

create_server

Create a new Server