Class: Sidekiq::DeferredJobs::Jobs

Inherits:
Object
  • Object
show all
Defined in:
lib/sidekiq/deferred_jobs.rb

Overview

Class for holding deferred jobs.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Jobs

Returns a new instance of Jobs.



250
251
252
253
 
# File 'lib/sidekiq/deferred_jobs.rb', line 250

def initialize
  @jobs = []
  @dedup_keys = Set.new
end

Instance Method Details

#clear!(filters = nil) ⇒ Array<Sidekiq::DeferredJobs::Job>

Clear any deferred jobs that match the filter.

Parameters:

  • filters (Array<Module>, Array<Hash>) (defaults to: nil) —

    Filter for jobs to clear

Returns:



269
270
271
272
273
274
 
# File 'lib/sidekiq/deferred_jobs.rb', line 269

def clear!(filters = nil)
  filter = Filter.new(filters)
  cleared_jobs = @jobs.select { |job| filter.match?(job.klass, job.opts) }
  @jobs -= cleared_jobs
  cleared_jobs
end

#defer(klass, args, opts = nil) ⇒ void

This method returns an undefined value.

Add a job to the deferred job list.

Parameters:

  • klass (Class) —

    Sidekiq worker class.

  • args (Array) —

    Sidekiq job arguments

  • opts (Hash, Nil) (defaults to: nil) —

    optional runtime jobs options



261
262
263
 
# File 'lib/sidekiq/deferred_jobs.rb', line 261

def defer(klass, args, opts = nil)
  @jobs << Job.new(klass, args&.dup, opts&.dup)
end

#enqueue!(filters = nil) ⇒ void

This method returns an undefined value.

Enqueue any deferred jobs that match the filter.

If a worker declares a uniqueness constraint (either with Sidekiq Enterprise or the sidekiq-unique-jobs gem), then duplicate jobs with the same class, arguments, and queue are collapsed into a single job within this call.

If enqueuing a job raises an error, the error will be propagated to the caller and any jobs that have not yet been enqueued will be retained in the deferred job list so they are not silently lost and can be enqueued again. The keys of unique jobs that were already enqueued are remembered until a call completes without an error so that retrying does not enqueue duplicates.

Parameters:

  • filters (Array<Module>, Array<Hash>) (defaults to: nil) —

    Filter for jobs to enqueue



290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
 
# File 'lib/sidekiq/deferred_jobs.rb', line 290

def enqueue!(filters = nil)
  filter = Filter.new(filters)
  unmatched_jobs = []
  pending_jobs = @jobs
  begin
    until pending_jobs.empty?
      job = pending_jobs.first
      if filter.match?(job.klass, job.opts)
        options = Sidekiq::DeferredJobs.worker_options(job.klass, job.opts)
        dedup_key = nil
        dedup_key = [job.klass, job.args, options["queue"]] if unique_job?(options)
        unless dedup_key && @dedup_keys.include?(dedup_key)
          Sidekiq::DeferredJobs.enqueue_job(job.klass, job.args, job.opts)
        end
        @dedup_keys << dedup_key if dedup_key
      else
        unmatched_jobs << job
      end
      pending_jobs.shift
    end
  ensure
    @jobs = unmatched_jobs + pending_jobs
    @dedup_keys.clear if pending_jobs.empty?
  end
end