class Delayed::Heartbeat::Worker

Public Class Methods

dead_workers(timeout_seconds) click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 43
def self.dead_workers(timeout_seconds)
  where('last_heartbeat_at < ?', Time.now.utc - timeout_seconds)
end
delete_workers(workers) click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 51
def self.delete_workers(workers)
  where(id: workers.map(&:id)).delete_all if workers.present?
end
workers_with_different_version(current_version) click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 47
def self.workers_with_different_version(current_version)
  where('version != ?', current_version)
end

Public Instance Methods

job() click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 25
def job
  jobs.first
end
jobs() click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 21
def jobs
  Delayed::Job.where(locked_by: name, failed_at: nil)
end
unlock_jobs(mark_attempt_failed: true) click to toggle source

Returns the jobs that were unlocked

# File lib/delayed/heartbeat/worker.rb, line 30
def unlock_jobs(mark_attempt_failed: true)
  orphaned_jobs = jobs.to_a
  return orphaned_jobs unless orphaned_jobs.present?

  if mark_attempt_failed
    mark_job_attempts_failed(orphaned_jobs)
  else
    Delayed::Job.where(id: orphaned_jobs.map(&:id)).update_all(locked_at: nil, locked_by: nil)
  end

  orphaned_jobs
end

Private Instance Methods

mark_job_attempt_failed(dj_worker, job) click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 64
def mark_job_attempt_failed(dj_worker, job)
  # If there are more attempts this reschedules the job otherwise marks it as failed
  # and runs appropriate callbacks
  dj_worker.reschedule(job)
end
mark_job_attempts_failed(jobs) click to toggle source
# File lib/delayed/heartbeat/worker.rb, line 57
def mark_job_attempts_failed(jobs)
  dj_worker = Delayed::Worker.new
  jobs.each do |job|
    mark_job_attempt_failed(dj_worker, job)
  end
end