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;
  • list_dead_tasks - return a list of dead tasks, newest first;
  • find_dead_task - return a single dead task;
  • 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.



18
19
20
21
 
# File 'lib/rage/deferred/backends/disk.rb', line 18

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:



52
53
54
 
# File 'lib/rage/deferred/backends/disk.rb', line 52

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) 


28
29
30
 
# File 'lib/rage/deferred/backends/disk.rb', line 28

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

#find_dead_task(id) ⇒ Hash?

Return a single dead-lettered task.

Parameters:

  • id (String) —

    the id of the dead-task record

Returns:

  • (Hash, nil) 


67
68
69
 
# File 'lib/rage/deferred/backends/disk.rb', line 67

def find_dead_task(id)
  @dead_tasks_storage.find(id)
end

#list_dead_tasks(limit: nil, offset: 0) ⇒ Array<Hash>

Return a list of dead-lettered tasks, newest first.

Parameters:

  • limit (Integer, nil) (defaults to: nil) —

    the maximum number of records to return

  • offset (Integer) (defaults to: 0) —

    the number of records to skip

Returns:

  • (Array<Hash>) 


60
61
62
 
# File 'lib/rage/deferred/backends/disk.rb', line 60

def list_dead_tasks(limit: nil, offset: 0)
  @dead_tasks_storage.list(limit:, offset:)
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)>



40
41
42
 
# File 'lib/rage/deferred/backends/disk.rb', line 40

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:



75
76
77
 
# File 'lib/rage/deferred/backends/disk.rb', line 75

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) 


34
35
36
 
# File 'lib/rage/deferred/backends/disk.rb', line 34

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