Class: Rage::Deferred::Backends::Disk

Inherits:
Object
  • Object
show all
Defined in:
lib/rage/deferred/backends/disk.rb

Overview

Rage::Deferred::Backends implements a storage layer to persist deferred tasks. A storage should implement the following instance methods:

  • add_task - called when a task has to be added to the storage;
  • remove_task - called when a task has to be removed from the storage;
  • pending_tasks - the method should iterate over the underlying storage and return a list of tasks to replay;
  • add_dead_task - called when a task has exhausted its retries or aborted them;
  • each_dead_task - yields one record for each task ID that exists when traversal starts; if multiple records have the same task ID, it yields the newest record; records added during traversal are not included;
  • find_dead_task - returns the newest dead-task record with the exact task ID;
  • remove_dead_tasks - permanently delete dead tasks;

Instance Method Summary collapse

Constructor Details

#initialize(path:, prefix:, fsync_frequency:) ⇒ Disk

Returns a new instance of Disk.



20
21
22
23
 
# File 'lib/rage/deferred/backends/disk.rb', line 20

def initialize(path:, prefix:, fsync_frequency:)
  @tasks_storage = TasksStorage.new(path:, prefix:, fsync_frequency:)
  @dead_tasks_storage = DeadTasksStorage.new(path:, prefix:)
end

Instance Method Details

#add_dead_task(task_id, context, exception, task_class:, attempts:) ⇒ String

Add a task to the dead-tasks store.

Parameters:

  • task_id (String) —

    the id the task was persisted with

  • context (Array) —

    the serializable execution context of the task

  • exception (Exception) —

    the exception raised during the last attempt

  • task_class (Class, String) —

    the class of the task

  • attempts (Integer) —

    the number of attempts made to process the task

Returns:

  • (String) —

    the id of the dead-task record

Raises:



54
55
56
 
# File 'lib/rage/deferred/backends/disk.rb', line 54

def add_dead_task(task_id, context, exception, task_class:, attempts:)
  @dead_tasks_storage.add(task_id, context, exception, task_class:, attempts:)
end

#add_task(task, publish_at: nil, task_id: nil) ⇒ String

Add a record to the log representing a new task.

Parameters:

  • task (Rage::Deferred::Task)
  • publish_at (Integer, nil) (defaults to: nil)
  • task_id (String, nil) (defaults to: nil)

Returns:

  • (String) 


30
31
32
 
# File 'lib/rage/deferred/backends/disk.rb', line 30

def add_task(task, publish_at: nil, task_id: nil)
  @tasks_storage.add(task, publish_at:, task_id:)
end

#pending_tasks ⇒ Array<(String, Rage::Deferred::Task, Integer, Integer)>

Return a list of pending tasks in the storage.

Returns:

  • (Array<(String, Rage::Deferred::Task, Integer, Integer)>) —

    Array<(String, Rage::Deferred::Task, Integer, Integer)>



42
43
44
 
# File 'lib/rage/deferred/backends/disk.rb', line 42

def pending_tasks
  @tasks_storage.pending_tasks
end

#remove_dead_tasks(ids) ⇒ Integer

Permanently delete dead-lettered tasks.

Parameters:

  • ids (String, Array<String>) —

    the ids of the dead-task records

Returns:

  • (Integer) —

    the number of deleted records

Raises:



81
82
83
 
# File 'lib/rage/deferred/backends/disk.rb', line 81

def remove_dead_tasks(ids)
  @dead_tasks_storage.remove(ids)
end

#remove_task(task_id) ⇒ Object

Add a record to the log representing a task removal.

Parameters:

  • task_id (String) 


36
37
38
 
# File 'lib/rage/deferred/backends/disk.rb', line 36

def remove_task(task_id)
  @tasks_storage.remove(task_id)
end