Module: Sidekiq::DeferredJobs

Defined in:
lib/sidekiq/deferred_jobs.rb

Overview

Adds the ability to defer enqueuing Sidekiq jobs until the end of a block of code. See Sidekiq::DeferredJobs::DeferBlock for the methods that are added to Sidekiq itself (Sidekiq.defer_jobs, Sidekiq.abort_deferred_jobs!, and Sidekiq.enqueue_deferred_jobs!).

Defined Under Namespace

Modules: DeferBlock, DeferredSetter, DeferredWorker Classes: Filter, Job, Jobs

Class Method Summary collapse

Class Method Details

.defer(filter, &block) ⇒ Object

Defer enqueuing Sidekiq workers within the block until the end of the block. Any workers that normally would have been enqueued with a perform_async call will instead be queued up and run in an ensure clause at the end of the block.

Blocks can be nested. Deferred jobs are only enqueued at the end of the outermost block and the filters from all of the enclosing blocks apply, so a job is deferred if it matches the filter on any block it is nested inside of.

Parameters:

  • filter (Array<Module>, Array<Hash>) —

    An array of either classes, modules, or hashes. If this is provided, only workers that match either a class or module or which have sidekiq_options that match a hash will be deferred. All other workers will be enqueued as normal.

Returns:

  • (Object) —

    the return value of the block



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/sidekiq/deferred_jobs.rb', line 25

def defer(filter, &block)
  new_filter = Filter.new(filter)
  jobs, filters = Thread.current[:sidekiq_deferred_jobs_jobs]
  unless jobs
    filters = []
    jobs = Jobs.new
    Thread.current[:sidekiq_deferred_jobs_jobs] = [jobs, filters]
  end
  filters.push(new_filter)
  begin
    yield
  ensure
    filters.pop
    if filters.empty?
      Thread.current[:sidekiq_deferred_jobs_jobs] = nil
      jobs.enqueue!
    end
  end
end

.defer?(klass, opts = nil) ⇒ Boolean

Return true if the specified class with optional options should be deferred. Jobs scheduled to run in the future and jobs that run inline are never deferred.

Parameters:

  • klass (Class) —

    A Sidekiq worker class

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

    Optional options set at runtime for the worker.

Returns:

  • (Boolean) 


66
67
68
69
70
71
 
# File 'lib/sidekiq/deferred_jobs.rb', line 66

def defer?(klass, opts = nil)
  return false if scheduled_opts?(opts) || inline_opts?(opts)
  _jobs, filters = Thread.current[:sidekiq_deferred_jobs_jobs]
  return false if filters.nil?
  filters.any? { |filter| filter.match?(klass, opts) }
end

.defer_worker(klass, args, opts = nil) ⇒ void

This method returns an undefined value.

Schedule a worker to be enqueued at the end of the outermost defer block. If there is no defer block in progress, the worker is enqueued immediately.

Parameters:

  • klass (Class) —

    Sidekiq worker class

  • args (Array) —

    Sidekiq job arguments

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

    Optional sidekiq options specified for the job



80
81
82
83
84
85
86
87
 
# File 'lib/sidekiq/deferred_jobs.rb', line 80

def defer_worker(klass, args, opts = nil)
  jobs, _filters = Thread.current[:sidekiq_deferred_jobs_jobs]
  if jobs
    jobs.defer(klass, args, opts)
  else
    enqueue_job(klass, args, opts)
  end
end

.enqueue_job(klass, args, opts = nil) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Enqueue a job through the normal Sidekiq mechanism.

Parameters:

  • klass (Class) —

    Sidekiq worker class

  • args (Array) —

    Sidekiq job arguments

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

    Optional runtime options for the job

Returns:

  • (Object) —

    the value returned by perform_async



118
119
120
121
122
123
124
 
# File 'lib/sidekiq/deferred_jobs.rb', line 118

def enqueue_job(klass, args, opts = nil)
  if opts
    klass.set(opts).perform_async(*args)
  else
    klass.perform_async(*args)
  end
end

.inline_opts?(opts) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return true if the options indicate a job that runs inline at the call site rather than being pushed to a queue. Inline jobs are never deferred.

Parameters:

  • opts (Hash, Nil) —

    Runtime options set for a job.

Returns:

  • (Boolean) 


106
107
108
109
 
# File 'lib/sidekiq/deferred_jobs.rb', line 106

def inline_opts?(opts)
  return false unless opts
  opts["sync"] == true || opts[:sync] == true
end

.scheduled_opts?(opts) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return true if the options indicate a job scheduled to run in the future. Scheduled jobs are never deferred since they are not enqueued immediately anyway.

Parameters:

  • opts (Hash, Nil) —

    Runtime options set for a job.

Returns:

  • (Boolean) 


95
96
97
98
 
# File 'lib/sidekiq/deferred_jobs.rb', line 95

def scheduled_opts?(opts)
  return false unless opts
  !!(opts["at"] || opts[:at])
end

.undeferred(&block) ⇒ Object

Disable deferred workers within the block. All workers will be enqueued normally within the block. Any jobs already deferred by an enclosing block are left alone and will still be enqueued when that block exits.

Returns:

  • (Object) —

    the return value of the block



50
51
52
53
54
55
56
57
58
 
# File 'lib/sidekiq/deferred_jobs.rb', line 50

def undeferred(&block)
  save_val = Thread.current[:sidekiq_deferred_jobs_jobs]
  begin
    Thread.current[:sidekiq_deferred_jobs_jobs] = nil
    yield
  ensure
    Thread.current[:sidekiq_deferred_jobs_jobs] = save_val
  end
end

.worker_options(klass, opts = nil) ⇒ Hash

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Merge runtime options with the worker class sidekiq_options.

Parameters:

  • klass (Class) —

    Sidekiq worker class

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

    Optional runtime options for the job

Returns:

  • (Hash) —

    the merged options with string keys



132
133
134
135
136
137
138
 
# File 'lib/sidekiq/deferred_jobs.rb', line 132

def worker_options(klass, opts = nil)
  if opts
    klass.sidekiq_options.merge(opts.transform_keys(&:to_s))
  else
    klass.sidekiq_options
  end
end