Class: RailsAi::Agents::TaskQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/rails_ai/agents/task_queue.rb

Constant Summary collapse

PRIORITY_LEVELS = 
{
  critical: 4,
  high: 3,
  normal: 2,
  low: 1
}.freeze

Instance Method Summary collapse

Constructor Details

#initialize ⇒ TaskQueue

Returns a new instance of TaskQueue.



13
14
15
16
17
 
# File 'lib/rails_ai/agents/task_queue.rb', line 13

def initialize
  @tasks = []
  @total_processed = 0
  @mutex = Mutex.new
end

Instance Method Details

#clear! ⇒ Object 



65
66
67
68
 
# File 'lib/rails_ai/agents/task_queue.rb', line 65

def clear!
  @mutex.synchronize { @tasks.clear }
  defined?(Rails) && Rails.logger && Rails.logger.info("Task queue cleared")
end

#dequeue(timeout: nil) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/rails_ai/agents/task_queue.rb', line 37

def dequeue(timeout: nil)
  start_time = Time.now
  
  loop do
    @mutex.synchronize do
      return @tasks.shift if @tasks.any?
    end

    if timeout && (Time.now - start_time) > timeout
      return nil
    end

    sleep(0.1)
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


61
62
63
 
# File 'lib/rails_ai/agents/task_queue.rb', line 61

def empty?
  @mutex.synchronize { @tasks.empty? }
end

#enqueue(task, priority: :normal) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/rails_ai/agents/task_queue.rb', line 19

def enqueue(task, priority: :normal)
  task_with_priority = task.merge(
    id: task[:id] || SecureRandom.uuid,
    priority: priority,
    priority_score: PRIORITY_LEVELS[priority] || 2,
    enqueued_at: Time.now,
    status: :pending
  )

  @mutex.synchronize do
    @tasks << task_with_priority
    @tasks.sort_by! { |t| [-t[:priority_score], t[:enqueued_at]] }
  end

  defined?(Rails) && Rails.logger && Rails.logger.info("Task enqueued: #{task[:description]} (priority: #{priority})")
  task_with_priority
end

#get_tasks_by_priority(priority) ⇒ Object 



83
84
85
86
87
 
# File 'lib/rails_ai/agents/task_queue.rb', line 83

def get_tasks_by_priority(priority)
  @mutex.synchronize do
    @tasks.select { |t| t[:priority] == priority }
  end
end

#get_tasks_by_status(status) ⇒ Object 



77
78
79
80
81
 
# File 'lib/rails_ai/agents/task_queue.rb', line 77

def get_tasks_by_status(status)
  @mutex.synchronize do
    @tasks.select { |t| t[:status] == status }
  end
end

#mark_processed(task_id) ⇒ Object 



89
90
91
92
 
# File 'lib/rails_ai/agents/task_queue.rb', line 89

def mark_processed(task_id)
  @total_processed += 1
  defined?(Rails) && Rails.logger && Rails.logger.info("Task processed: #{task_id} (total: #{@total_processed})")
end

#peek ⇒ Object 



53
54
55
 
# File 'lib/rails_ai/agents/task_queue.rb', line 53

def peek
  @mutex.synchronize { @tasks.first }
end

#remove_task(task_id) ⇒ Object 



70
71
72
73
74
75
 
# File 'lib/rails_ai/agents/task_queue.rb', line 70

def remove_task(task_id)
  @mutex.synchronize do
    task = @tasks.find { |t| t[:id] == task_id }
    @tasks.delete(task) if task
  end
end

#size ⇒ Object 



57
58
59
 
# File 'lib/rails_ai/agents/task_queue.rb', line 57

def size
  @mutex.synchronize { @tasks.length }
end

#stats ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/rails_ai/agents/task_queue.rb', line 94

def stats
  @mutex.synchronize do
    {
      total_tasks: @tasks.length,
      total_processed: @total_processed,
      by_priority: @tasks.group_by { |t| t[:priority] }.transform_values(&:length),
      by_status: @tasks.group_by { |t| t[:status] }.transform_values(&:length),
      oldest_task: @tasks.min_by { |t| t[:enqueued_at] }&.dig(:enqueued_at)
    }
  end
end

#total_processed ⇒ Object 



106
107
108
 
# File 'lib/rails_ai/agents/task_queue.rb', line 106

def total_processed
  @total_processed
end