Class: Debci::Drain

Inherits:
Object
  • Object
show all
Defined in:
lib/debci/drain.rb

Defined Under Namespace

Classes: CLI, LifeCycle

Instance Method Summary collapse

Instance Method Details

#discard_job?(job) ⇒ Boolean

Returns:

  • (Boolean) 


92
93
94
95
96
97
98
99
 
# File 'lib/debci/drain.rb', line 92

def discard_job?(job)
  if job.status.nil?
    return false
  end

  Debci.log("DROP ##{job.run_id} (#{job.package.name} #{job.suite}/#{job.arch}/#{job.backend})")
  true
end

#get_job(payload) ⇒ Object 



82
83
84
85
86
87
88
89
90
 
# File 'lib/debci/drain.rb', line 82

def get_job(payload)
  run_id_param = payload.split.find { |part|  part =~ /^run-id:/ }
  return nil unless run_id_param

  run_id = run_id_param.split(":").last.to_i
  return nil unless run_id

  Debci::Job.includes(:package).find(run_id)
end

#queues ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/debci/drain.rb', line 66

def queues
  combinations = Debci.config.arch_list.map do |arch|
    Debci.config.backend_list.map do |backend|
      [arch, backend]
    end
  end.flatten(1)

  existing = combinations.select do |arch, backend|
    Debci::AMQP.queue_exists?(arch, backend)
  end

  existing.map do |arch, backend|
    Debci::AMQP.get_queue(arch, backend)
  end
end

#run ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/debci/drain.rb', line 35

def run
  life_cycle = LifeCycle.new
  seen = {}

  channel = Debci::AMQP.amqp_channel

  queues.each do |queue|
    queue.subscribe(manual_ack: true) do |delivery_info, _properties, payload|
      job = get_job(payload)
      next unless job

      if discard_job?(job)
        # acknowledge the message so it vanishes from the queue
        channel.acknowledge(delivery_info.delivery_tag, false)
        life_cycle.retain
      elsif seen.key?(job.run_id)
        # If we've seen this job before, this means we went over the
        # entire queue so it's time to stop
        life_cycle.expire!
      else
        # In the first time we see a given job, reject the message and
        # send it back to the queue so a worker can pick it up later
        channel.reject(delivery_info.delivery_tag, true)
        seen[job.run_id] = true
      end
    end
  end

  sleep 1 until life_cycle.expired?
end