Module: Resque::Plugins::LonelyJob

Defined in:
lib/resque-lonely_job.rb,
lib/resque-lonely_job/version.rb

Constant Summary collapse

LOCK_TIMEOUT =

5 days

60 * 60 * 24 * 5
VERSION = 
"1.1.3"

Instance Method Summary collapse

Instance Method Details

#around_perform(*args) ⇒ Object 



55
56
57
58
59
60
61
 
# File 'lib/resque-lonely_job.rb', line 55

def around_perform(*args)
  begin
    yield
  ensure
    unlock_queue(*args)
  end
end

#before_perform(*args) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/resque-lonely_job.rb', line 42

def before_perform(*args)
  unless can_lock_queue?(*args)
    # Sleep so the CPU's rest
    sleep(requeue_interval)

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

    # and don't perform
    raise Resque::Job::DontPerform
  end
end

#can_lock_queue?(*args) ⇒ Boolean

Returns:

  • (Boolean) 


22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/resque-lonely_job.rb', line 22

def can_lock_queue?(*args)
  now = Time.now.to_i
  key = redis_key(*args)
  timeout = lock_timeout

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

#lock_timeout ⇒ Object 



8
9
10
 
# File 'lib/resque-lonely_job.rb', line 8

def lock_timeout
  Time.now.to_i + LOCK_TIMEOUT + 1
end

#redis_key(*args) ⇒ Object

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



18
19
20
 
# File 'lib/resque-lonely_job.rb', line 18

def redis_key(*args)
  "lonely_job:#{@queue}"
end

#reenqueue(*args) ⇒ Object 



38
39
40
 
# File 'lib/resque-lonely_job.rb', line 38

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

#requeue_interval ⇒ Object 



12
13
14
 
# File 'lib/resque-lonely_job.rb', line 12

def requeue_interval
  self.instance_variable_get(:@requeue_interval) || 1
end

#unlock_queue(*args) ⇒ Object 



34
35
36
 
# File 'lib/resque-lonely_job.rb', line 34

def unlock_queue(*args)
  Resque.redis.del(redis_key(*args))
end