Class: Fractor::PersistentWorkQueue

Inherits:
WorkQueue
  • Object
show all
Defined in:
lib/fractor/persistent_work_queue.rb

Overview

A work queue with persistence support. Automatically saves queue state to disk for crash recovery.

Examples:

Basic usage

queue = PersistentWorkQueue.new("data/queue.json")
queue << work_item  # Automatically saved

With custom persister

persister = QueuePersister::YAMLPersister.new("data/queue.yml")
queue = PersistentWorkQueue.new(persister: persister)

Instance Attribute Summary collapse

Attributes inherited from WorkQueue

#queue

Instance Method Summary collapse

Methods inherited from WorkQueue

#dequeue, #empty?, #pop_batch, #register_with_supervisor, #size

Constructor Details

#initialize(path_or_persister = nil, auto_save: true, save_interval: nil) ⇒ PersistentWorkQueue

Initialize a persistent work queue.

Parameters:

  • path_or_persister (String, QueuePersister::Base) (defaults to: nil) —

    File path or persister instance

  • auto_save (Boolean) (defaults to: true) —

    Automatically save after each enqueue

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

    Seconds between auto-saves (nil = disabled)



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/fractor/persistent_work_queue.rb', line 26

def initialize(path_or_persister = nil, auto_save: true, save_interval: nil)
  super()

  @persister = case path_or_persister
               when String
                 QueuePersister::JSONPersister.new(path_or_persister)
               when QueuePersister::Base, nil
                 path_or_persister
               else
                 raise ArgumentError,
                       "path_or_persister must be a String or QueuePersister::Base"
               end

  @auto_save = @persister ? auto_save : false
  @save_interval = save_interval
  @dirty = false
  @save_thread = nil

  # Start auto-save thread if interval is specified
  start_save_thread if @save_interval
end

Instance Attribute Details

#persister ⇒ Object (readonly)

Returns the value of attribute persister.



19
20
21
 
# File 'lib/fractor/persistent_work_queue.rb', line 19

def persister
  @persister
end

Instance Method Details

#<<(work_item) ⇒ void

This method returns an undefined value.

Alias for enqueue

Parameters:



62
63
64
 
# File 'lib/fractor/persistent_work_queue.rb', line 62

def <<(work_item)
  enqueue(work_item)
end

#clear ⇒ Boolean

Clear the queue and remove persisted data.

Returns:

  • (Boolean) —

    true if cleared successfully



125
126
127
128
129
130
131
132
133
 
# File 'lib/fractor/persistent_work_queue.rb', line 125

def clear
  @queue.clear
  @persister&.clear
  @dirty = false
  true
rescue StandardError => e
  warn "Failed to clear queue: #{e.message}"
  false
end

#close ⇒ void

This method returns an undefined value.

Save and close the queue, cleaning up resources.



145
146
147
148
 
# File 'lib/fractor/persistent_work_queue.rb', line 145

def close
  save if @dirty
  stop_save_thread if @save_thread
end

#dequeue_batch(max_items = 10) ⇒ Array<Fractor::Work>

Retrieve multiple work items from the queue. Marks queue as dirty for persistence.

Parameters:

  • max_items (Integer) (defaults to: 10) —

    Maximum number of items to retrieve

Returns:



71
72
73
74
75
 
# File 'lib/fractor/persistent_work_queue.rb', line 71

def dequeue_batch(max_items = 10)
  items = super
  @dirty = true if items.any?
  items
end

#dirty? ⇒ Boolean

Check if the queue has unsaved changes.

Returns:

  • (Boolean) —

    true if there are unsaved changes



138
139
140
 
# File 'lib/fractor/persistent_work_queue.rb', line 138

def dirty?
  @dirty
end

#enqueue(work_item) ⇒ void

This method returns an undefined value.

Add a work item to the queue with automatic persistence.

Parameters:



52
53
54
55
56
 
# File 'lib/fractor/persistent_work_queue.rb', line 52

def enqueue(work_item)
  super
  @dirty = true
  save if @auto_save
end

#load ⇒ Integer

Load queue state from disk, restoring previous work items.

Returns:

  • (Integer) —

    Number of items restored



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/fractor/persistent_work_queue.rb', line 96

def load
  return 0 unless @persister

  items = @persister.load
  return 0 unless items

  count = 0
  items.each do |item|
    # Convert hash back to Work object if needed
    work = if item.is_a?(Hash)
             deserialize_work(item)
           else
             item
           end

    @queue << work if work.is_a?(Fractor::Work)
    count += 1
  end

  @dirty = false
  count
rescue StandardError => e
  warn "Failed to load queue: #{e.message}"
  0
end

#save ⇒ Boolean

Save the current queue state to disk.

Returns:

  • (Boolean) —

    true if saved successfully



80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/fractor/persistent_work_queue.rb', line 80

def save
  return false unless @persister

  # Get all items from the queue without removing them
  items = peek_all
  @persister.save(items)
  @dirty = false
  true
rescue StandardError => e
  warn "Failed to save queue: #{e.message}"
  false
end