Module: Wurk::QueueConsumers

Defined in:
lib/wurk/queue_consumers.rb

Overview

Who listens to which queue, read from what every process already publishes: the queues list in its heartbeat info. That list is the union across a process's capsules, so a queue served by one capsule counts. Expired identities have no info and are not counted. No new Redis keys.

Defined Under Namespace

Classes: Watch

Class Method Summary collapse

Class Method Details

.counts(conn) ⇒ Object

{ queue => live subscribers }. A quiet process still counts: it is a deliberate pause, not a queue nobody thought to list.



18
19
20
21
22
23
 
# File 'lib/wurk/queue_consumers.rb', line 18

def counts(conn)
  ids = conn.call('SMEMBERS', Keys::PROCESSES)
  return {} if ids.empty?

  tally(conn.pipelined { |pipe| ids.each { |id| pipe.call('HGET', id, 'info') } })
end

.subscribed(info) ⇒ Object 



31
32
33
34
35
36
37
 
# File 'lib/wurk/queue_consumers.rb', line 31

def subscribed(info)
  return [] if info.nil?

  Array(::JSON.parse(info)['queues']).map(&:to_s).uniq
rescue ::JSON::ParserError, ::TypeError
  []
end

.tally(infos) ⇒ Object 



25
26
27
28
29
 
# File 'lib/wurk/queue_consumers.rb', line 25

def tally(infos)
  infos.each_with_object(Hash.new(0)) do |info, out|
    subscribed(info).each { |queue| out[queue] += 1 }
  end
end