Module: Resque::Plugins::UniqueAtRuntime::ClassMethods

Defined in:
lib/resque/plugins/unique_at_runtime.rb

Instance Method Summary collapse

Instance Method Details

#around_perform_unlock_runtime(*args) ⇒ Object 



147
148
149
150
151
152
153
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 147

def around_perform_unlock_runtime(*args)
  @unlock_queue_executed = false

  yield
ensure
  unlock_queue(*args)
end

#before_perform_lock_runtime(*args) ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 129

def before_perform_lock_runtime(*args)
  if (key = queue_locked?(*args))
    Resque::UniqueAtRuntime.debug("failed to lock queue with #{key}")

    # Sleep so the CPU's rest
    sleep(runtime_requeue_interval)

    # can't get the lock, so re-enqueue the task
    reenqueue(*args)

    # and don't perform
    raise Resque::Job::DontPerform
  else
    Resque::UniqueAtRuntime.debug("check passed will perform")
    true
  end
end

#can_lock_queue?(*args) ⇒ Boolean

returns true if the job signature can be locked (is not currently locked)

Returns:

  • (Boolean) 


61
62
63
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 61

def can_lock_queue?(*args)
  !queue_locked?(*args)
end

#on_failure_unlock_runtime(*args) ⇒ Object

There may be scenarios where the around_perform's ensure unlock± duplicates the on_failure unlock, but that's a small price to pay for uniqueness.



158
159
160
161
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 158

def on_failure_unlock_runtime(*args)
  Resque::UniqueAtRuntime.debug("on failure unlock")
  unlock_queue(*args)
end

#queue_locked?(*args) ⇒ Boolean

returns the locking key if locked, otherwise false

Returns:

  • (Boolean) 


66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 66

def queue_locked?(*args)
  now = Time.now.to_i
  key = unique_at_runtime_redis_key(*args)
  timeout = runtime_lock_timeout_at(now)

  Resque::UniqueAtRuntime.debug("attempting to lock queue with #{key}")

  case atomic_lock_result(key, timeout, now)
  when 1
    false
  when 0
    key
  else
    legacy_queue_locked(key, timeout, now)
  end
end

#reenqueue(*args) ⇒ Object 



125
126
127
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 125

def reenqueue(*args)
  Resque.enqueue(self, *args)
end

#runtime_lock_timeout ⇒ Object 



37
38
39
40
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 37

def runtime_lock_timeout
  instance_variable_get(:@runtime_lock_timeout) ||
    instance_variable_set(:@runtime_lock_timeout, Resque::UniqueAtRuntime.configuration&.lock_timeout)
end

#runtime_lock_timeout_at(now) ⇒ Object 



33
34
35
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 33

def runtime_lock_timeout_at(now)
  now + runtime_lock_timeout + 1
end

#runtime_requeue_interval ⇒ Object 



42
43
44
45
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 42

def runtime_requeue_interval
  instance_variable_get(:@runtime_requeue_interval) ||
    instance_variable_set(:@runtime_requeue_interval, Resque::UniqueAtRuntime.configuration&.requeue_interval)
end

#unique_at_runtime_key_base ⇒ Object 



47
48
49
50
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 47

def unique_at_runtime_key_base
  instance_variable_get(:@unique_at_runtime_key_base) ||
    instance_variable_set(:@unique_at_runtime_key_base, Resque::UniqueAtRuntime.configuration&.unique_at_runtime_key_base)
end

#unique_at_runtime_redis_key(*_) ⇒ Object

Overwrite this method to uniquely identify which mutex should be used for a resque worker.



54
55
56
57
58
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 54

def unique_at_runtime_redis_key(*_)
  queue = Resque.queue_from_class(self)
  Resque::UniqueAtRuntime.debug("getting key for #{queue}!")
  queue
end

#unlock_queue(*args) ⇒ Object 



113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 113

def unlock_queue(*args)
  if @unlock_queue_executed
    Resque::UniqueAtRuntime.debug("unlock queue already executed")
    return
  end

  key = unique_at_runtime_redis_key(*args)
  Resque::UniqueAtRuntime.debug("unlock queue with #{key}")
  Resque.redis.hdel(unique_at_runtime_key_base, key)
  @unlock_queue_executed = true
end