Class: CI::Queue::Redis::Worker

Inherits:
Base
  • Object
show all
Defined in:
lib/ci/queue/redis/worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#exhausted?, #progress, #queue_initialized?, #size, #to_a, #wait_for_master, #workers_count

Constructor Details

#initialize(redis, config) ⇒ Worker

Returns a new instance of Worker.



16
17
18
19
20
 
# File 'lib/ci/queue/redis/worker.rb', line 16

def initialize(redis, config)
  @reserved_test = nil
  @shutdown_required = false
  super(redis, config)
end

Instance Attribute Details

#total ⇒ Object (readonly)

Returns the value of attribute total.



14
15
16
 
# File 'lib/ci/queue/redis/worker.rb', line 14

def total
  @total
end

Instance Method Details

#acknowledge(test) ⇒ Object 



79
80
81
82
83
84
85
86
87
 
# File 'lib/ci/queue/redis/worker.rb', line 79

def acknowledge(test)
  test_key = test.id
  raise_on_mismatching_test(test_key)
  eval_script(
    :acknowledge,
    keys: [key('running'), key('processed')],
    argv: [test_key],
  ) == 1
end

#master? ⇒ Boolean

Returns:

  • (Boolean) 


41
42
43
 
# File 'lib/ci/queue/redis/worker.rb', line 41

def master?
  @master
end

#minitest_reporters ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/ci/queue/redis/worker.rb', line 66

def minitest_reporters
  require 'minitest/reporters/queue_reporter'
  require 'minitest/reporters/redis_reporter'
  @minitest_reporters ||= [
    Minitest::Reporters::QueueReporter.new,
    Minitest::Reporters::RedisReporter::Worker.new(
      redis: redis,
      build_id: build_id,
      worker_id: worker_id,
    )
  ]
end

#poll ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/ci/queue/redis/worker.rb', line 45

def poll
  wait_for_master
  until shutdown_required? || exhausted?
    if test = reserve
      yield index.fetch(test)
    else
      sleep 0.05
    end
  end
rescue ::Redis::BaseConnectionError
end

#populate(tests, random: Random.new) ⇒ Object 



22
23
24
25
26
27
 
# File 'lib/ci/queue/redis/worker.rb', line 22

def populate(tests, random: Random.new)
  @index = tests.map { |t| [t.id, t] }.to_h
  tests = Queue.shuffle(tests, random)
  push(tests.map(&:id))
  self
end

#populated? ⇒ Boolean

Returns:

  • (Boolean) 


29
30
31
 
# File 'lib/ci/queue/redis/worker.rb', line 29

def populated?
  !!defined?(@index)
end

#requeue(test, offset: Redis.requeue_offset) ⇒ Object 



89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/ci/queue/redis/worker.rb', line 89

def requeue(test, offset: Redis.requeue_offset)
  test_key = test.id
  raise_on_mismatching_test(test_key)

  requeued = eval_script(
    :requeue,
    keys: [key('processed'), key('requeues-count'), key('queue'), key('running')],
    argv: [config.max_requeues, config.global_max_requeues(total), test_key, offset],
  ) == 1

  @reserved_test = test_key unless requeued
  requeued
end

#retry_queue ⇒ Object 



57
58
59
60
 
# File 'lib/ci/queue/redis/worker.rb', line 57

def retry_queue
  log = redis.lrange(key('worker', worker_id, 'queue'), 0, -1).reverse.uniq
  Retry.new(log, config, redis: redis)
end

#shutdown! ⇒ Object 



33
34
35
 
# File 'lib/ci/queue/redis/worker.rb', line 33

def shutdown!
  @shutdown_required = true
end

#shutdown_required? ⇒ Boolean

Returns:

  • (Boolean) 


37
38
39
 
# File 'lib/ci/queue/redis/worker.rb', line 37

def shutdown_required?
  @shutdown_required
end

#supervisor ⇒ Object 



62
63
64
 
# File 'lib/ci/queue/redis/worker.rb', line 62

def supervisor
  Supervisor.new(redis_url, config)
end