class

JoobQ::Queue(T)

Inherits JoobQ::BaseQueue < Reference < Object

The Queue class now focuses solely on queue operations

Constants

DISPATCH_START = Time.instant

PAUSE_CACHE_TTL = 500.milliseconds

Constructors

new(name : String, total_workers : Int32, throttle_limit : NamedTuple(limit: Int32, period: Time::Span) | Nil = nil, *, store : Store = ::JoobQ.config.store, handler : Proc(T, Nil) | Nil = nil)

Source

Instance methods

add(job : String)

Source

add(job : T)

Source

add!(job : String) : String

Strict variant of add: a payload that does not deserialize into this queue's job type raises (JSON::ParseException, JSON::SerializableError) instead of being logged, and the job id comes back. Store and connection errors raise as well. The job always lands on this queue: a queue in the payload is overridden, because the store pushes onto job.queue, which would otherwise let a payload pick any Redis key. add, add_batch and add_batch_strings still trust the queue in the payload, so do not feed them untrusted input.

Source

add_batch(jobs : Array(T))

Batch add jobs with pipelining for improved performance

Source

add_batch_strings(jobs : Array(String))

Add batch of job strings

Source

add_delayed(job : String, at : Time) : String

Schedules a serialized job to move onto the queue at at, and returns its job id. Raises like add! when the payload does not fit this queue's job type, and, like add!, ignores a queue in the payload. The delayed-job scheduler moves it when it is due.

Source

apply_concurrency(count : Int32)

Resizes this process only; used when following a size set elsewhere.

Source

claim_job(worker_id : String) : String | Nil

Source

claim_jobs_batch(worker_id : String, batch_size : Int32 = 5) : Array(String)

Source

claim_paused?

Source

cleanup_completed_job_pipelined(worker_id : String, job_json : String) : Nil

Enhanced cleanup for successfully completed jobs IMPORTANT: job_json must be the FULL job JSON string, not just the job ID

Source

cleanup_job_processing_pipelined(worker_id : String, job_json : String) : Nil

Pipelined job cleanup for improved performance IMPORTANT: job_json must be the FULL job JSON string, not just the job ID

Source

cleanup_jobs_batch_pipelined(worker_id : String, job_ids : Array(String)) : Nil

Batch job cleanup for improved performance

Source

clear

Source

delete_job(job : String)

Source

dispatching?

Source

job_type

Source

jobs(page_number : Int32 = 1, page_size : Int32 = 200)

Source

mark_as_dead(job : String)

Source

name

Source

next_job

Source

parse_job(job : String) : T

Source

pause!

Pausing is shared through the store, so it holds across processes.

Source

pause_claiming(duration : Time::Span) : Nil

Source

paused?

Source

perform(job : Job) : Nil

Override for explicitly injected execution dependencies; serialized jobs do not carry process-local store handles.

Source

processing_queue_size

Get count of jobs currently in processing queue

Source

release_job_claim(worker_id : String) : Nil

Source

release_job_claims_batch(worker_id : String, job_count : Int32) : Nil

Source

resize(count : Int32)

Changes the number of workers at runtime. The new size is stored so every process serving this queue follows it (see WorkerRegistry).

resume!

retry(job : String)

running?

running_workers

size

start

Starting a running queue does nothing. The first start recovers jobs stranded by a crashed process (and this process's own previous incarnation) back to the head of the queue, then publishes this process's liveness heartbeat before any worker claims, so another process's recovery can never reclaim a job this process holds.

status

stop!

store

take_claims(max : Int32) : Array(Tuple(String, Int64))

A worker's claim source in dispatcher mode: wait up to the block timeout for the first claim, then drain up to max total without blocking. Preserves FIFO (the dispatcher claims in order) and batch throughput under load.

Source

throttle_limit

Source

total_workers

Source

verify_job_removed_from_processing?(job_id : String) : Bool

Verify that a job has been properly removed from processing queue

Source

worker_manager

Source