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

#empty?, #progress, #size, #to_a, #wait_for_master, #workers_count

Constructor Details

#initialize(tests, redis:, build_id:, worker_id:, timeout:, max_requeues: 0, requeue_tolerance: 0.0) ⇒ Worker

Returns a new instance of Worker.



16
17
18
19
20
21
22
23
24
25
 
# File 'lib/ci/queue/redis/worker.rb', line 16

def initialize(tests, redis:, build_id:, worker_id:, timeout:, max_requeues: 0, requeue_tolerance: 0.0)
  @reserved_test = nil
  @max_requeues = max_requeues
  @global_max_requeues = (tests.size * requeue_tolerance).ceil
  @shutdown_required = false
  super(redis: redis, build_id: build_id)
  @worker_id = worker_id
  @timeout = timeout
  push(tests)
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, success) ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/ci/queue/redis/worker.rb', line 72

def acknowledge(test, success)
  if @reserved_test == test
    @reserved_test = nil
  else
    raise ReservationError, "Acknowledged #{test.inspect} but #{@reserved_test.inspect} was reserved"
  end

  if !success && should_requeue?(test)
    requeue(test)
    false
  else
    ack(test)
    true
  end
end

#master? ⇒ Boolean

Returns:

  • (Boolean) 


35
36
37
 
# File 'lib/ci/queue/redis/worker.rb', line 35

def master?
  @master
end

#minitest_reporters ⇒ Object 



61
62
63
64
65
66
67
68
69
70
 
# File 'lib/ci/queue/redis/worker.rb', line 61

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

#poll ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/ci/queue/redis/worker.rb', line 39

def poll
  wait_for_master
  until shutdown_required? || empty?
    if test = reserve
      yield test
    else
      sleep 0.05
    end
  end
rescue ::Redis::BaseConnectionError
end

#retry_queue(**args) ⇒ Object 



51
52
53
54
55
56
57
58
59
 
# File 'lib/ci/queue/redis/worker.rb', line 51

def retry_queue(**args)
  Retry.new(
    redis.lrange(key('worker', worker_id, 'queue'), 0, -1).reverse.uniq,
    redis: redis,
    build_id: build_id,
    worker_id: worker_id,
    **args
  )
end

#shutdown! ⇒ Object 



27
28
29
 
# File 'lib/ci/queue/redis/worker.rb', line 27

def shutdown!
  @shutdown_required = true
end

#shutdown_required? ⇒ Boolean

Returns:

  • (Boolean) 


31
32
33
 
# File 'lib/ci/queue/redis/worker.rb', line 31

def shutdown_required?
  @shutdown_required
end