Class: Wurk::Preflight::Owners

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/preflight/owners.rb

Overview

Live processes and the private lists they own, judged by the same rule the Reaper uses: an identity is alive while its heartbeat info field exists. A pre-nonce key can only match on <host>:<pid>.

Defined Under Namespace

Classes: Process

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(reader) ⇒ Owners

rubocop:disable Metrics/AbcSize



20
21
22
23
24
25
26
27
28
 
# File 'lib/wurk/preflight/owners.rb', line 20

def initialize(reader) # rubocop:disable Metrics/AbcSize
  ids = reader.call('SMEMBERS', Keys::PROCESSES)
  rows = reader.pipeline(ids.map { |id| ['HMGET', id, 'info', 'quiet'] })
  live, gone = ids.zip(rows).partition { |_, (info, _)| info }
  @stale = gone.map(&:first)
  @processes = live.map { |id, (info, quiet)| build(id, info, quiet) }
  @live = @processes.flat_map { |p| [p.identity, p.identity.split(':').first(2).join(':')] }.to_set
  @public_queues = reader.call('SMEMBERS', Keys::QUEUES_SET).to_set { |q| "#{Keys::QUEUE_PREFIX}#{q}" }
end

Instance Attribute Details

#processes ⇒ Object (readonly)

Returns the value of attribute processes.



18
19
20
 
# File 'lib/wurk/preflight/owners.rb', line 18

def processes
  @processes
end

#stale ⇒ Object (readonly)

Returns the value of attribute stale.



18
19
20
 
# File 'lib/wurk/preflight/owners.rb', line 18

def stale
  @stale
end

Instance Method Details

#classify(keys) ⇒ Object

{ 'owner_alive' => [...], 'reclaimable' => [...], 'unparseable' => [...] }, ignoring public queues whose own name holds a |.



35
36
37
38
39
40
41
42
43
 
# File 'lib/wurk/preflight/owners.rb', line 35

def classify(keys)
  out = { 'owner_alive' => [], 'reclaimable' => [], 'unparseable' => [] }
  keys.each do |key|
    next if public_queue?(key)

    out[verdict(key)] << key
  end
  out
end

#foreign ⇒ Object 



45
 
# File 'lib/wurk/preflight/owners.rb', line 45

def foreign = @processes.reject(&:wurk)

#public_queue?(key) ⇒ Boolean

Returns:

  • (Boolean) 


30
 
# File 'lib/wurk/preflight/owners.rb', line 30

def public_queue?(key) = @public_queues.include?(key)

#queue_names ⇒ Object 



31
 
# File 'lib/wurk/preflight/owners.rb', line 31

def queue_names = @public_queues.map { |key| key.delete_prefix(Keys::QUEUE_PREFIX) }