Class: Resque::DataStore::Workers

Inherits:
Object
  • Object
show all
Defined in:
lib/resque/data_store.rb

Instance Method Summary collapse

Constructor Details

#initialize(redis) ⇒ Workers

Returns a new instance of Workers.



214
215
216
 
# File 'lib/resque/data_store.rb', line 214

def initialize(redis)
  @redis = redis
end

Instance Method Details

#acquire_pruning_dead_worker_lock(worker, expiry) ⇒ Object 



277
278
279
 
# File 'lib/resque/data_store.rb', line 277

def acquire_pruning_dead_worker_lock(worker, expiry)
  @redis.set(redis_key_for_worker_pruning, worker.to_s, :ex => expiry, :nx => true)
end

#all_heartbeats ⇒ Object 



273
274
275
 
# File 'lib/resque/data_store.rb', line 273

def all_heartbeats
  @redis.hgetall(HEARTBEAT_KEY)
end

#get_worker_payload(worker_id) ⇒ Object

return the worker’s payload i.e. job



230
231
232
 
# File 'lib/resque/data_store.rb', line 230

def get_worker_payload(worker_id)
  @redis.get("worker:#{worker_id}")
end

#heartbeat(worker) ⇒ Object 



264
265
266
267
 
# File 'lib/resque/data_store.rb', line 264

def heartbeat(worker)
  heartbeat = @redis.hget(HEARTBEAT_KEY, worker.to_s)
  heartbeat && Time.parse(heartbeat)
end

#heartbeat!(worker, time) ⇒ Object 



269
270
271
 
# File 'lib/resque/data_store.rb', line 269

def heartbeat!(worker, time)
  @redis.hset(HEARTBEAT_KEY, worker.to_s, time.iso8601)
end

#register_worker(worker) ⇒ Object 



238
239
240
241
242
243
 
# File 'lib/resque/data_store.rb', line 238

def register_worker(worker)
  @redis.pipelined do
    @redis.sadd(:workers, worker)
    worker_started(worker)
  end
end

#remove_heartbeat(worker) ⇒ Object 



260
261
262
 
# File 'lib/resque/data_store.rb', line 260

def remove_heartbeat(worker)
  @redis.hdel(HEARTBEAT_KEY, worker.to_s)
end

#set_worker_payload(worker, data) ⇒ Object 



281
282
283
 
# File 'lib/resque/data_store.rb', line 281

def set_worker_payload(worker, data)
  @redis.set(redis_key_for_worker(worker), data)
end

#unregister_worker(worker, &block) ⇒ Object 



249
250
251
252
253
254
255
256
257
258
 
# File 'lib/resque/data_store.rb', line 249

def unregister_worker(worker, &block)
  @redis.pipelined do
    @redis.srem(:workers, worker)
    @redis.del(redis_key_for_worker(worker))
    @redis.del(redis_key_for_worker_start_time(worker))
    @redis.hdel(HEARTBEAT_KEY, worker.to_s)

    block.call
  end
end

#worker_done_working(worker, &block) ⇒ Object 



289
290
291
292
293
294
 
# File 'lib/resque/data_store.rb', line 289

def worker_done_working(worker, &block)
  @redis.pipelined do
    @redis.del(redis_key_for_worker(worker))
    block.call
  end
end

#worker_exists?(worker_id) ⇒ Boolean

Returns:

  • (Boolean) 


234
235
236
 
# File 'lib/resque/data_store.rb', line 234

def worker_exists?(worker_id)
  @redis.sismember(:workers, worker_id)
end

#worker_ids ⇒ Object 



218
219
220
 
# File 'lib/resque/data_store.rb', line 218

def worker_ids
  Array(@redis.smembers(:workers))
end

#worker_start_time(worker) ⇒ Object 



285
286
287
 
# File 'lib/resque/data_store.rb', line 285

def worker_start_time(worker)
  @redis.get(redis_key_for_worker_start_time(worker))
end

#worker_started(worker) ⇒ Object 



245
246
247
 
# File 'lib/resque/data_store.rb', line 245

def worker_started(worker)
  @redis.set(redis_key_for_worker_start_time(worker), Time.now.to_s)
end

#workers_map(worker_ids) ⇒ Object

Given a list of worker ids, returns a map of those ids to the worker’s value in redis, even if that value maps to nil



224
225
226
227
 
# File 'lib/resque/data_store.rb', line 224

def workers_map(worker_ids)
  redis_keys = worker_ids.map { |id| "worker:#{id}" }
  @redis.mapped_mget(*redis_keys)
end