From 05ac3617002579bcd9b8f3b3aa5c099f929d0b04 Mon Sep 17 00:00:00 2001 From: Dan Buch Date: Tue, 18 Feb 2014 00:08:24 -0500 Subject: [PATCH] Extracting helpers and server methods into modules both for complexity reasons and so that RuboCop can properly analyze the methods. --- lib/resque/scheduler/server.rb | 304 +++++++++++++++++---------------- 1 file changed, 161 insertions(+), 143 deletions(-) diff --git a/lib/resque/scheduler/server.rb b/lib/resque/scheduler/server.rb index 4d9f9954..dd01b332 100644 --- a/lib/resque/scheduler/server.rb +++ b/lib/resque/scheduler/server.rb @@ -9,181 +9,199 @@ module Scheduler module Server def self.included(base) base.class_eval do - helpers do - def format_time(t) - t.strftime('%Y-%m-%d %H:%M:%S %z') - end + helpers { include HelperMethods } + include ServerMethods - def queue_from_class_name(class_name) - Resque.queue_from_class( - Resque::Scheduler::Util.constantize(class_name) - ) - end + get('/schedule') { schedule } + post('/schedule/requeue') { schedule_requeue } + post('/schedule/requeue_with_params') do + schedule_requeue_with_params + end + get('/delayed') { delayed } + get('/delayed/jobs/:klass') { delayed_jobs_klass } + post('/delayed/search') { delayed_search } + get('/delayed/:timestamp') { delayed_timestamp } + post('/delayed/queue_now') { delayed_queue_now } + post('/delayed/cancel_now') { delayed_cancel_now } + post('/delayed/clear') { delayed_clear } + end + end - def find_job(worker) - worker = worker.downcase - results = Array.new + module ServerMethods + def schedule + Resque.reload_schedule! if Resque::Scheduler.dynamic + erb scheduler_template('scheduler') + end - # Check working jobs - working = [*Resque.working] - work = working.select do |w| - w.job && w.job['payload'] && - w.job['payload']['class'].downcase.include?(worker) - end - work.each do |w| - results += [ - w.job['payload'].merge( - 'queue' => w.job['queue'], 'where_at' => 'working' - ) - ] - end + def schedule_requeue + @job_name = params['job_name'] || params[:job_name] + config = Resque.schedule[@job_name] + @parameters = config['parameters'] || config[:parameters] + if @parameters + erb scheduler_template('requeue-params') + else + Resque::Scheduler.enqueue_from_config(config) + redirect u('/overview') + end + end - # Check delayed Jobs - dels = Array.new - schedule_size = Resque.delayed_queue_schedule_size - Resque.delayed_queue_peek(0, schedule_size).each do |d| - Resque.delayed_timestamp_peek( - d, 0, Resque.delayed_timestamp_size(d)).each do |j| - dels << j.merge!('timestamp' => d) - end - end - results += dels.select do |j| - j['class'].downcase.include?(worker) && - j.merge!('where_at' => 'delayed') - end + def schedule_requeue_with_params + job_name = params['job_name'] || params[:job_name] + config = Resque.schedule[job_name] + # Build args hash from post data (removing the job name) + submitted_args = params.reject do |key, value| + key == 'job_name' || key == :job_name + end - # Check Queues - Resque.queues.each do |queue| - queued = Resque.peek(queue, 0, Resque.size(queue)) - queued = [queued] unless queued.is_a?(Array) - results += queued.select do |j| - j['class'].downcase.include?(worker) && - j.merge!('queue' => queue, 'where_at' => 'queued') - end - end - results - end + # Merge constructed args hash with existing args hash for + # the job, if it exists + config_args = config['args'] || config[:args] || {} + config_args = config_args.merge(submitted_args) - def schedule_interval(config) - if config['every'] - schedule_interval_every(config['every']) - elsif config['cron'] - 'cron: ' + config['cron'].to_s - else - 'Not currently scheduled' - end - end + # Insert the args hash into config and queue the resque job + config = config.merge('args' => config_args) + Resque::Scheduler.enqueue_from_config(config) + redirect u('/overview') + end - def schedule_interval_every(every) - every = [*every] - s = 'every: ' << every.first + def delayed + erb scheduler_template('delayed') + end - return s unless every.length > 1 + def delayed_jobs_klass + begin + klass = Resque::Scheduler::Util.constantize(params[:klass]) + @args = JSON.load(URI.decode(params[:args])) + @timestamps = Resque.scheduled_at(klass, *@args) + rescue + @timestamps = [] + end - s << ' (' - meta = every.last.map do |key, value| - "#{key.to_s.gsub(/_/, ' ')} #{value}" - end - s << meta.join(', ') << ')' - end + erb scheduler_template('delayed_schedules') + end - def schedule_class(config) - if config['class'].nil? && !config['custom_job_class'].nil? - config['custom_job_class'] - else - config['class'] - end - end + def delayed_search + @jobs = find_job(params[:search]) + erb scheduler_template('search') + end - def scheduler_template(name) - File.read( - File.expand_path("../server/views/#{name}.erb", __FILE__) - ) - end - end + def delayed_timestamp + erb scheduler_template('delayed_timestamp') + end - get '/schedule' do - Resque.reload_schedule! if Resque::Scheduler.dynamic - erb scheduler_template('scheduler') + def delayed_queue_now + timestamp = params['timestamp'].to_i + if timestamp > 0 + Resque::Scheduler.enqueue_delayed_items_for_timestamp(timestamp) end + redirect u('/overview') + end - post '/schedule/requeue' do - @job_name = params['job_name'] || params[:job_name] - config = Resque.schedule[@job_name] - @parameters = config['parameters'] || config[:parameters] - if @parameters - erb scheduler_template('requeue-params') - else - Resque::Scheduler.enqueue_from_config(config) - redirect u('/overview') - end - end + def delayed_cancel_now + klass = Resque::Scheduler::Util.constantize(params['klass']) + timestamp = params['timestamp'] + args = Resque.decode params['args'] + Resque.remove_delayed_job_from_timestamp(timestamp, klass, *args) + redirect u('/delayed') + end - post '/schedule/requeue_with_params' do - job_name = params['job_name'] || params[:job_name] - config = Resque.schedule[job_name] - # Build args hash from post data (removing the job name) - submitted_args = params.reject do |key, value| - key == 'job_name' || key == :job_name - end + def delayed_clear + Resque.reset_delayed_queue + redirect u('delayed') + end + end - # Merge constructed args hash with existing args hash for - # the job, if it exists - config_args = config['args'] || config[:args] || {} - config_args = config_args.merge(submitted_args) + module HelperMethods + def format_time(t) + t.strftime('%Y-%m-%d %H:%M:%S %z') + end - # Insert the args hash into config and queue the resque job - config = config.merge('args' => config_args) - Resque::Scheduler.enqueue_from_config(config) - redirect u('/overview') + def queue_from_class_name(class_name) + Resque.queue_from_class( + Resque::Scheduler::Util.constantize(class_name) + ) + end + + def find_job(worker) + worker = worker.downcase + results = [] + + # Check working jobs + working = [*Resque.working] + work = working.select do |w| + w.job && w.job['payload'] && + w.job['payload']['class'].downcase.include?(worker) + end + work.each do |w| + results += [ + w.job['payload'].merge( + 'queue' => w.job['queue'], 'where_at' => 'working' + ) + ] end - get '/delayed' do - erb scheduler_template('delayed') + # Check delayed Jobs + dels = [] + schedule_size = Resque.delayed_queue_schedule_size + Resque.delayed_queue_peek(0, schedule_size).each do |d| + Resque.delayed_timestamp_peek( + d, 0, Resque.delayed_timestamp_size(d)).each do |j| + dels << j.merge!('timestamp' => d) + end + end + results += dels.select do |j| + j['class'].downcase.include?(worker) && + j.merge!('where_at' => 'delayed') end - get '/delayed/jobs/:klass' do - begin - klass = Resque::Scheduler::Util.constantize(params[:klass]) - @args = JSON.load(URI.decode(params[:args])) - @timestamps = Resque.scheduled_at(klass, *@args) - rescue - @timestamps = [] + # Check Queues + Resque.queues.each do |queue| + queued = Resque.peek(queue, 0, Resque.size(queue)) + queued = [queued] unless queued.is_a?(Array) + results += queued.select do |j| + j['class'].downcase.include?(worker) && + j.merge!('queue' => queue, 'where_at' => 'queued') end - - erb scheduler_template('delayed_schedules') end + results + end - post '/delayed/search' do - @jobs = find_job(params[:search]) - erb scheduler_template('search') + def schedule_interval(config) + if config['every'] + schedule_interval_every(config['every']) + elsif config['cron'] + 'cron: ' + config['cron'].to_s + else + 'Not currently scheduled' end + end - get '/delayed/:timestamp' do - erb scheduler_template('delayed_timestamp') - end + def schedule_interval_every(every) + every = [*every] + s = 'every: ' << every.first - post '/delayed/queue_now' do - timestamp = params['timestamp'].to_i - if timestamp > 0 - Resque::Scheduler.enqueue_delayed_items_for_timestamp(timestamp) - end - redirect u('/overview') - end + return s unless every.length > 1 - post '/delayed/cancel_now' do - klass = Resque::Scheduler::Util.constantize(params['klass']) - timestamp = params['timestamp'] - args = Resque.decode params['args'] - Resque.remove_delayed_job_from_timestamp(timestamp, klass, *args) - redirect u('/delayed') + s << ' (' + meta = every.last.map do |key, value| + "#{key.to_s.gsub(/_/, ' ')} #{value}" end + s << meta.join(', ') << ')' + end - post '/delayed/clear' do - Resque.reset_delayed_queue - redirect u('delayed') + def schedule_class(config) + if config['class'].nil? && !config['custom_job_class'].nil? + config['custom_job_class'] + else + config['class'] end end + + def scheduler_template(name) + File.read( + File.expand_path("../server/views/#{name}.erb", __FILE__) + ) + end end end end