process sis batches using account-specific strands, closes #4415

Change-Id: I69b41689aef047f5302955c27e94c76ece0236f4
Reviewed-on: https://gerrit.instructure.com/3688
Tested-by: Hudson <hudson@instructure.com>
Reviewed-by: JT Olds <jt@instructure.com>
Reviewed-by: Zach Wily <zach@instructure.com>

Author: codekitchen

app/models/sis_batch.rb

@@ -90,7 +90,7 @@ def process
     self.workflow_state = "failed"
     self.save
   end
-  handle_asynchronously_with_queue :process, 'sis_imports'
+  handle_asynchronously :process, :strand => proc { |sis_batch| "sis_batch:account:#{sis_batch.account_id}" }
 
   named_scope :needs_processing, lambda{
     {:conditions => ["sis_batches.workflow_state = 'needs_processing'"], :order => :created_at}

config/delayed_jobs.yml.example

@@ -5,13 +5,6 @@ production:
     max_priority: 10
   - queue: canvas_queue
     workers: 4
-    
-  # warning: don't set the worker pool size to be greater than 1 if you want
-  #   sis imports to be processed serially
-  - queue: sis_imports
-    workers: 1
-    
-  - periodic: config/periodic_jobs.rb
   max_run_time: 28800
   # if set, workers will process this many jobs and then die, causing the pool
   # to spawn another worker. this can help return memory to the OS.
@@ -23,4 +16,3 @@ production:
 default:
   workers:
   - queue: canvas_queue
-  - periodic: config/periodic_jobs.rb

vendor/plugins/delayed_job/lib/delayed/message_sending.rb

@@ -5,6 +5,16 @@ def send_later(method, *args)
     end
 
     def send_later_enqueue_args(method, enqueue_args = {}, *args)
+      # support procs/methods as enqueue arguments
+      duped = false
+      enqueue_args.each do |k,v|
+        if v.respond_to?(:call)
+          enqueue_args = enqueue_args.dup unless duped
+          duped = true
+          enqueue_args[k] = v.call(self)
+        end
+      end
+
       Delayed::Job.enqueue(Delayed::PerformableMethod.new(self, method.to_sym, args), enqueue_args)
     end