IronWorker Requeuing Jobs | Ruby script.

iron_requeuing_jobs.rb

#=============================================================================     
#title           :iron_requeuing_jobs.rb
#description     :This script is requeing jobs by job status to Iron Worker
#author		 :Quentin Rousseau <[email protected]>
#date            :2014-07-23
#version         :1.0
#usage		 :ruby iron_requeuing_jobs.rb
#dependencies    :gem install iron_worker_ng
#==============================================================================

require 'iron_worker_ng'
  
CODE_NAME = 'YOUR_CODE_NAME'
IRON_TOKEN = 'YOUR_IRON_TOKEN'
IRON_PROJECT_ID = 'YOUR_IRON_PROJECT_ID'

# Feel free to select jobs status you want to requeue by setting the property to property = 1
queued = nil #in the queue, waiting to run
running	= nil #running
complete = nil #finished running
error = nil #error during processing
cancelled = nil #cancelled by user
killed	= nil #killed by system
timeout	= nil #exceeded processing time threshold

# Other properties
per_page = 100 #The number of tasks to return per page. Note this is a maximum value, so there may be less tasks returned if there aren’t enough results. Default is 30, maximum is 100.
job_max_timeout = 3600 #The maximum runtime of your task in seconds. No task can exceed 3600 seconds (60 minutes). The default is 3600 but can be set to a shorter duration.
job_priority = 0 #The priority queue to run the task in. Valid values are 0, 1, and 2. 0 is the default.
from_time =  nil #Limit the retrieved tasks to only those that were created after the time specified in the value. Time should be formatted as the number of seconds since the Unix epoch.
to_time = nil #Limit the retrieved tasks to only those that were created before the time specified in the value. Time should be formatted as the number of seconds since the Unix epoch.

# Initialisation
page_count = 0
@iron_client = IronWorkerNG::Client.new(token: IRON_TOKEN , project_id: IRON_PROJECT_ID)

# Helpers functions
def enqueue(code_name, args, priority = 0, timeout = 3600)
  if !already_in_queue?(code_name, args)
    task = @iron_client.tasks.create(code_name, args, priority: priority, timeout: timeout)
    puts "[REQUEUED] Job with args : #{args}"
  else
    puts "[ERROR] Job with args : #{args} already queued or running"
  end
end

def already_in_queue?(code_name, args)
  @iron_client.tasks.list(per_page: 100, code_name: code_name, running: 1, queued: 1).any? { |n| n.payload == args }
end

# Fetching
tasks = @iron_client.tasks.list(code_name: CODE_NAME, page: page_count, from_time: from_time, to_time: to_time, per_page: per_page, queued: queued, running: running, complete: complete, error: error, cancelled: cancelled, killed: killed, timeout: timeout)

while(tasks and !tasks.empty?)
  tasks.each do |n|
    enqueue(CODE_NAME, n.payload, job_priority, job_max_timeout) if n.payload and !n.payload.empty?
  end
  
  page_count += 1
  puts "[FETCHING] PAGE #{page_count}"
  tasks = @iron_client.tasks.list(code_name: CODE_NAME, page: page_count, from_time: from_time, to_time: to_time, per_page: per_page, queued: queued, running: running, complete: complete, error: error, cancelled: cancelled, killed: killed, timeout: timeout)
end