Class: RSMP::Supervisor

Inherits:
Node
  • Object
show all
Includes:
Modules::Configuration, Modules::Connection, Modules::Sites
Defined in:
lib/rsmp/node/supervisor/supervisor.rb,
lib/rsmp/options/supervisor_options.rb,
lib/rsmp/node/supervisor/modules/sites.rb,
lib/rsmp/node/supervisor/modules/connection.rb,
lib/rsmp/node/supervisor/modules/configuration.rb

Overview

RSMP supervisor (server) that accepts site connections.

Defined Under Namespace

Modules: Modules Classes: Options

Instance Attribute Summary collapse

Attributes inherited from Node

#archive, #clock, #collector, #deferred, #task

Attributes included from EventSource

#event_receivers

Attributes included from Task

#task

Attributes included from Logging

#archive

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Modules::Sites

#aggregated_status_changed, #check_site_already_connected, #check_site_id, #find_site, #find_site_from_ip_port, #site_connected?, #site_ids_changed, #wait_for_site, #wait_for_site!, #wait_for_site_disconnect, #wait_for_site_disconnect!

Methods included from Modules::Connection

#accept?, #accept_connection, #authorize_ip, #build_proxy_settings, #check_max_sites, #close, #format_ip_and_port, #handle_connection, #peek_version_message, #reject_connection, #report_connection_rejection, #retrieve_site_id, #setup_proxy, #validate_and_start_proxy

Methods included from Modules::Configuration

#check_configured_core_version, #check_configured_sxls, #check_site_sxls, #denormalize_site_sxls, #denormalize_supervisor_sxls, #handle_supervisor_settings, #ip_to_site_settings, #site_id_to_site_setting

Methods inherited from Node

#author, #check_required_settings, #clear_deferred, #defer, #do_deferred, #inspect, #now, #process_deferred, #stop_subtasks

Methods included from EventSource

#add_event_receiver, #initialize_event_source, #publish_event, #remove_event_receiver

Methods included from Task

#initialize_task, #restart, #start, #stop_subtasks, #stop_task, #task_status, #wait, #wait_for_condition, #wait_for_condition!, #wait_for_termination

Methods included from Logging

#author, #initialize_logging, #log

Constructor Details

#initialize(options = {}) ⇒ Supervisor

Returns a new instance of Supervisor.



12
13
14
15
16
17
18
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 12

def initialize(options = {})
  handle_supervisor_settings(options[:supervisor_settings] || {})
  super
  @proxies = []
  @ready_condition = Async::Notification.new
  @site_id_condition = Async::Notification.new
end

Instance Attribute Details

#core_version ⇒ Object (readonly)

Returns the value of attribute core_version.



8
9
10
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 8

def core_version
  @core_version
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



8
9
10
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 8

def logger
  @logger
end

#proxies ⇒ Object (readonly)

Returns the value of attribute proxies.



8
9
10
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 8

def proxies
  @proxies
end

#ready_condition ⇒ Object (readonly)

Returns the value of attribute ready_condition.



8
9
10
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 8

def ready_condition
  @ready_condition
end

#site_id_condition ⇒ Object

Returns the value of attribute site_id_condition.



10
11
12
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 10

def site_id_condition
  @site_id_condition
end

#supervisor_settings ⇒ Object (readonly)

Returns the value of attribute supervisor_settings.



8
9
10
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 8

def supervisor_settings
  @supervisor_settings
end

Class Method Details

.build_id_from_ip_port(ip, port) ⇒ Object 



106
107
108
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 106

def self.build_id_from_ip_port(ip, port)
  Digest::MD5.hexdigest("#{ip}:#{port}")[0..8]
end

Instance Method Details

#build_outbound_proxies ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 69

def build_outbound_proxies
  site_entries = (@supervisor_settings['sites'] || {}).except('default')
  site_entries.each_pair do |site_id, site_settings|
    endpoints = site_settings['supervisors'] || []
    merged_settings = site_id_to_site_setting site_id
    endpoints.each do |endpoint|
      @proxies << SiteProxy.new({
                                  supervisor: self,
                                  task: @task,
                                  settings: @supervisor_settings,
                                  site_id: site_id,
                                  site_settings: merged_settings,
                                  ip: endpoint['ip'],
                                  port: endpoint['port'],
                                  logger: @logger,
                                  archive: @archive,
                                  collect: @collect
                                })
    end
  end
end

#build_proxy(settings) ⇒ Object 



102
103
104
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 102

def build_proxy(settings)
  SiteProxy.new settings
end

#client_role? ⇒ Boolean

Returns:

  • (Boolean) 


24
25
26
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 24

def client_role?
  @supervisor_settings['connection_role'] == 'client'
end

#connect_to_sites ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 56

def connect_to_sites
  log 'Starting supervisor in client connection role',
      level: :info,
      timestamp: @clock.now
  build_outbound_proxies
  @ready_condition.signal
  barrier = Async::Barrier.new(parent: @task)
  @proxies.each { |proxy| proxy.start(parent: barrier) }
  barrier.wait
ensure
  barrier.cancel if barrier && !barrier.empty?
end

#run ⇒ Object

listen for connections



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 33

def run
  return connect_to_sites if client_role?

  log "Starting supervisor on port #{@supervisor_settings['port']}",
      level: :info,
      timestamp: @clock.now

  @endpoint = IO::Endpoint.tcp('0.0.0.0', @supervisor_settings['port'])
  barrier = Async::Barrier.new(parent: @task)
  @accept_task = barrier.async do |task|
    task.annotate 'supervisor accept loop'
    @endpoint.accept do |socket|
      handle_connection(socket)
    end
    task.wait_all
  end

  @ready_condition.signal
  barrier.wait
ensure
  barrier.cancel if barrier && !barrier.empty?
end

#server_role? ⇒ Boolean

Returns:

  • (Boolean) 


28
29
30
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 28

def server_role?
  !client_role?
end

#site_id ⇒ Object 



20
21
22
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 20

def site_id
  @supervisor_settings['site_id']
end

#stop ⇒ Object

stop



92
93
94
95
96
97
98
99
100
 
# File 'lib/rsmp/node/supervisor/supervisor.rb', line 92

def stop
  log "Stopping supervisor #{@supervisor_settings['site_id']}", level: :info

  @accept_task&.cancel if @accept_task&.running?
  @accept_task = nil

  @endpoint = nil
  super
end