Class: Roundhouse::Workers

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/roundhouse/api.rb

Overview

Programmatic access to the current active worker set.

WARNING WARNING WARNING

This is live data that can change every millisecond. If you call #size => 5 and then expect #each to be called 5 times, you’re going to have a bad time.

workers = Roundhouse::Workers.new
workers.size => 2
workers.each do |process_id, thread_id, work|
  # process_id is a unique identifier per Roundhouse process
  # thread_id is a unique identifier per thread
  # work is a Hash which looks like:
  # { 'queue' => name, 'run_at' => timestamp, 'payload' => msg }
  # run_at is an epoch Integer.
end

Instance Method Summary collapse

Instance Method Details

#each ⇒ Object 



821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
 
# File 'lib/roundhouse/api.rb', line 821

def each
  Roundhouse.redis do |conn|
    procs = conn.smembers('processes')
    procs.sort.each do |key|
      valid, workers = conn.pipelined do
        conn.exists(key)
        conn.hgetall("#{key}:workers")
      end
      next unless valid
      workers.each_pair do |tid, json|
        yield key, tid, Roundhouse.load_json(json)
      end
    end
  end
end

#size ⇒ Object

Note that #size is only as accurate as Roundhouse’s heartbeat, which happens every 5 seconds. It is NOT real-time.

Not very efficient if you have lots of Roundhouse processes but the alternative is a global counter which can easily get out of sync with crashy processes.



843
844
845
846
847
848
849
850
851
852
853
854
855
856
 
# File 'lib/roundhouse/api.rb', line 843

def size
  Roundhouse.redis do |conn|
    procs = conn.smembers('processes')
    if procs.empty?
      0
    else
      conn.pipelined do
        procs.each do |key|
          conn.hget(key, 'busy')
        end
      end.map(&:to_i).inject(:+)
    end
  end
end