Class: RFlow::Master
- Inherits:
-
DaemonProcess
- Object
- DaemonProcess
- RFlow::Master
- Defined in:
- lib/rflow/master.rb
Instance Attribute Summary collapse
-
#brokers ⇒ Object
readonly
Returns the value of attribute brokers.
-
#shards ⇒ Object
readonly
Returns the value of attribute shards.
Instance Method Summary collapse
-
#initialize(config) ⇒ Master
constructor
A new instance of Master.
- #run_process ⇒ Object
- #spawn_subprocesses ⇒ Object
- #subprocesses ⇒ Object
Methods inherited from DaemonProcess
#daemonize!, #run!, #shutdown!
Constructor Details
#initialize(config) ⇒ Master
Returns a new instance of Master.
10 11 12 13 14 15 |
# File 'lib/rflow/master.rb', line 10 def initialize(config) super(config['rflow.application_name'], 'Master', pid_file_path: config['rflow.pid_file_path']) @shards = config.shards.map {|config| Shard.new(config) } RFlow.logger.context_width = @shards.flat_map(&:workers).map(&:name).map(&:length).max @brokers = config.connections.flat_map(&:brokers).map {|config| Broker.build(config) } end |
Instance Attribute Details
#brokers ⇒ Object (readonly)
Returns the value of attribute brokers.
8 9 10 |
# File 'lib/rflow/master.rb', line 8 def brokers @brokers end |
#shards ⇒ Object (readonly)
Returns the value of attribute shards.
7 8 9 |
# File 'lib/rflow/master.rb', line 7 def shards @shards end |
Instance Method Details
#run_process ⇒ Object
29 30 31 32 33 |
# File 'lib/rflow/master.rb', line 29 def run_process EM.run do # TODO: Monitor the workers end end |
#spawn_subprocesses ⇒ Object
17 18 19 20 21 22 23 |
# File 'lib/rflow/master.rb', line 17 def spawn_subprocesses RFlow.logger.debug "Running #{brokers.count} brokers" if brokers.count > 0 brokers.each(&:spawn!) RFlow.logger.debug "#{brokers.count} brokers started: #{brokers.map { |w| "#{w.name} (#{w.pid})" }.join(', ')}" if brokers.count > 0 shards.each(&:run!) end |
#subprocesses ⇒ Object
25 26 27 |
# File 'lib/rflow/master.rb', line 25 def subprocesses brokers + shards.flat_map(&:workers) end |