sqlxmq

https://docs.rs/sqlxmq/latest/
Admin
sqlxmq is a job queue library built on sqlx and PostgreSQL that enables CRUD applications to run...
Tokens:70,000
Snippets:346
Trust Score:9.7
Update:1 year ago
Show doc for...
Tokens:
Raw
### Setup and Run Job Registry

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Initializes a JobRegistry with specified jobs, sets a shared context, and then runs the registry using a database pool. This is a common pattern for starting background job processing.

```Rust
let mut registry = JobRegistry::new(&[example_job1, example_job2, example_job_with_ctx]);
registry.set_context(42).set_context("Hello, world!");
registry.runner(pool).run().await.unwrap()
```

--------------------------------

### Define and Run SQLxMQ Job Runner

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Demonstrates how to define a job using the `#[job]` attribute, create a `JobRegistry`, configure a job runner with concurrency settings, and start it using a PostgreSQL connection pool. This setup allows the application to listen for and process jobs asynchronously.

```rust
#![allow(unused_imports)]
use std::error::Error;
use sqlxmq::{job, CurrentJob, JobRegistry};

// Define an example job that can be processed by sqlxmq.
#[job]
async fn example_job(
    current_job: CurrentJob,
) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
    // Job logic goes here. For example, process data from the job payload.
    // println!("Processing job: {:?}", current_job.payload());
    Ok(())
}

// Placeholder for connecting to the database.
async fn connect_to_db() -> sqlx::Result<sqlx::Pool<sqlx::Postgres>> {
    // In a real application, you would establish a connection pool here.
    // Example: sqlx::PgPool::connect("postgres://user:password@host:port/database").await
    unimplemented!("Database connection not implemented")
}

#[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, e.g., set an error handler.
    // registry.set_error_handler(...)

    // And add context that can be accessed within jobs.
    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.
    // For demonstration, we might await indefinitely or until a shutdown signal.
    // For this example, we'll just let it run until main exits.
    Ok(())
}

```

--------------------------------

### SQLx MQ Job Spawning Example

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Demonstrates how to build and spawn a job using the sqlxmq crate. This example shows how to override default job configurations, such as setting a specific channel name and providing JSON payload data, before spawning the job asynchronously.

```Rust
/// Example function demonstrating job spawning
///
/// # Arguments
///
/// * `pool` - A sqlx::Pool connection to the database.
///
/// # Returns
///
/// A Result indicating success or an error.
async fn example(pool: sqlx::Pool<sqlx::Postgres>) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
    // Assuming 'example_job' is a job defined elsewhere using the #[job] macro
    // example_job.builder()
    //     // This is where we can override job configuration
    //     .set_channel_name("bar")
    //     .set_json("John")? // Example of setting JSON payload
    //     .spawn(&pool)
    //     .await?;
    Ok(())
}
```

--------------------------------

### Start PostgreSQL Listener for SQLxMQ

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Sets up and starts a PostgreSQL LISTEN/NOTIFY listener for the SQLxMQ job runner. It connects to the database, subscribes to specified channels (or a default 'mq' channel), and spawns a background task to continuously receive notifications and signal the job runner.

```rust
async fn start_listener(job_runner: Arc<JobRunner>) -> Result<OwnedHandle, sqlx::Error> {
    let mut listener = PgListener::connect_with(&job_runner.options.pool).await?;
    if let Some(channels) = &job_runner.options.channel_names {
        let names: Vec<String> = channels.iter().map(|c| format!("mq_{}", c)).collect();
        listener
            .listen_all(names.iter().map(|s| s.as_str()))
            .await?;
    } else {
        listener.listen("mq").await?;
    }
    Ok(OwnedHandle::new(task::spawn(async move {
        let mut num_errors = 0;
        loop {
            if num_errors > 0 || listener.recv().await.is_ok() {
                job_runner.notify.notify_one();
                num_errors = 0;
            } else {
                tokio::time::sleep(Duration::from_secs(1 << num_errors)).await;
                num_errors += 1;
            }
        }
    })))
}
```

--------------------------------

### Setup SQLx Postgres Test Pool

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

This function sets up a PostgreSQL database connection pool for testing. It ensures logger initialization, loads environment variables, acquires a lock for exclusive access, connects to the database using a URL from the environment, and cleans up relevant test tables ('mq_payloads' and 'mq_msgs'). It returns a guard that holds the lock and the database pool.

```Rust
use std::env;
    use std::ops::Deref;
    use std::sync::{Arc, Once};
    use std::time::Duration;
    use futures::channel::mpsc;
    use futures::StreamExt;
    use sqlx::{Pool, Postgres};
    use tokio::sync::{Mutex, MutexGuard};
    use tokio::task;

    // Helper struct to manage the test pool and its associated lock.
    // field 0 is never read, but its drop is important
    #[allow(dead_code)]
    struct TestGuard<T>(MutexGuard<'static, ()>, T);

    impl<T> Deref for TestGuard<T> {
        type Target = T;

        fn deref(&self) -> &T {
            &self.1
        }
    }

    async fn test_pool() -> TestGuard<Pool<Postgres>> {
        static INIT_LOGGER: Once = Once::new();
        static TEST_MUTEX: Mutex<()> = Mutex::const_new(());

        let guard = TEST_MUTEX.lock().await;

        // Load environment variables from .env file
        let _ = dotenvy::dotenv();

        // Initialize pretty_env_logger only once
        INIT_LOGGER.call_once(pretty_env_logger::init);

        // Connect to the PostgreSQL database
        let pool = Pool::connect(&env::var("DATABASE_URL").unwrap())
            .await
            .unwrap();

        // Truncate the mq_payloads table
        sqlx::query("TRUNCATE TABLE mq_payloads")
            .execute(&pool)
            .await
            .unwrap();
        // Delete from mq_msgs table, excluding entries with uuid_nil()
        sqlx::query("DELETE FROM mq_msgs WHERE id != uuid_nil()")
            .execute(&pool)
            .await
            .unwrap();

        TestGuard(guard, pool)
    }
```

--------------------------------

### SQLxMQ: Job Retry Mechanism Test Setup

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Sets up a test scenario to verify the job retry mechanism. It initializes a job runner and configures a backoff delay, preparing to test how failed jobs are handled and retried by the SQLxMQ system.

```rust
    #[tokio::test]
    async fn it_retries_failed_jobs() {
        {
            let pool = &*test_pool().await;
            let (_runner, counter) = test_job_runner(pool, move |_| async {}).await;

            let backoff = default_pause() + 300;

            assert_eq!(counter.load(Ordering::SeqCst), 0);
```

--------------------------------

### Define a Job with sqlxmq

Source: https://docs.rs/sqlxmq/latest/sqlxmq/index

Demonstrates how to define a background job using the `sqlxmq` crate. It shows the use of the `#[job]` attribute for configuration, the `CurrentJob` type for job context, decoding JSON payloads, performing work, and marking the job as complete. This example highlights transactional job processing and payload handling.

```rust
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(())
}
```

--------------------------------

### Job Execution Logging and Error Handling

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/registry.rs

This snippet illustrates the process of executing a job, logging its start and completion times, and handling potential errors. It uses the `log` crate for informational messages and measures execution duration.

```rust
let start_time = Instant::now();
log::info!("Job `{}` started.", name);
if let Err(e) = f.await {
    error_handler(name, e.into());
} else {
    log::info!(
        "Job `{}` completed in {}s.",
        name,
        start_time.elapsed().as_secs_f64()
    );
}
```

--------------------------------

### JobRunnerOptions Configuration and Execution

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.JobRunnerOptions

API for configuring and running job runners. This includes methods for initialization, setting concurrency limits, specifying subscribed channels, controlling keep-alive behavior, and starting the runner or testing a single job.

```APIDOC
JobRunnerOptions:
  __init__ (via `new` method):
    pub fn new<F: Fn(CurrentJob) + Send + Sync + 'static>(pool: &Pool<Postgres>, f: F) -> Self
      - Initializes a new job runner builder.
      - Parameters:
        - pool: &Pool<Postgres> - The database connection pool to use.
        - f: F - The asynchronous function to execute for each job. Must implement Fn(CurrentJob) + Send + Sync + 'static.
      - Returns: Self - A new JobRunnerOptions instance.

  set_concurrency:
    pub fn set_concurrency(&mut self, min_concurrency: usize, max_concurrency: usize) -> &mut Self
      - Sets the concurrency limits for the job runner.
      - When the number of active jobs falls below `min_concurrency`, the runner polls for more, up to `max_concurrency`.
      - The difference between min and max dictates the maximum batch size.
      - Parameters:
        - min_concurrency: usize - The minimum number of concurrent jobs to maintain.
        - max_concurrency: usize - The maximum number of concurrent jobs allowed.
      - Returns: &mut Self - A mutable reference to the JobRunnerOptions for chaining.

  set_channel_names:
    pub fn set_channel_names<'a>(&'a mut self, channel_names: &[&str]) -> &'a mut Self
      - Specifies the channels the job runner will subscribe to.
      - If unspecified, the job runner subscribes to all channels.
      - Parameters:
        - channel_names: &[&str] - A slice of string slices, where each string is a channel name.
      - Returns: &'a mut Self - A mutable reference to the JobRunnerOptions for chaining.

  set_keep_alive:
    pub fn set_keep_alive(&mut self, keep_alive: bool) -> &mut Self
      - Configures whether to automatically keep jobs alive while they are running.
      - Defaults to `true`.
      - Parameters:
        - keep_alive: bool - Set to `true` to enable automatic keep-alive, `false` otherwise.
      - Returns: &mut Self - A mutable reference to the JobRunnerOptions for chaining.

  run:
    pub async fn run(&self) -> Result<JobRunnerHandle, Error>
      - Starts the job runner in the background.
      - The runner will continue executing jobs until the returned `JobRunnerHandle` is dropped.
      - Returns: Result<JobRunnerHandle, Error> - A handle to the running job runner or an error if startup fails.

  test_one:
    pub async fn test_one(&self) -> Result<(), Error>
      - Executes a single job and then returns. Intended for use in tests.
      - Assumes a job has been spawned and is ready to run.
      - Returns: Result<(), Error> - Ok(()) on success, or an Error if the job execution fails.
```

--------------------------------

### Test Job Checkpointing with Conditional Logic in Rust

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

This example demonstrates testing the checkpointing functionality of `sqlxmq`. It sets up a job runner that processes jobs, conditionally completing them or checkpointing them with new JSON state. The code shows how to use `JobBuilder` to set initial job parameters, including a JSON payload, and how the job runner logic interacts with `checkpoint()` and `complete()`.

```rust
use sqlxmq::{Checkpoint, JobBuilder};
use std::sync::atomic::{AtomicUsize, Ordering};

// Assuming 'test_pool', 'test_job_runner', 'default_pause' are defined elsewhere
// let pool = &*test_pool().await;
// let backoff = default_pause();

// Setup a job runner that handles checkpointing based on payload
let (_runner, counter) = test_job_runner(pool, move |mut current_job| async move {
    let state: bool = current_job.json().unwrap().unwrap();
    if state {
        // If state is true, complete the job
        current_job.complete().await.unwrap();
    } else {
        // If state is false, checkpoint the job with new state
        current_job
            .checkpoint(Checkpoint::new().set_json(&true).unwrap())
            .await
            .unwrap();
    }
})
.await;

// Assert initial state
assert_eq!(counter.load(Ordering::SeqCst), 0);

// Create and spawn a job that will trigger checkpointing
JobBuilder::new("foo")
    .set_retry_backoff(Duration::from_millis(backoff))
    .set_retries(5)
    .set_json(&false) // Set initial JSON payload to false
    .spawn(pool)
    .await
    .unwrap();

// Wait for the job to be processed and checkpointed
// pause().await; // Assuming pause() is a helper for yielding execution

// Assert that the job was checkpointed (counter incremented)
// assert_eq!(counter.load(Ordering::SeqCst), 1); // Example assertion, actual count depends on runner logic
```

--------------------------------

### sqlxmq Crate Overview

Source: https://docs.rs/sqlxmq/latest/sqlxmq/all

Provides details about the sqlxmq crate, its version, license, dependencies, and links to external resources like GitHub and crates.io. It also lists the available platform targets and feature flags.

```rust
Crate: sqlxmq
Version: 0.6.0
Description: A reliable job queue using PostgreSQL as a backing store
License: MIT OR Apache-2.0
Dependencies:
  - anymap2 ^0.13.0
  - chrono ^0.4.19
  - dotenvy ^0.15.3
  - log ^0.4.14
  - serde ^1.0.124
  - serde_json ^1.0.64
  - sqlx ^0.8
  - sqlxmq_macros ^0.6.0
  - tokio ^1.8.3
  - uuid ^1.1.2
Platforms:
  - i686-pc-windows-msvc
  - x86_64-apple-darwin
  - x86_64-pc-windows-msvc
  - x86_64-unknown-linux-gnu
Features:
  - (List of available feature flags)
```

--------------------------------

### Start Job Runner (Rust)

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Starts the job runner in the background. The runner operates asynchronously and will stop when the returned `JobRunnerHandle` is dropped. It returns a handle to manage the running process.

```rust
/// Start the job runner in the background. The job runner will stop when the
/// returned handle is dropped.
pub async fn run(&self) -> Result<JobRunnerHandle, sqlx::Error> {
    let options = self.clone();
    let job_runner = Arc::new(JobRunner {
        options,
        running_jobs: AtomicUsize::new(0),
        notify: Notify::new(),
    });
    let listener_task = start_listener(job_runner.clone()).await?;
    let handle = OwnedHandle::new(task::spawn(main_loop(job_runner.clone(), listener_task)));
    Ok(JobRunnerHandle {
        runner: job_runner,
        handle: Some(handle),
    })
}
```

--------------------------------

### sqlxmq Source Files

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/spawn.rs

Lists the primary source files for the sqlxmq Rust crate. These files contain the implementation details for the job queue functionality.

```Rust
hidden.rs
lib.rs
registry.rs
runner.rs
spawn.rs
utils.rs
```

--------------------------------

### sqlxmq Crate Overview

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.CurrentJob

Provides metadata and links related to the sqlxmq Rust crate, version 0.6.0. It lists dependencies, license information, and links to source code and crates.io.

```rust
Dependencies:
  anymap2 ^0.13.0
  chrono ^0.4.19
  dotenvy ^0.15.3
  log ^0.4.14
  serde ^1.0.124
  serde_json ^1.0.64
  sqlx ^0.8
  sqlxmq_macros ^0.6.0
  tokio ^1.8.3
  uuid ^1.1.2

License: MIT OR Apache-2.0
Source: https://github.com/Diggsey/sqlxmq
Crates.io: https://crates.io/crates/sqlxmq
```

--------------------------------

### Get Job Name (Rust)

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Retrieve the name associated with the current job.

```rust
/// Returns the name of this job.
pub fn name(&self) -> &str {
    &self.name
}
```

--------------------------------

### sqlxmq Crate Overview

Source: https://docs.rs/sqlxmq/latest/sqlxmq/fn.should_retry

Provides essential information about the sqlxmq crate, including its purpose, version, license, and links to external resources like GitHub and crates.io.

```APIDOC
Project: sqlxmq
Version: 0.6.0
Description: A reliable job queue using PostgreSQL as a backing store.
License: MIT OR Apache-2.0
Repository: https://github.com/Diggsey/sqlxmq
Crates.io: https://crates.io/crates/sqlxmq
Documentation: https://docs.rs/sqlxmq/latest/sqlxmq/

Dependencies:
  - anymap2 ^0.13.0
  - chrono ^0.4.19
  - dotenvy ^0.15.3
  - log ^0.4.14
  - serde ^1.0.124
  - serde_json ^1.0.64
  - sqlx ^0.8
  - sqlxmq_macros ^0.6.0
  - tokio ^1.8.3
  - uuid ^1.1.2

Platform Support:
  - i686-pc-windows-msvc
  - x86_64-apple-darwin
  - x86_64-pc-windows-msvc
  - x86_64-unknown-linux-gnu

Feature Flags:
  - (Details not provided in source text, refer to crate documentation)
```

--------------------------------

### Get Job ID (Rust)

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Retrieve the unique identifier (UUID) for the current job.

```rust
/// Returns the ID of this job.
pub fn id(&self) -> Uuid {
    self.id
}
```

--------------------------------

### Listen for Jobs with sqlxmq

Source: https://docs.rs/sqlxmq/latest/sqlxmq

Demonstrates how to set up a job runner that listens for and executes background jobs. It involves creating a job registry, configuring concurrency, and running the listener.

```rust
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(())
}
```

--------------------------------

### Get Raw Binary Payload (Rust)

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Retrieve the raw byte slice of the job's payload, if available.

```rust
/// Returns the raw binary payload for this job.
pub fn raw_bytes(&self) -> Option<&[u8]> {
    // Implementation would typically access a byte buffer field
    // For example: self.payload_bytes.as_deref()
    // This is a placeholder based on the function signature.
    None
}
```

--------------------------------

### Rust: Test Registry Usage for Named Jobs

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Illustrates testing the usage of a job registry in Rust with sqlxmq. Shows how to spawn different types of named jobs using their builders and verifies their execution context.

```Rust
let pool = &*test_pool().await;
            let _runner = named_job_runner(pool).await;

            example_job1.builder().spawn(pool).await.unwrap();
            example_job2.builder().spawn(pool).await.unwrap();
            example_job_with_ctx.builder().spawn(pool).await.unwrap();
            pause().await;
```

--------------------------------

### Get Raw JSON Payload (Rust)

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Retrieve the raw JSON string of the job's payload, if available.

```rust
/// Returns the raw JSON payload for this job.
pub fn raw_json(&self) -> Option<&str> {
    self.payload_json.as_deref()
}
```

--------------------------------

### Rust Example: Checking Retry Status

Source: https://docs.rs/sqlxmq/latest/sqlxmq/fn.should_retry

Demonstrates how to use the `should_retry` function from the sqlxmq crate to determine if a job should be retried based on an error.

```rust
use sqlxmq::should_retry;
use std::error::Error;

// Assume 'err' is an error encountered during job processing
let err: Box<dyn Error> = /* ... get error ... */;

if should_retry(&err) {
    println!("Job should be retried.");
} else {
    println!("Job should not be retried.");
}
```

--------------------------------

### Listen for Jobs with sqlxmq

Source: https://docs.rs/sqlxmq/latest/index

Demonstrates how to set up a job runner that listens for and executes background jobs. It involves creating a job registry, configuring concurrency, and running the listener.

```rust
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(())
}
```

--------------------------------

### Rust JobRunnerHandle Struct

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Provides a handle to interact with a running job runner. This allows external control or monitoring of the runner, potentially including starting, stopping, or querying its status.

```rust
/// Job runner handle
pub struct JobRunnerHandle {
    runner: Arc<JobRunner>,
    handle: Option<OwnedHandle>,
}
```

--------------------------------

### sqlxmq Crate API Overview

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.UnknownJobError

Provides a summary of the key components available in the sqlxmq Rust crate, including structs, functions, and attribute macros for managing message queues with SQLx.

```APIDOC
Crate: sqlxmq

Structs:
  - Checkpoint: Represents a checkpoint for job processing.
  - CurrentJob: Represents the job currently being processed.
  - JobBuilder: Used for constructing new jobs.
  - JobRegistry: Manages the registration and lookup of jobs.
  - JobRunnerHandle: A handle to control a running job runner.
  - JobRunnerOptions: Configuration options for job runners.
  - NamedJob: Represents a job with a specific name.
  - OwnedHandle: An owned handle to a job runner.
  - UnknownJobError: Error returned when a job name is not found in the registry.

Functions:
  - clear(): Clears a specific job queue.
  - clear_all(): Clears all job queues.
  - commit(): Commits a job, marking it as completed.
  - exists(): Checks if a job queue exists.
  - should_retry(): Determines if a job should be retried based on certain conditions.

Attribute Macros:
  - job: Used to define and register jobs within the system.
```

--------------------------------

### Rust: Get TypeId from Any trait

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.NamedJob

Retrieves the unique TypeId for a given type, enabling runtime type identification. This method is part of the core::any::Any trait.

```rust
fn type_id(&self) -> TypeId
```

--------------------------------

### sqlxmq JobRunnerOptions API

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.JobRunnerOptions

Documentation for the JobRunnerOptions struct in the sqlxmq crate, detailing its methods for configuring and running jobs.

```APIDOC
JobRunnerOptions:
  A struct for configuring the job runner.

  Methods:
    new()
      Description: Creates a new JobRunnerOptions instance with default settings.
      Returns: JobRunnerOptions

    run(pool: sqlx::PgPool)
      Description: Starts the job runner with the provided PostgreSQL connection pool.
      Parameters:
        pool: A reference to the sqlx::PgPool.
      Returns: A future that resolves when the runner stops.

    set_channel_names(channel_names: Vec<String>)
      Description: Sets the names of the channels to listen for jobs on.
      Parameters:
        channel_names: A vector of strings representing channel names.
      Returns: Self (for chaining)

    set_concurrency(concurrency: usize)
      Description: Sets the maximum number of concurrent jobs to process.
      Parameters:
        concurrency: The number of concurrent jobs.
      Returns: Self (for chaining)

    set_keep_alive(keep_alive: std::time::Duration)
      Description: Sets the duration for how long the runner should wait before checking for new jobs.
      Parameters:
        keep_alive: A std::time::Duration specifying the keep-alive interval.
      Returns: Self (for chaining)

    test_one(pool: sqlx::PgPool)
      Description: Processes a single job from the queue for testing purposes.
      Parameters:
        pool: A reference to the sqlx::PgPool.
      Returns: A future that resolves when one job is processed or if no job is available.
```

--------------------------------

### Get Number of Running Jobs (Rust)

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.JobRunnerHandle

Retrieves the current count of jobs that are actively being processed by the JobRunnerHandle. This method provides a snapshot of the active job count.

```rust
pub fn num_running_jobs(&self) -> usize
```

--------------------------------

### Get Default Pause Duration

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Determines the default pause duration in milliseconds. It returns 1000ms if running in a CI environment, otherwise 200ms. This function helps manage polling or waiting intervals.

```Rust
fn default_pause() -> u64 {
    if is_ci() {
        1000
    } else {
        200
    }
}
```

--------------------------------

### Spawn SQLxMQ Jobs with JobBuilder

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Demonstrates creating and spawning jobs using the JobBuilder. Jobs can be configured with specific channel names and whether they should be processed in order. The code shows multiple instances of job creation and spawning.

```rust
JobBuilder::new("foo")
    .set_channel_name("bar")
    .spawn(pool)
    .await
    .unwrap();
JobBuilder::new("foo")
    .set_channel_name("baz")
    .spawn(pool)
    .await
    .unwrap();
```

--------------------------------

### Get Job Name - sqlxmq Rust

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/registry.rs

Returns the static string name associated with this job registry. This method provides a simple way to access the identifier of the job type managed by the registry.

```rust
pub const fn name(&self) -> &'static str {
    self.name
}
```

--------------------------------

### sqlxmq Attribute Macros

Source: https://docs.rs/sqlxmq/latest/sqlxmq/all

Details the attribute macros provided by the sqlxmq crate. These macros are used to define and register job handlers within the application, simplifying job processing setup.

```rust
#[derive(job)]
// Example usage for defining a job handler
```

--------------------------------

### JobRunnerOptions Clone Implementation

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.JobRunnerOptions

Provides methods for creating copies of JobRunnerOptions instances. Includes a standard clone method and a clone_from method for efficient copying.

```APIDOC
impl Clone for JobRunnerOptions

fn clone(&self) -> JobRunnerOptions
  - Returns a copy of the value.

fn clone_from(&mut self, source: &Self)
  - Performs copy-assignment from `source`.
```

--------------------------------

### Get Number of Running Jobs in Rust

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/runner.rs

Retrieves the current count of jobs actively being processed by the `JobRunner`. This method provides a snapshot of the system's current workload. It uses `Ordering::Relaxed` for efficient, non-synchronizing access to the atomic counter.

```rust
pub fn num_running_jobs(&self) -> usize {
    self.runner.running_jobs.load(Ordering::Relaxed)
}
```

--------------------------------

### sqlxmq Library Overview

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

This snippet provides a structural overview of the sqlxmq Rust library, highlighting its purpose as a job queue built on `sqlx` and `PostgreSQL`. It details the benefits of using a SQL database for job queuing, such as consistent backups and transactional operations, and lists key features like batch processing, scheduled jobs, reliable delivery, and automatic retries.

```rust
#![deny(missing_docs, unsafe_code)]
//!
//! # sqlxmq
//!
//! A job queue built on `sqlx` and `PostgreSQL`.
//!
//! This library allows a CRUD application to run background jobs without complicating its
//! deployment. The only runtime dependency is `PostgreSQL`, so this is ideal for applications
//! already using a `PostgreSQL` database.
//!
//! Although using a SQL database as a job queue means compromising on latency of
//! delivered jobs, there are several show-stopping issues present in ordinary job
//! queues which are avoided altogether.
//!
//! With most other job queues, in-flight jobs are state that is not covered by normal
//! database backups. Even if jobs _are_ backed up, there is no way to restore both
//! a database and a job queue to a consistent point-in-time without manually
//! resolving conflicts.
//!
//! By storing jobs in the database, existing backup procedures will store a perfectly
//! consistent state of both in-flight jobs and persistent data. Additionally, jobs can
//! be spawned and completed as part of other transactions, making it easy to write correct
//! application code.
//!
//! Leveraging the power of `PostgreSQL`, this job queue offers several features not
//! present in other job queues.
//!
//! # Features
//!
//! - **Send/receive multiple jobs at once.**
//!   This reduces the number of queries to the database.
//! - **Send jobs to be executed at a future date and time.**
//!   Avoids the need for a separate scheduling system.
//! - **Reliable delivery of jobs.**
//! - **Automatic retries with exponential backoff.**
//!   Number of retries and initial backoff parameters are configurable.
//! - **Transactional sending of jobs.**
//!   Avoids sending spurious jobs if a transaction is rolled back.
//! - **Transactional completion of jobs.**

```

--------------------------------

### SQLxMQ: Basic Job Spawning and Completion Test

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

Demonstrates spawning a job using JobBuilder and awaiting its completion. This snippet is part of a larger test suite for the SQLxMQ library, verifying basic job lifecycle management.

```rust
            let pool = &*test_pool().await;
            let (tx, mut rx) = mpsc::unbounded();

            let (_runner, counter) = test_job_runner(pool, move |job| {
                let tx = tx.clone();
                async move {
                    tx.unbounded_send(job).unwrap();
                }
            })
            .await;

            assert_eq!(counter.load(Ordering::SeqCst), 0);
            JobBuilder::new("foo").spawn(pool).await.unwrap();
            JobBuilder::new("bar").spawn(pool).await.unwrap();

            pause().await;
            assert_eq!(counter.load(Ordering::SeqCst), 2);

            for _ in 0..2 {
                let mut job = rx.next().await.unwrap();
                job.complete().await.unwrap();
            }
```

--------------------------------

### sqlxmq Dependencies

Source: https://docs.rs/sqlxmq/latest/settings

Lists the direct and development dependencies for the sqlxmq Rust crate. These include libraries like sqlx for database interaction, tokio for asynchronous operations, and serde for serialization.

```Rust
Dependencies:
  * [ anymap2 ^0.13.0 _normal_ ]
  * [ chrono ^0.4.19 _normal_ ]
  * [ dotenvy ^0.15.3 _normal_ ]
  * [ log ^0.4.14 _normal_ ]
  * [ serde ^1.0.124 _normal_ ]
  * [ serde_json ^1.0.64 _normal_ ]
  * [ sqlx ^0.8 _normal_ ]
  * [ sqlxmq_macros ^0.6.0 _normal_ ]
  * [ tokio ^1.8.3 _normal_ ]
  * [ uuid ^1.1.2 _normal_ ]
  * [ dotenvy ^0.15.3 _dev_ ]
  * [ futures ^0.3.13 _dev_ ]
  * [ pretty_env_logger ^0.4.0 _dev_ ]
  * [ tokio ^1 _dev_ ]
```

--------------------------------

### JobRegistry Methods

Source: https://docs.rs/sqlxmq/latest/sqlxmq/struct.JobRegistry

Provides an overview of the core methods available on the JobRegistry struct for managing jobs, setting up error handling, providing execution context, and creating job runners.

```APIDOC
JobRegistry:
  new(jobs: &[&'static [NamedJob]]) -> Self
    Construct a new job registry from the provided job list.
    Parameters:
      jobs: A slice of static NamedJob references.
    Returns: A new JobRegistry instance.

  set_error_handler(&mut self, error_handler: impl Fn(&[str], Box<dyn Error + Send + 'static>) + Send + Sync + 'static) -> &mut Self
    Set a function to be called whenever a job returns an error.
    Parameters:
      error_handler: A closure that accepts the job name and an error, and is Send + Sync + 'static.
    Returns: A mutable reference to the JobRegistry.

  set_context<C: Clone + Send + Sync + 'static>(&mut self, context: C) -> &mut Self
    Provide context for the jobs.
    Parameters:
      context: The context data, which must be Clone, Send, Sync, and 'static.
    Returns: A mutable reference to the JobRegistry.

  context<C: Clone + Send + Sync + 'static>(&self) -> C
    Access job context.
    Will panic if context with this type has not been provided.
    Parameters:
      None.
    Returns: The job context of type C.

  resolve_job(&self, name: &[str]) -> Option<&'static [NamedJob]>
    Look-up a job by name.
    Parameters:
      name: The name of the job to resolve.
    Returns: An Option containing a reference to the NamedJob if found, otherwise None.

  default_error_handler(name: &[str], error: Box<dyn Error + Send + 'static>)
    The default error handler implementation, which simply logs the error.
    Parameters:
      name: The name of the job that failed.
      error: The error that occurred.

  runner(self, pool: &Pool<Postgres>) -> JobRunnerOptions
    Construct a job runner from this registry and the provided connection pool.
    Parameters:
      pool: A connection pool for the Postgres database.
    Returns: JobRunnerOptions to configure and start the job runner.
```

--------------------------------

### Listen for Jobs with sqlxmq

Source: https://docs.rs/sqlxmq/latest/sqlxmq/index

Demonstrates how to create a job runner that listens for new jobs and executes them. It requires a Postgres connection pool and a job registry configured with specific jobs. The runner can be configured for concurrency and runs in the background.

```rust
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(())
}
```

--------------------------------

### Define and Run a Job with Custom Attributes

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/lib.rs

This snippet shows how to define a job using the `#[job]` attribute with specific configurations like channel name, ordering, retries, and backoff delay. It also demonstrates completing the job within the handler.

```rust
use sqlxmq::{Job, JobBuilder, JobRunnerHandle, JobRunnerOptions, CurrentJob};
use sqlx::Pool;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::future::Future;
use std::error::Error;

// Example job definition with specific attributes
#[job(channel_name = "foo", ordered, retries = 3, backoff_secs = 2.0)]
async fn example_job1(
    mut current_job: CurrentJob,
) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
    current_job.complete().await?;
    Ok(())
}

// Helper function to set channel name via proto
fn job_proto<'a, 'b>(builder: &'a mut JobBuilder<'b>) -> &'a mut JobBuilder<'b> {
    builder.set_channel_name("bar")
}

// Example job definition using a proto function for configuration
#[job(proto(job_proto))]
async fn example_job2(
    mut current_job: CurrentJob,
) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
    current_job.complete().await?;
    Ok(())
}

// Example job definition with context parameters
#[job]
async fn example_job_with_ctx(
    mut current_job: CurrentJob,
    ctx1: i32,
    ctx2: &'static str,
) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
    assert_eq!(ctx1, 42);
    assert_eq!(ctx2, "Hello, world!");
    current_job.complete().await?;
    Ok(())
}

// Helper function to set up and run a job runner for testing
async fn test_job_runner<F: Future + Send + 'static>(
    pool: &Pool<Postgres>,
    f: impl (Fn(CurrentJob) -> F) + Send + Sync + 'static,
) -> (JobRunnerHandle, Arc<AtomicUsize>) 
where
    F::Output: Send + 'static,
{
    let counter = Arc::new(AtomicUsize::new(0));
    let counter2 = counter.clone();
    let runner = JobRunnerOptions::new(pool, move |job| {
        counter2.fetch_add(1, Ordering::SeqCst);
        task::spawn(f(job));
    })
    .run()
    .await
    .unwrap();
    (runner, counter)
}

// Example of a named job runner setup
async fn named_job_runner(pool: &Pool<Postgres>) -> JobRunnerHandle {
    // Placeholder for actual runner setup
    // JobRunnerOptions::new(pool, |job| { /* ... */ }).run().await.unwrap()
    unimplemented!();
}

```

--------------------------------

### sqlxmq Crate Dependencies

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/spawn.rs

Lists the direct and development dependencies for the sqlxmq Rust crate version 0.6.0. These dependencies are crucial for the crate's functionality, including database interaction, asynchronous operations, and serialization.

```Rust
anymap2 ^0.13.0
chrono ^0.4.19
dotenvy ^0.15.3
log ^0.4.14
serde ^1.0.124
serde_json ^1.0.64
sqlx ^0.8
sqlxmq_macros ^0.6.0
tokio ^1.8.3
uuid ^1.1.2

# Dev Dependencies
dotenvy ^0.15.3
futures ^0.3.13
pretty_env_logger ^0.4.0
tokio ^1
```

--------------------------------

### sqlxmq Registry Methods

Source: https://docs.rs/sqlxmq/latest/src/sqlxmq/registry.rs

Provides methods for configuring and interacting with the sqlxmq job registry, including setting error handlers, managing job context, and resolving jobs.

```APIDOC
Registry:
  Methods:
    set_error_handler(error_handler: impl Fn(&str, BoxedError) + Send + Sync + 'static) -> &mut Self
      - Sets a function to be called whenever a job returns an error.
      - Parameters:
        - error_handler: A closure that accepts the job name (as &str) and the boxed error.
      - Returns: A mutable reference to the Registry for chaining.

    set_context<C: Clone + Send + Sync + 'static>(context: C) -> &mut Self
      - Provides context data that can be accessed by jobs.
      - Parameters:
        - context: The context data, which must be cloneable, sendable, and static.
      - Returns: A mutable reference to the Registry for chaining.

    context<C: Clone + Send + Sync + 'static>() -> C
      - Accesses the job context. Panics if context of the specified type has not been provided.
      - Parameters:
        - C: The type of the context to retrieve.
      - Returns: A clone of the provided context.
      - Panics: If no context of type C has been set.

    resolve_job(name: &str) -> Option<&'static NamedJob>
      - Looks up a registered job by its name.
      - Parameters:
        - name: The name of the job to find.
      - Returns: An Option containing a reference to the NamedJob if found, otherwise None.

    default_error_handler(name: &str, error: BoxedError)
      - The default error handler implementation, which logs the error using the 'log' crate.
      - Parameters:
        - name: The name of the job that failed.
        - error: The boxed error that occurred.

    spawn_internal<E: Into<Box<dyn Error + Send + Sync + 'static>>>(name: &'static str, f: impl Future<Output = Result<(), E>> + Send + 'static)
      - Internal method to spawn a job future, handling potential errors via the configured error handler.
      - Parameters:
        - name: The static name of the job.
        - f: The future representing the job's execution.
      - Note: This method is marked #[doc(hidden)] and intended for internal use.
```

--------------------------------

### sqlxmq Attribute Macros

Source: https://docs.rs/sqlxmq/latest/sqlxmq/fn.exists

Provides attribute macros for simplifying job registration and definition within the sqlxmq framework.

```APIDOC
job
  - An attribute macro used to define and register job handlers.
  - Typically applied to functions that process specific job types.
```
Tokens:
Raw