Class: SidekiqUniqueJobs::UniqueArgs

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Normalizer
Defined in:
lib/sidekiq_unique_jobs/unique_args.rb

Overview

This class exists to be testable and the entire api should be considered private rubocop:disable ClassLength

Constant Summary collapse

CLASS_NAME = 
'SidekiqUniqueJobs::UniqueArgs'

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Normalizer

jsonify

Constructor Details

#initialize(item) ⇒ UniqueArgs

Returns a new instance of UniqueArgs.



21
22
23
24
25
26
27
28
29
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 21

def initialize(item)
  Sidekiq::Logging.with_context(CLASS_NAME) do
    @item = item
    @worker_class            ||= worker_class_constantize(@item[CLASS_KEY])
    @item[UNIQUE_PREFIX_KEY] ||= unique_prefix
    @item[UNIQUE_ARGS_KEY]     = unique_args(@item[ARGS_KEY])
    @item[UNIQUE_DIGEST_KEY]   = unique_digest
  end
end

Class Method Details

.digest(item) ⇒ Object 



17
18
19
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 17

def self.digest(item)
  new(item).unique_digest
end

Instance Method Details

#digestable_hash ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 45

def digestable_hash
  @item.slice(CLASS_KEY, QUEUE_KEY, UNIQUE_ARGS_KEY).tap do |hash|
    if unique_on_all_queues?
      logger.debug { "#{__method__} deleting queue: #{@item[QUEUE_KEY]}" }
      hash.delete(QUEUE_KEY)
    end
    if unique_across_workers?
      logger.debug { "#{__method__} deleting class: #{@item[CLASS_KEY]}" }
      hash.delete(CLASS_KEY)
    end
  end
end

#filter_by_proc(args) ⇒ Object 



112
113
114
115
116
117
118
119
120
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 112

def filter_by_proc(args)
  if unique_args_method.nil?
    logger.warn { "#{__method__} : unique_args_method is nil. Returning (#{args})" }
    return args
  end
  filter_args = unique_args_method.call(args)
  logger.debug { "#{__method__} : #{args} -> #{filter_args}" }
  filter_args
end

#filter_by_symbol(args) ⇒ Object 



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 122

def filter_by_symbol(args)
  unless @worker_class.respond_to?(unique_args_method)
    logger.warn do
      "#{__method__} : #{@worker_class} does not respond to #{unique_args_method}). Returning (#{args})"
    end
    return args
  end

  filter_args = @worker_class.send(unique_args_method, args)
  logger.debug { "#{__method__} : #{unique_args_method}(#{args}) => #{filter_args}" }
  filter_args
rescue ArgumentError => ex
  logger.fatal "#{__method__} : #{@worker_class}'s #{unique_args_method} needs at least one argument"
  logger.fatal ex
  args
end

#filtered_args(args) ⇒ Object

Filters unique arguments by proc or symbol returns provided arguments for other configurations



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 96

def filtered_args(args)
  return args if args.empty?
  json_args = Normalizer.jsonify(args)
  logger.debug { "#filtered_args #{args} => #{json_args}" }

  case unique_args_method
  when Proc
    filter_by_proc(json_args)
  when Symbol
    filter_by_symbol(json_args)
  else
    logger.debug { "#{__method__} arguments not filtered (using all arguments for uniqueness)" }
    json_args
  end
end

#sidekiq_worker_class? ⇒ Boolean

Returns:

  • (Boolean) 


85
86
87
88
89
90
91
92
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 85

def sidekiq_worker_class?
  if @worker_class.respond_to?(:get_sidekiq_options)
    true
  else
    logger.debug { "#{__method__} #{@worker_class} does not respond to :get_sidekiq_options" }
    nil
  end
end

#unique_across_workers? ⇒ Boolean

Returns:

  • (Boolean) 


75
76
77
78
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 75

def unique_across_workers?
  return unless sidekiq_worker_class?
  @item[UNIQUE_ACROSS_WORKERS_KEY] || @worker_class.get_sidekiq_options[UNIQUE_ACROSS_WORKERS_KEY]
end

#unique_args(args) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 58

def unique_args(args)
  if unique_args_enabled?
    filtered_args(args)
  else
    logger.debug { "#{__method__} : unique arguments disabled" }
    args
  end
rescue NameError
  # fallback to not filtering args when class can't be instantiated
  return args
end

#unique_args_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


80
81
82
83
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 80

def unique_args_enabled?
  return unless sidekiq_worker_class?
  unique_args_method # && !unique_args_method.is_a?(Boolean)
end

#unique_args_method ⇒ Object 



139
140
141
142
143
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 139

def unique_args_method
  @unique_args_method ||= @worker_class.get_sidekiq_options[UNIQUE_ARGS_KEY] if sidekiq_worker_class?
  @unique_args_method ||= :unique_args if @worker_class.respond_to?(:unique_args)
  @unique_args_method ||= Sidekiq.default_worker_options.stringify_keys[UNIQUE_ARGS_KEY]
end

#unique_digest ⇒ Object 



31
32
33
34
35
36
37
38
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 31

def unique_digest
  @unique_digest ||= begin
    digest = Digest::MD5.hexdigest(Sidekiq.dump_json(digestable_hash))
    digest = "#{unique_prefix}:#{digest}"
    logger.debug { "#{__method__} : #{digestable_hash} into #{digest}" }
    digest
  end
end

#unique_on_all_queues? ⇒ Boolean

Returns:

  • (Boolean) 


70
71
72
73
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 70

def unique_on_all_queues?
  return unless sidekiq_worker_class?
  @item[UNIQUE_ON_ALL_QUEUES_KEY] || @worker_class.get_sidekiq_options[UNIQUE_ON_ALL_QUEUES_KEY]
end

#unique_prefix ⇒ Object 



40
41
42
43
 
# File 'lib/sidekiq_unique_jobs/unique_args.rb', line 40

def unique_prefix
  return config.unique_prefix unless sidekiq_worker_class?
  @worker_class.get_sidekiq_options[UNIQUE_PREFIX_KEY] || config.unique_prefix
end