class
Amber::Jobs::MemoryQueueAdapter
- Amber::Jobs::MemoryQueueAdapter
- Amber::Jobs::QueueAdapter
- Reference
- Object
Overview
In-memory implementation of QueueAdapter.
This adapter stores jobs in memory using Mutex-protected data structures and provides scheduled job support via a sorted collection checked during dequeue. This is the default queue adapter and is suitable for development, testing, and single-instance applications.
Note: Job data will be lost when the application restarts since everything is stored in memory. For production applications that require persistence or multi-instance deployments, consider using a Redis-based or database adapter.
Thread Safety
All operations are protected by a Mutex to ensure safe concurrent access from multiple fibers.
Usage
adapter = Amber::Jobs::MemoryQueueAdapter.new
adapter.enqueue(envelope)
job = adapter.dequeue("default")Defined in:
amber/jobs/memory_queue_adapter.crConstructors
Instance Method Summary
-
#all_jobs : Array(JobEnvelope)
Returns all tracked job envelopes across all states.
-
#clear(queue : String) : Nil
Removes all jobs from the specified queue.
-
#close : Nil
Closes the adapter and cleans up resources.
-
#completed_size : Int32
Returns the number of completed jobs.
-
#dead_jobs : Array(JobEnvelope)
Returns all jobs that have been marked as dead (exceeded max retries).
-
#dead_size : Int32
Returns the number of dead jobs.
-
#dequeue(queue : String) : JobEnvelope | Nil
Removes and returns the next ready job from the specified queue.
-
#enqueue(envelope : JobEnvelope) : Nil
Adds a job envelope to the appropriate queue.
-
#failed_size : Int32
Returns the number of failed jobs.
-
#mark_completed(id : String) : Nil
Marks a job as completed by its ID.
-
#mark_dead(id : String, error : String) : Nil
Marks a job as dead.
-
#mark_failed(id : String, error : String) : Nil
Marks a job as failed by its ID, recording the error message.
-
#retry_failed(id : String) : Nil
Re-enqueues a failed job for retry by its ID.
-
#schedule(envelope : JobEnvelope, at : Time) : Nil
Schedules a job envelope for execution at a specific time.
-
#scheduled_size : Int32
Returns the number of scheduled jobs across all queues.
-
#size(queue : String) : Int32
Returns the number of pending jobs in the specified queue.
Instance methods inherited from class Amber::Jobs::QueueAdapter
all_jobs : Array(JobEnvelope)
all_jobs,
clear(queue : String) : Nil
clear,
close : Nil
close,
dead_jobs : Array(JobEnvelope)
dead_jobs,
dequeue(queue : String) : JobEnvelope | Nil
dequeue,
enqueue(envelope : JobEnvelope) : Nil
enqueue,
healthy? : Bool
healthy?,
mark_completed(id : String) : Nil
mark_completed,
mark_failed(id : String, error : String) : Nil
mark_failed,
retry_failed(id : String) : Nil
retry_failed,
schedule(envelope : JobEnvelope, at : Time) : Nil
schedule,
size(queue : String) : Int32
size
Constructor Detail
Instance Method Detail
Returns all jobs that have been marked as dead (exceeded max retries).
Removes and returns the next ready job from the specified queue.
Before checking the immediate queue, this method promotes any scheduled
jobs whose scheduled_at time has passed into their respective queues.
Returns nil if no jobs are ready.
Adds a job envelope to the appropriate queue.
If the envelope's scheduled_at is in the future, it is placed in the
scheduled set instead of the immediate queue. Otherwise, it is appended
to the back of the named queue.
Marks a job as dead. Called when a job has exceeded its max retries.
Marks a job as failed by its ID, recording the error message.
Re-enqueues a failed job for retry by its ID.
Resets the job's status to Pending and places it back in the immediate queue. If the job is not found in the failed set, this is a no-op.
Schedules a job envelope for execution at a specific time.
Updates the envelope's scheduled_at and places it in the scheduled set.
Returns the number of pending jobs in the specified queue.
This counts only jobs in the immediate queue, not scheduled jobs.