Module: Wurk::Worker::ClassMethods

Includes:
Job::Options::ClassMethods
Defined in:
lib/wurk/worker.rb

Overview

Class-level DSL mixed into every job class by Wurk::Worker. These are the public enqueue and configuration entry points.

The options DSL (sidekiq_options, get_sidekiq_options, the retry hooks, inheritance) is Job::Options::ClassMethods, shared with the ActiveJob adapter so both validate and inherit identically.

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#sidekiq_options_hash ⇒ Object Originally defined in module Job::Options::ClassMethods

#sidekiq_retries_exhausted_block ⇒ Object (readonly) Originally defined in module Job::Options::ClassMethods

Returns the value of attribute sidekiq_retries_exhausted_block.

#sidekiq_retry_in_block ⇒ Object (readonly) Originally defined in module Job::Options::ClassMethods

Returns the value of attribute sidekiq_retry_in_block.

Instance Method Details

#build_client(pool = get_sidekiq_options['pool'], client_class: nil) ⇒ Object

client_class swaps the enqueue client (e.g. TransactionAwareClient via Wurk.transactional_push!). Resolution order: per-call set(client_class:), then the class option, then the live process default, then Wurk::Client. The default_job_options fallback keeps a global transactional_push! order-independent: a class whose options memoized before the opt-in (its inherited copy is a stale dup) still routes through the new client.



205
206
207
208
209
 
# File 'lib/wurk/worker.rb', line 205

def build_client(pool = get_sidekiq_options['pool'], client_class: nil)
  klass = client_class || get_sidekiq_options['client_class'] ||
          Wurk.default_job_options['client_class'] || Wurk::Client
  klass.new(pool: pool)
end

#clear ⇒ Object 



224
225
226
 
# File 'lib/wurk/worker.rb', line 224

def clear
  ::Wurk::Queues.clear_class(to_s)
end

#client_push(item) ⇒ Object

Raises:

  • (ArgumentError) 


189
190
191
192
193
194
195
196
197
 
# File 'lib/wurk/worker.rb', line 189

def client_push(item)
  raise ArgumentError, "Job arguments to #{name || self} must have string keys" if symbol_keyed?(item)

  # `pool` is a transient enqueue-time attribute: a per-call `set(pool:)`
  # overrides the class-level option, then it's deleted so it never reaches
  # the wire (normalize_item strips any class-level pool re-merged below).
  pool = item.delete('pool') || get_sidekiq_options['pool']
  build_client(pool, client_class: item.delete('client_class')).push(item)
end

#delay ⇒ Object Also known as: delay_for, delay_until

Raises:

  • (ArgumentError) 


265
266
267
 
# File 'lib/wurk/worker.rb', line 265

def delay(*)
  raise ArgumentError, "#{name || self}.delay is removed in Sidekiq 7+. Use #{name || 'klass'}.perform_async."
end

#drain ⇒ Object

Run & remove every fake job for this class — including ones it enqueues mid-drain. Returns the count processed.



230
231
232
233
234
235
236
237
 
# File 'lib/wurk/worker.rb', line 230

def drain
  count = 0
  while (job = ::Wurk::Queues.shift_class(to_s))
    process_job(job)
    count += 1
  end
  count
end

#execute_job(worker, args) ⇒ Object 



261
262
263
 
# File 'lib/wurk/worker.rb', line 261

def execute_job(worker, args)
  worker.perform(*args)
end

#get_sidekiq_options ⇒ Object Originally defined in module Job::Options::ClassMethods

Sidekiq's public API name — must stay get_sidekiq_options.

#inherited(subclass) ⇒ Object Originally defined in module Job::Options::ClassMethods

#jobs ⇒ Object

Fake jobs enqueued for this class, across every queue.



220
221
222
 
# File 'lib/wurk/worker.rb', line 220

def jobs
  JobsView.new(to_s, ::Wurk::Queues.jobs_by_class[to_s] || [])
end

#perform_async ⇒ String?

Enqueue the job to run as soon as a worker is free. Arguments are forwarded to #perform and must be JSON-serializable (string/number/bool/nil/array/hash).

Examples:

EmailJob.perform_async(user.id, "welcome")

Returns:

  • (String, nil) —

    the job id (jid), or nil if a client middleware halted the push



142
143
144
 
# File 'lib/wurk/worker.rb', line 142

def perform_async(*)
  Wurk::Worker::Setter.new(self, {}).perform_async(*)
end

#perform_bulk(items) ⇒ Array<String>

Enqueue many jobs in one round-trip via the Lua bulk path.

Examples:

ImportJob.perform_bulk([[1], [2], [3]])

Parameters:

  • items (Array<Array>) —

    one args array per job

Returns:

  • (Array<String>) —

    the job ids, in order



174
175
176
 
# File 'lib/wurk/worker.rb', line 174

def perform_bulk(items, **)
  Wurk::Worker::Setter.new(self, {}).perform_bulk(items, **)
end

#perform_in(interval) ⇒ String? Also known as: perform_at

Schedule the job for later. perform_at is an alias taking an absolute time; perform_in takes a relative interval.

Examples:

ReminderJob.perform_in(1.hour, lead.id)
ReminderJob.perform_at(Time.now + 3600, lead.id)

Parameters:

  • interval (Numeric, Time) —

    seconds-from-now, or an absolute Time

Returns:

  • (String, nil) —

    the job id (jid)



163
164
165
 
# File 'lib/wurk/worker.rb', line 163

def perform_in(interval, *)
  Wurk::Worker::Setter.new(self, {}).perform_in(interval, *)
end

#perform_inline ⇒ true? Also known as: perform_sync

Run the job synchronously in the current thread, through both middleware chains (see Setter#perform_inline). Useful in tests.

Returns:

  • (true, nil) —

    nil when middleware halted the job, true otherwise



150
151
152
 
# File 'lib/wurk/worker.rb', line 150

def perform_inline(*)
  Wurk::Worker::Setter.new(self, {}).perform_inline(*)
end

#perform_one ⇒ Object

Run & remove the first fake job for this class; EmptyQueueError if none.

Raises:



240
241
242
243
244
245
 
# File 'lib/wurk/worker.rb', line 240

def perform_one
  job = ::Wurk::Queues.shift_class(to_s)
  raise ::Wurk::Testing::EmptyQueueError, "no #{self} jobs were found" if job.nil?

  process_job(job)
end

#process_job(job_hash) ⇒ Object

Execute a normalized job hash through the inline server-middleware chain (empty by default — see Wurk::Testing.server_middleware). Returns the value of the server-middleware invoke (i.e. the worker's perform return), matching Sidekiq::Testing — so perform_one yields the job result.



252
253
254
255
256
257
258
259
 
# File 'lib/wurk/worker.rb', line 252

def process_job(job_hash)
  instance = new
  instance.jid = job_hash['jid']
  instance.bid = job_hash['bid'] if instance.respond_to?(:bid=)
  ::Wurk::Testing.server_middleware.invoke(instance, job_hash, job_hash['queue'] || queue) do
    execute_job(instance, job_hash['args'])
  end
end

#queue ⇒ Object

--- Sidekiq::Testing class-level helpers (spec §24.3) -------------- Only meaningful in :fake / :inline mode; the in-memory store is empty otherwise.



215
216
217
 
# File 'lib/wurk/worker.rb', line 215

def queue
  get_sidekiq_options['queue']
end

#queue_as(queue) ⇒ Object 



130
131
132
 
# File 'lib/wurk/worker.rb', line 130

def queue_as(queue)
  sidekiq_options('queue' => queue.to_s)
end

#set(opts) ⇒ Wurk::Worker::Setter

Return a per-call option carrier so a single enqueue can override class-level options (queue, scheduling, pool, …).

Examples:

ReportJob.set(queue: "low").perform_async(.id)

Parameters:

  • opts (Hash) —

    per-call overrides

Returns:



185
186
187
 
# File 'lib/wurk/worker.rb', line 185

def set(opts)
  Wurk::Worker::Setter.new(self, opts)
end

#sidekiq_options(opts = {}) ⇒ Hash Originally defined in module Job::Options::ClassMethods

Set per-class job options (merged over any inherited options).

Examples:

sidekiq_options queue: "mailers", retry: 3, unique_for: 10.minutes

Opt into Wurk::Status tracking

sidekiq_options track: true

Bound one attempt, and the job as a whole

sidekiq_options timeout: 30, deadline: 5.minutes

Collapse a burst of enqueues into one job

sidekiq_options collapse: { policy: :debounce, wait: 5, max_wait: 60 }

Parameters:

  • opts (Hash) (defaults to: {}) —

    any of queue:, retry:, dead:, backtrace:, expires_in:, tags:, pool:, unique_for:, track:, timeout:, deadline:, collapse:, … (see the migration guide's sidekiq_options table for the full set)

Returns:

  • (Hash) —

    the merged, string-keyed options hash

#sidekiq_retries_exhausted(&block) ⇒ Object Originally defined in module Job::Options::ClassMethods

#sidekiq_retry_in(&block) ⇒ Object Originally defined in module Job::Options::ClassMethods