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-rt = "2.6.0"
background-jobs = "0.15.0"
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 background_jobs::[Job, BoxError};
use std::future::{ready, 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 Error = BoxError;
    type Future = Ready<Result<(), BoxError>>;

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

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

        ready(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 background_jobs::[Job, BoxError};
use std::future::{ready, 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 Error = BoxError;
    type Future = Ready<Result<(), BoxError>>;

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

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

        ready(Ok(()))
    }
}

Running jobs

By default, this crate ships with the actix-rt 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 background_jobs::{ServerConfig, memory_storage::Storage, actix::WorkerConfig, BoxError};

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

    // Configure and start our workers
   let arbiter = Arbiter::new();

   // Configure and start our workers
   let queue_handle =
       WorkerConfig::new_in_arbiter(arbiter.handle(), storage, |_| MyState::new("My App"))
           .register::<MyJob>()
           .set_worker_count(DEFAULT_QUEUE, 16)
           .start();

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

    // 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.

Available Features

feature	description
actix-rt	Enables the actix-based job runner and the ActixTimer for the in-memory job storage implementation
metrics	Enables the metrics subscriber to extract metrics from the MetricsStorage adapter
postgres	Enables the postgres job storage adapter
sled	Enables the sled job storage adapter
tokio	Enables the tokio-based job runner and the TokioTimer for the in-memory job storage impelementation
completion-logging	Enables a tracing event that occurs whenever a job completes
error-logging	Enables a tracing event that occurs whenever a job fails

Modules

actix

dev

Useful types and methods for developing Storage and Processor implementations.

memory_storage

metrics

postgres

sled

tokio

Structs

BoxError

A simple error box that provides no additional formatting utilities

Enums

Backoff

Different styles for retrying jobs

MaxRetries

How many times a job should be retried before giving up

Traits

Job

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

UnsendJob

The UnsendJob trait defines parameters pertaining to an instance of a background job

UnsendSpawner

The mechanism used to spawn Unsend futures, making them Send