require 'iron_mq' require 'fanforce/api' require 'timeout' class Fanforce::Worker::Runner include Fanforce::Worker::Utils MAX_EXECUTION_TIME = 3300 class Timeout < RuntimeError; end def initialize(worker_data, min_execution_time=300, &code_block) raise "min_execution_time was set to #{min_execution_time}, which is #{min_execution_time - MAX_EXECUTION_TIME} seconds too long" if min_execution_time > MAX_EXECUTION_TIME log.debug 'LOADING WORKER ENV' @queue_id = worker_data['queue_id'] || (raise 'worker_data must contain queue_id') @worker_env = worker_data['env_vars'] || {} @min_execution_time = min_execution_time @code_block = code_block load_env load_jobs end def load_jobs log.debug 'PROCESSING JOBS...' log.debug '------------------------------------------------------------------------------------' job_num = 0 job_data = nil while job_has_enough_time_to_run and (job = Fanforce::Worker.iron_mq.queue(@queue_id).get(timeout: 3600)) do log.debug "- JOB #{job_num+=1}: #{job.body}" timeout(worker_time_remaining, Timeout) do job_data = nil job_data = Fanforce.decode_json(job.body) run_job(job, job_data, &@code_block) end delete_job log.debug '------------------------------------------------------------------------------------' end delete_job log.debug 'WINDING DOWN WORKER!' rescue Exception => e handle_job_loading_error(e, job, job_data) end def load_env if File.exists?('.developmentenv.rb') require '.developmentenv' elsif File.exists?('.stagingenv.rb') require '.stagingenv' elsif File.exists?('.productionenv.rb') require '.productionenv' end end def run_job(job, job_data, &code_block) @current_job = job @current_params = job_data[:params] @current_retries = job_data[:retries] set_env_vars(@worker_env) code_block.call(job_data[:params].clone, retries: job_data[:retries], queue_id: @queue_id) delete_job(job) rescue Exception => e handle_job_error(e, job, job_data) end def handle_job_loading_error(e, job, job_data) raise($!, "#{$!}: THERE IS NO JOB", $!.backtrace) if job.nil? delete_job(job) log.debug 'REMOVED JOB FROM QUEUE, BUT COULD NOT SAVE TO ERROR CACHE...' raise($!, "#{$!}: #{job_data.to_json}", $!.backtrace) end def handle_job_error(e, job, job_data) raise($!, "#{$!}: THERE IS NO JOB", $!.backtrace) if job.nil? delete_job(job) require_relative 'errors' log.debug 'REMOVED JOB FROM QUEUE, AND SAVING TO ERROR CACHE...' Fanforce::Worker::Errors.add(@queue_id, e, job_data, @worker_env) end def worker_time_remaining time_since_load = Time.now - Fanforce::Worker::LOADED_AT MAX_EXECUTION_TIME - time_since_load end def job_has_enough_time_to_run time_since_load = Time.now - Fanforce::Worker::LOADED_AT return false if time_since_load > MAX_EXECUTION_TIME return false if worker_time_remaining < @min_execution_time return true end def set_env_vars(vars) vars.each {|k,v| ENV[k.to_s]=v } end def delete_job(job=nil) return if job.nil? and @current_job.nil? (job || @current_job).delete @current_job = nil end end