Class: ActiveWorker::JobQueue::QueueManager

Inherits:
Object
  • Object
show all
Defined in:
lib/active_worker/job_queue/queue_manager.rb

Instance Method Summary collapse

Instance Method Details

#active_jobs ⇒ Object 



5
6
7
 
# File 'lib/active_worker/job_queue/queue_manager.rb', line 5

def active_jobs
  Resque.working.map {|w| create_job_hash_from_worker(w)}.compact
end

#active_jobs_for_configurations(configuration_ids) ⇒ Object 



9
10
11
12
13
14
15
16
 
# File 'lib/active_worker/job_queue/queue_manager.rb', line 9

def active_jobs_for_configurations(configuration_ids)
  workers = Resque.working.select do|w|
    configuration_ids.include? configuration_id_from_worker(w)
  end
  workers.map do |w|
    create_job_hash_from_worker(w)
  end
end

#args_from_worker(worker) ⇒ Object 



40
41
42
43
 
# File 'lib/active_worker/job_queue/queue_manager.rb', line 40

def args_from_worker(worker)
  payload = worker.job["payload"]
  payload["args"].first if payload
end

#configuration_id_from_worker(worker) ⇒ Object 



18
19
20
21
 
# File 'lib/active_worker/job_queue/queue_manager.rb', line 18

def configuration_id_from_worker(worker)
  params = params_from_worker(worker)
  params.first if params
end

#create_job_hash_from_worker(worker) ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/active_worker/job_queue/queue_manager.rb', line 23

def create_job_hash_from_worker(worker)
  worker_id = worker.to_s.split(":")
  host = worker_id[0]
  pid = worker_id[1]
  queues = worker_id[2].split(",")
  args = args_from_worker(worker)

  if worker_id && host && pid && queues && args
    return {"host" => host, "queues" => queues, "pid" => pid, "args" => args }
  end
end

#params_from_worker(worker) ⇒ Object 



35
36
37
38
 
# File 'lib/active_worker/job_queue/queue_manager.rb', line 35

def params_from_worker(worker)
  args = args_from_worker(worker)
  args["params"] if args
end