Struct Worker 

pub struct Worker { /* private fields */ }

Worker for processing jobs from a queue.

Implementations

impl Worker

pub fn new(queue: Queue) -> Self

Create a new worker.

pub fn with_config(queue: Queue, config: WorkerConfig) -> Self

Create a worker with custom configuration.

pub async fn register_handler<F, Fut>( &mut self, job_type: impl Into<String>, handler: F, )

where F: Fn(Job) -> Fut + Send + Sync + 'static, Fut: Future<Output = QueueResult<()>> + Send + 'static,

Register a job handler.

The handler is inserted synchronously before this call returns, so a register_handler(...).await immediately followed by start() never races a worker that dequeues a job before the handler is present.

Examples

use armature_queue::*;

let queue = Queue::new("redis://localhost:6379", "default").await?;
let mut worker = Worker::new(queue);

worker
    .register_handler("send_email", |job| async move {
        // Send email logic
        println!("Sending email: {:?}", job.data);
        Ok(())
    })
    .await;

pub async fn start(&mut self) -> QueueResult<()>

Start the worker.

pub async fn process_batch( &self, job_type: &str, max_batch_size: usize, ) -> QueueResult<Vec<JobId>>

Process multiple jobs of the same type in parallel

This method dequeues and processes multiple jobs of the same type concurrently, providing significant throughput improvements.

Performance

• Sequential: O(n * job_time)
• Parallel: O(max(job_times))
• Speedup: 3-5x higher throughput

Examples

// Process up to 10 image processing jobs in parallel
let processed = worker.process_batch("process_image", 10).await?;
println!("Processed {} jobs", processed.len());

pub async fn register_cpu_intensive_handler<F>( &mut self, job_type: impl Into<String>, handler: F, )

where F: Fn(Job) -> QueueResult<()> + Send + Sync + 'static,

Register a CPU-intensive handler that runs in blocking thread pool

For CPU-bound operations (image processing, encryption, etc.), use this method to avoid blocking the async runtime.

Examples

worker
    .register_cpu_intensive_handler("resize_image", |job| {
        // CPU-intensive work here
        let image_path = job.data["path"].as_str().unwrap();
        // ... resize image ...
        Ok(())
    })
    .await;

pub async fn stop(&mut self) -> QueueResult<()>

Stop the worker gracefully.

This is stop_with_timeout using self.config.job_timeout as the grace period. That bound is not arbitrary: each worker task wraps its current job’s handler invocation in tokio::time::timeout(job_timeout, ...) (see the loop spawned by start), so job_timeout is already the worst-case time a task can be stuck inside a handler before it self-times-out the job and loops back to observe the stop flag. Waiting that long guarantees every task gets the chance to either finish its current job or hit its own internal timeout – and therefore call queue.complete()/queue.fail() – before shutdown falls back to aborting anything still outstanding.

This wrapper discards the StopOutcome that stop_with_timeout returns, for backward-compatible callers that only care whether shutdown was initiated successfully. Call stop_with_timeout directly if you need to know whether any tasks panicked or had to be force-aborted.

pub async fn stop_with_timeout( &mut self, timeout: Duration, ) -> QueueResult<StopOutcome>

Stop the worker, waiting up to timeout for in-flight jobs to finish gracefully before forcibly aborting any worker tasks still running.

Setting running to false only signals the spawned tasks to stop – each task only observes the flag at the top of its while *running.read().await loop (see start), so a task currently inside handler(job.clone()).await keeps running until that call returns (or times out) and it loops back around. This method waits for that to happen naturally, up to timeout, so the handler gets a chance to reach queue.complete()/queue.fail() instead of being cancelled mid-flight and leaving the job orphaned “Processing” in Redis forever. Only once timeout elapses are any still-running tasks forcibly aborted as a last resort – at that point whatever job they were mid-handler on is still orphaned, exactly as the previous unconditional-abort behavior left it.

Returns a StopOutcome reporting how many worker tasks finished gracefully, how many panicked mid-handler during the drain (logged at error level unconditionally when it happens), and how many had to be force-aborted after the grace period elapsed (logged at warn level unconditionally when it happens). A panicked worker task is not treated the same as a normal completion: a JoinError from join_next() means the task’s handler crashed, not that it returned Ok(()), and the in-flight job it was processing is left orphaned “Processing” with no other signal unless this is observed.

Examples

use armature_queue::*;
use std::time::Duration;

// Give in-flight jobs up to 10 seconds to finish before force-killing them.
let outcome = worker.stop_with_timeout(Duration::from_secs(10)).await?;
if outcome.panicked > 0 || outcome.force_aborted > 0 {
    eprintln!("shutdown was not fully graceful: {:?}", outcome);
}

pub async fn is_running(&self) -> bool

Check if the worker is running.