sqlxmq

A job runner is responsible for listening for new jobs and executing them. You set this up using a JobRegistry.

Workflow:

1. Initialize Registry: Create a JobRegistry by passing a slice of job functions: JobRegistry::new(&[job_function]).
2. Configure Registry: Use registry.set_context(value) to provide context that will be passed to job arguments.
3. Create Runner: Call .runner(&pool) on the registry, providing a sqlx PostgreSQL connection pool.
4. Configure Runner: Use .set_concurrency(min, max) to specify the minimum and maximum number of concurrent jobs.
5. Start Runner: Call .run().await? to start the runner. The runner will continue to listen and execute jobs until the runner instance is dropped.

use std::error::Error;

use sqlxmq::JobRegistry;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // You'll need to provide a Postgres connection pool.
    let pool = connect_to_db().await?;

    // Construct a job registry from our single job.
    let mut registry = JobRegistry::new(&[example_job]);
    // Here is where you can configure the registry
    // registry.set_error_handler(...)

    // And add context
    registry.set_context("Hello");

    let runner = registry
        // Create a job runner using the connection pool.
        .runner(&pool)
        // Here is where you can configure the job runner
        // Aim to keep 10-20 jobs running at a time.
        .set_concurrency(10, 20)
        // Start the job runner in the background.
        .run()
        .await?;

    // The job runner will continue listening and running
    // jobs until `runner` is dropped.
    Ok(())
}

A JobRegistry manages the collection of available jobs and any shared context. A JobRunner is the component that actually listens for and executes those jobs.

Workflow:

1. Create a JobRegistry by passing a slice of your job functions.
2. (Optional) Use registry.set_context(value) to provide data accessible to all jobs.
3. Create a runner using .runner(&pool) where pool is a sqlx::PgPool.
4. Configure the runner (e.g., .set_concurrency(min, max)).
5. Call .run().await to start the runner. The runner will continue running until the returned handle is dropped.

use std::error::Error;
use sqlxmq::JobRegistry;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let pool = connect_to_db().await?;

    let mut registry = JobRegistry::new(&[example_job]);
    registry.set_context("Hello");

    let runner = registry
        .runner(&pool)
        .set_concurrency(10, 20)
        .run()
        .await?;

    Ok(())
}

To execute jobs, use JobRunnerOptions to configure a runner and then call .run(). You must provide a sqlx::Pool<Postgres> and a dispatch function that accepts a CurrentJob.

Key configuration options:

• set_concurrency(min, max): Sets the concurrency limits. The runner polls for more jobs when active jobs fall below min.
• set_channel_names(&[&str]): Subscribes the runner to specific channels. If omitted, it subscribes to all channels.
• set_keep_alive(bool): Enables/disables automatic background keep-alive signals for running jobs (defaults to true).

Calling .run() returns a JobRunnerHandle, which controls the lifecycle of the background runner task.

let mut options = JobRunnerOptions::new(&pool, |job| {
    // Your job logic here
});

options
    .set_concurrency(8, 16)
    .set_channel_names(&["email_notifications", "image_processing"]);

let mut handle = options.run().await?;

// The runner runs in the background until `handle` is dropped or `.stop()` is called.

To define a job, create an asynchronous function and decorate it with the #[job] attribute.

Function Signature Requirements:

• The first argument must always be a CurrentJob instance.
• Subsequent arguments are optional and can be used to access context provided via JobRegistry::set_context.
• The function must return a Result<(), Box<dyn Error + Send + Sync + 'static>>.

Key Operations within a Job:

• Payload Decoding: Use current_job.json()? to decode a JSON payload into a type that implements Deserialize.
• Completion: Call current_job.complete().await? to mark the job as successfully finished.

Configuration: Use the #[job] attribute to set default options, such as channel_name.

use std::error::Error;

use sqlxmq::{job, CurrentJob};

// Arguments to the `#[job]` attribute allow setting default job options.
#[job(channel_name = "foo")]
async fn example_job(
    // The first argument should always be the current job.
    mut current_job: CurrentJob,
    // Additional arguments are optional, but can be used to access context
    // provided via [`JobRegistry::set_context`].
    message: &'static str,
) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
    // Decode a JSON payload
    let who: Option<String> = current_job.json()?;

    // Do some work
    println!("{}, {}!", message, who.as_deref().unwrap_or("world"));

    // Mark the job as complete
    current_job.complete().await?;

    Ok(())
}

The JobBuilder provides several methods to customize job behavior. Note that most methods return &mut Self to allow chaining.