Class: QueuePulse::Source::Fake

Inherits:
Source
  • Object
show all
Defined in:
lib/queue_pulse/source.rb

Overview

In-memory implementation for tests and local experiments.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(now: Time.now) ⇒ Fake

Returns a new instance of Fake.



63
64
65
66
67
68
69
70
71
 
# File 'lib/queue_pulse/source.rb', line 63

def initialize(now: Time.now)
  super()
  @available = true
  @clock = now
  @failed = []
  @queue_stats = []
  @claimed = []
  @processes = [] # array of heartbeat Times
end

Instance Attribute Details

#available ⇒ Object

Returns the value of attribute available.



61
62
63
 
# File 'lib/queue_pulse/source.rb', line 61

def available
  @available
end

#clock ⇒ Object

Returns the value of attribute clock.



61
62
63
 
# File 'lib/queue_pulse/source.rb', line 61

def clock
  @clock
end

Instance Method Details

#add_claimed(job_id:, job_class:, queue_name:, claimed_at:) ⇒ Object 



87
88
89
 
# File 'lib/queue_pulse/source.rb', line 87

def add_claimed(job_id:, job_class:, queue_name:, claimed_at:)
  @claimed << ClaimedRecord.new(job_id: job_id, job_class: job_class, queue_name: queue_name, claimed_at: claimed_at)
end

#add_failed(id:, job_class:, queue_name:, error_message:, failed_at: @clock) ⇒ Object

--- test setters --------------------------------------------------------



77
78
79
80
 
# File 'lib/queue_pulse/source.rb', line 77

def add_failed(id:, job_class:, queue_name:, error_message:, failed_at: @clock)
  @failed << FailedRecord.new(id: id, job_class: job_class, queue_name: queue_name,
                              error_message: error_message, failed_at: failed_at)
end

#add_process(last_heartbeat_at:) ⇒ Object 



91
92
93
 
# File 'lib/queue_pulse/source.rb', line 91

def add_process(last_heartbeat_at:)
  @processes << last_heartbeat_at
end

#available? ⇒ Boolean

Returns:

  • (Boolean) 


73
 
# File 'lib/queue_pulse/source.rb', line 73

def available? = @available

#failed_count_since(since_id) ⇒ Object 



104
105
106
 
# File 'lib/queue_pulse/source.rb', line 104

def failed_count_since(since_id)
  @failed.count { |f| f.id > since_id }
end

#failed_executions_since(since_id, limit:) ⇒ Object

--- Source interface ----------------------------------------------------



96
97
98
 
# File 'lib/queue_pulse/source.rb', line 96

def failed_executions_since(since_id, limit:)
  @failed.select { |f| f.id > since_id }.sort_by(&:id).first(limit)
end

#live_process_count(heartbeat_within:) ⇒ Object 



117
118
119
120
 
# File 'lib/queue_pulse/source.rb', line 117

def live_process_count(heartbeat_within:)
  cutoff = @clock - heartbeat_within
  @processes.count { |hb| hb >= cutoff }
end

#max_failed_execution_id ⇒ Object 



100
101
102
 
# File 'lib/queue_pulse/source.rb', line 100

def max_failed_execution_id
  @failed.map(&:id).max
end

#now ⇒ Object 



74
 
# File 'lib/queue_pulse/source.rb', line 74

def now = @clock

#queue_stats ⇒ Object 



108
109
110
 
# File 'lib/queue_pulse/source.rb', line 108

def queue_stats
  @queue_stats
end

#set_queue(queue_name, depth:, oldest_enqueued_at: nil) ⇒ Object 



82
83
84
85
 
# File 'lib/queue_pulse/source.rb', line 82

def set_queue(queue_name, depth:, oldest_enqueued_at: nil)
  @queue_stats.reject! { |q| q.queue_name == queue_name }
  @queue_stats << QueueStat.new(queue_name: queue_name, depth: depth, oldest_enqueued_at: oldest_enqueued_at)
end

#stuck_claimed_executions(older_than:) ⇒ Object 



112
113
114
115
 
# File 'lib/queue_pulse/source.rb', line 112

def stuck_claimed_executions(older_than:)
  cutoff = @clock - older_than
  @claimed.select { |c| c.claimed_at < cutoff }
end

#total_process_count ⇒ Object 



122
123
124
 
# File 'lib/queue_pulse/source.rb', line 122

def total_process_count
  @processes.size
end