Module: Resque::Integration::Ordered

Defined in:
lib/resque/integration/ordered.rb

Defined Under Namespace

Modules: Overrides Classes: Uniqueness

Constant Summary collapse

ARGS_EXPIRATION = 
1.week

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.extended(base) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/resque/integration/ordered.rb', line 32

def self.extended(base)
  unless base.singleton_class.include?(::Resque::Integration::Unique)
    base.extend ::Resque::Integration::Unique
  end

  unless base.singleton_class.include?(::Resque::Integration::Continuous)
    base.extend ::Resque::Integration::Continuous
  end

  base.singleton_class.class_eval do
    attr_accessor :max_iterations, :uniqueness

    prepend Overrides
  end
end

Instance Method Details

#in_ordered_queue?(*args) ⇒ Boolean

Returns:

  • (Boolean) 


143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/resque/integration/ordered.rb', line 143

def in_ordered_queue?(*args)
  meta = enqueued?(*args)
  return false unless meta

  decoded_args = Resque.decode(Resque.encode(args))

  args_key = ordered_queue_key(meta.meta_id)

  args_meta_id = nil

  ::Resque.redis.lrange(args_key, 0, -1).each do |job_args|
    job_args = ::Resque.decode(job_args)
    meta_id = job_args.shift

    if job_args == decoded_args
      args_meta_id = meta_id
      break
    end
  end

  return false unless args_meta_id

  get_meta(args_meta_id)
end

#ordered_meta_id(args) ⇒ Object 



139
140
141
 
# File 'lib/resque/integration/ordered.rb', line 139

def ordered_meta_id(args)
  Digest::SHA1.hexdigest([Time.now.to_f, rand, self, args].join)
end

#ordered_queue_key(meta_id) ⇒ Object 



135
136
137
 
# File 'lib/resque/integration/ordered.rb', line 135

def ordered_queue_key(meta_id)
  "ordered:#{meta_id}"
end

#ordered_queue_size(meta_id) ⇒ Object 



131
132
133
 
# File 'lib/resque/integration/ordered.rb', line 131

def ordered_queue_size(meta_id)
  Resque.redis.llen(ordered_queue_key(meta_id)).to_i
end