Class: Wakame::Master

Inherits:
Object
  • Object
show all
Includes:
AMQPClient, QueueDeclare
Defined in:
lib/wakame/master.rb

Instance Attribute Summary collapse

Attributes included from AMQPClient

#amqp_client, #mq, #queue_subscribers

Instance Method Summary collapse

Methods included from QueueDeclare

included

Methods included from AMQPClient

#add_subscriber, #amq, #amqp_server_uri, #close, #connect, #connected?, #define_queue, included, #publish_to

Constructor Details

#initialize(opts = {}) ⇒ Master

Returns a new instance of Master.



361
362
363
 
# File 'lib/wakame/master.rb', line 361

def initialize(opts={})
  pre_setup
end

Instance Attribute Details

#action_manager ⇒ Object (readonly)

Returns the value of attribute action_manager.



358
359
360
 
# File 'lib/wakame/master.rb', line 358

def action_manager
  @action_manager
end

#agent_monitor ⇒ Object (readonly)

Returns the value of attribute agent_monitor.



358
359
360
 
# File 'lib/wakame/master.rb', line 358

def agent_monitor
  @agent_monitor
end

#cluster_manager ⇒ Object (readonly)

Returns the value of attribute cluster_manager.



358
359
360
 
# File 'lib/wakame/master.rb', line 358

def cluster_manager
  @cluster_manager
end

#command_queue ⇒ Object (readonly)

Returns the value of attribute command_queue.



358
359
360
 
# File 'lib/wakame/master.rb', line 358

def command_queue
  @command_queue
end

#managers ⇒ Object (readonly)

Returns the value of attribute managers.



359
360
361
 
# File 'lib/wakame/master.rb', line 359

def managers
  @managers
end

#started_at ⇒ Object (readonly)

Returns the value of attribute started_at.



358
359
360
 
# File 'lib/wakame/master.rb', line 358

def started_at
  @started_at
end

Instance Method Details

#actor_request(agent_id, path, *args) ⇒ Object 



366
367
368
369
 
# File 'lib/wakame/master.rb', line 366

def actor_request(agent_id, path, *args)
  request = Wakame::Packets::ActorRequest.new(agent_id, Util.gen_id, path, *args)
  ActorRequest.new(self, request)
end

#cleanup ⇒ Object 



372
373
374
375
 
# File 'lib/wakame/master.rb', line 372

def cleanup
  @managers.each { |m| m.terminate }
  @command_queue.shutdown
end

#init ⇒ Object

post_setup



385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
 
# File 'lib/wakame/master.rb', line 385

def init
  raise 'has to be put in EM.run context' unless EM.reactor_running?
  @command_queue = register_manager(CommandQueue.new)

  # WorkerThread has to run earlier than other managers.
  @agent_monitor = register_manager(AgentMonitor.new)
  @cluster_manager = register_manager(ClusterManager.new)
  @action_manager = register_manager(ActionManager.new)

  @managers.each {|m|
    Wakame.log.debug("Initializing Manager Module: #{m.class}")
    m.init
  }

  Wakame.log.info("Started master process : WAKAME_ROOT=#{Wakame.config.root_path} WAKAME_ENV=#{Wakame.config.environment}")
end

#register_manager(manager) ⇒ Object

Raises:

  • (ArgumentError) 


377
378
379
380
381
382
 
# File 'lib/wakame/master.rb', line 377

def register_manager(manager)
  raise ArgumentError unless manager.kind_of? Manager
  manager.master = self
  @managers << manager
  manager
end