Class: Sidekiq::Status::ClientMiddleware

Inherits:
Object
  • Object
show all
Includes:
Storage
Defined in:
lib/sidekiq-status/client_middleware.rb

Overview

Should be in the client middleware chain

Constant Summary

Constants included from Storage

Storage::BATCH_LIMIT, Storage::RESERVED_FIELDS

Instance Method Summary collapse

Constructor Details

#initialize(opts = {}) ⇒ ClientMiddleware

Parameterized initialization, use it when adding middleware to client chain chain.add Sidekiq::Status::ClientMiddleware, :expiration => 60 * 5

Parameters:

  • (defaults to: {})

    middleware initialization options

Options Hash (opts):

  • :expiration (Fixnum) —

    ttl for complete jobs



13
14
15
 
# File 'lib/sidekiq-status/client_middleware.rb', line 13

def initialize(opts = {})
  @expiration = opts[:expiration]
end

Instance Method Details

#call(worker_class, msg, queue, redis_pool = nil) ⇒ Object

Uses msg id and puts :queued status in the job's Redis hash

Parameters:

  • if includes Sidekiq::Status::Worker, the job gets processed with the plugin

  • job arguments

  • the queue's name

  • (defaults to: nil)

    optional redis connection pool



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/sidekiq-status/client_middleware.rb', line 22

def call(worker_class, msg, queue, redis_pool=nil)

  # Determine the actual job class
  klass = (!msg["args"][0].is_a?(String) && msg["args"][0]["job_class"]) || worker_class rescue worker_class
  job_class = if klass.is_a?(Class)
                klass
              elsif Module.const_defined?(klass)
                Module.const_get(klass)
              else
                nil
              end

  # Store data if the job is a Sidekiq::Status::Worker
  if job_class && job_class.ancestors.include?(Sidekiq::Status::Worker)
    initial_metadata = {
      jid: msg['jid'],
      status: :queued,
      worker: JOB_CLASS.new(msg, queue).display_class,
      args: display_args(msg, queue),
      enqueued_at: Time.now.to_i
    }
    store_for_id msg['jid'], initial_metadata, job_class.new.expiration || @expiration, redis_pool
  end

  yield

end

#display_args(msg, queue) ⇒ Object 



50
51
52
53
54
55
56
 
# File 'lib/sidekiq-status/client_middleware.rb', line 50

def display_args(msg, queue)
  job = JOB_CLASS.new(msg, queue)
  return job.display_args.to_a.empty? ? "{}" : job.display_args.to_json
rescue Exception => e
  # For Sidekiq ~> 2.7
  return msg['args'].to_a.empty? ? nil : msg['args'].to_json
end