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 



91
92
93
94
95
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 91

def around_perform_unlock_runtime(*args)
  yield
ensure
  unlock_queue(*args)
end

#before_perform_lock_runtime(*args) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 73

def before_perform_lock_runtime(*args)
  if (key = queue_locked?(*args))
    Resque::UniqueAtRuntime.runtime_unique_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.runtime_unique_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)



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

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.



100
101
102
103
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 100

def on_failure_unlock_runtime(*args)
  Resque::UniqueAtRuntime.runtime_unique_debug('on failure unlock')
  unlock_queue(*args)
end

#queue_locked?(*args) ⇒ Boolean

returns the locking key if locked, otherwise false



48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 48

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

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

  # Per http://redis.io/commands/setnx
  return false if Resque.redis.setnx(key, timeout)
  return key if Resque.redis.get(key).to_i > now
  return false if Resque.redis.getset(key, timeout).to_i <= now

  key
end

#reenqueue(*args) ⇒ Object 



69
70
71
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 69

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

#runtime_lock_timeout ⇒ Object 



27
28
29
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 27

def runtime_lock_timeout
  instance_variable_get(:@runtime_lock_timeout) || Resque::UniqueAtRuntime.uniq_config&.lock_timeout
end

#runtime_lock_timeout_at(now) ⇒ Object 



23
24
25
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 23

def runtime_lock_timeout_at(now)
  now + runtime_lock_timeout + 1
end

#runtime_requeue_interval ⇒ Object 



31
32
33
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 31

def runtime_requeue_interval
  instance_variable_get(:@runtime_requeue_interval) || Resque::UniqueAtRuntime.uniq_config&.requeue_interval
end

#unique_at_runtime_key_base ⇒ Object 



105
106
107
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 105

def unique_at_runtime_key_base
  instance_variable_get(:@unique_at_runtime_key_base) || Resque::UniqueAtRuntime.uniq_config&.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.



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

def unique_at_runtime_redis_key(*_)
  Resque::UniqueAtRuntime.runtime_unique_debug("getting key for #{@queue}!")
  "#{unique_at_runtime_key_base}:#{@queue}"
end

#unlock_queue(*args) ⇒ Object 



63
64
65
66
67
 
# File 'lib/resque/plugins/unique_at_runtime.rb', line 63

def unlock_queue(*args)
  key = unique_at_runtime_redis_key(*args)
  Resque::UniqueAtRuntime.runtime_unique_debug("unlock queue with #{key}")
  Resque.redis.del(key)
end