class Amber::Jobs::Worker

Overview

Executes background jobs by polling the queue adapter.

A Worker runs in a background fiber, repeatedly checking configured queues for available jobs. When a job is found, it deserializes and executes it, handling retries and dead job marking on failure.

Usage

worker = Amber::Jobs::Worker.new(
  adapter: Amber::Jobs.adapter,
  list_of_queues: ["default", "critical"],
  polling_interval: 1.second
)
worker.start

Work Stealing

When idle_only is set to true, the worker will only process jobs when there are no pending HTTP requests. This allows idle web server instances to contribute to job processing without impacting request latency.

Defined in:

amber/jobs/worker.cr

Constant Summary

Log = ::Log.for(self)

Constructors

Class Method Summary

Instance Method Summary

Constructor Detail

def self.new(adapter : QueueAdapter, list_of_queues : Array(String) = ["default"], polling_interval : Time::Span = 1.second, idle_only : Bool = false) #

[View source]

Class Method Detail

def self.pending_request_count : Int64 #

Tracks pending HTTP request count for work-stealing mode. This is incremented/decremented by the HTTP server middleware.


[View source]
def self.pending_request_count=(pending_request_count : Int64) #

Tracks pending HTTP request count for work-stealing mode. This is incremented/decremented by the HTTP server middleware.


[View source]
def self.pending_request_mutex : Mutex #

[View source]
def self.pending_request_mutex=(pending_request_mutex : Mutex) #

[View source]

Instance Method Detail

def adapter : QueueAdapter #

[View source]
def idle_only? : Bool #

[View source]
def is_running? : Bool #

[View source]
def jobs_processed : Int64 #

[View source]
def list_of_queues : Array(String) #

[View source]
def polling_interval : Time::Span #

[View source]
def process_next_job : Bool #

Attempts to dequeue and execute a single job from the configured queues.

This method is exposed publicly for testing purposes.


[View source]
def start : Nil #

Starts the worker in a background fiber.

The worker will poll all configured queues in order, executing any available jobs. This is a no-op if the worker is already running.


[View source]
def stop : Nil #

Stops the worker.

The worker will finish processing its current job (if any), complete its current sleep cycle, and then stop.


[View source]