Class: RSMP::SupervisorProxy

Inherits:
Proxy
  • Object
show all
Includes:
Modules::AggregatedStatus, Modules::Alarms, Modules::Commands, Modules::MessageBuffer, Modules::Status
Defined in:
lib/rsmp/proxy/supervisor/supervisor_proxy.rb,
lib/rsmp/proxy/supervisor/modules/alarms.rb,
lib/rsmp/proxy/supervisor/modules/status.rb,
lib/rsmp/proxy/supervisor/modules/commands.rb,
lib/rsmp/proxy/supervisor/modules/message_buffer.rb,
lib/rsmp/proxy/supervisor/modules/status_updates.rb,
lib/rsmp/proxy/supervisor/modules/aggregated_status.rb

Overview

Proxy used by sites to connect to a remote supervisor.

Defined Under Namespace

Modules: Modules

Constant Summary

Constants inherited from Proxy

Proxy::WRAPPING_DELIMITER

Instance Attribute Summary collapse

Attributes inherited from Proxy

#accepted_sxls, #archive, #collector, #connection_info, #core_version, #core_version_string, #ip, #node, #port, #rejected_sxls, #session_id, #site_settings, #state, #sxl_interfaces, #sxls

Attributes included from Task

#task

Attributes included from Distributor

#receivers

Attributes included from Logging

#archive, #logger

Instance Method Summary collapse

Methods included from Modules::MessageBuffer

#buffer_message, #clone_message, #enqueue_buffered_message, #flush_buffered_message?, #flush_message_buffer, #message_buffer_max_messages, #message_buffer_settings, #normalize_aggregated_status_buffer, #prepare_aggregated_status_for_buffer, #prepare_message_for_buffer, #prepare_status_update_for_buffer, #site_originated_buffer_candidate?, #status_buffer_selector?, #status_buffer_selector_matches?, #status_buffer_selectors

Methods included from Modules::AggregatedStatus

#process_aggregated_status_request, #send_aggregated_status, #send_all_aggregated_status

Methods included from Modules::Alarms

#handle_alarm_acknowledge, #handle_alarm_request, #handle_alarm_resume, #handle_alarm_suspend, #process_alarm, #send_active_alarms

Methods included from Modules::Commands

#build_command_rvs, #check_required_command_arguments, #command_catalogue_item, #command_catalogue_match, #execute_commands, #lookup_catalogue_command, #mark_command_unknown, #mark_commands_undefined, #process_command_request, #required_command_argument_names, #simplify_command_requests

Methods included from Modules::Status

#add_status_subscription, #build_undefined_statuses, #fetch_status_value, #fetch_status_values, #get_status_subscribe_interval, #process_status_request, #process_status_subcribe, #process_status_unsubcribe, #prune_unbuffered_status_subscriptions, #remove_status_subscription, #rsmpify_value

Methods included from Modules::StatusUpdates

#add_status_update, #build_status_item, #build_status_list, #check_on_change_update, #check_status_subscription, #collect_component_status_updates, #current_status_value, #each_status_name, #encode_status_value, #fetch_last_sent_status, #interval_update_due?, #precomputed_status_value, #send_component_status_update, #send_status_updates, #status_item_value, #status_update_timer, #status_updates_due, #store_last_sent_status, #store_last_sent_status_item

Methods inherited from Proxy

#author, #build_sxl_interfaces, #clear, #clock, #command_codes, #command_items, #connected?, #disconnected?, #inspect, #log, #multiple_command_codes?, #now, #primary_sxl, #ready?, #receive_alarms?, #reject_multiple_command_codes, #revive, #schemas, #setup, #state_changed, #sxl, #sxl_interface, #sxl_interface_for, #sxl_version, #tlc, version_meets_requirement?

Methods included from Proxy::Modules::Lifecycle

#begin_session, #close_from_result, #close_socket, #close_stream, #connect!, #connection_end_failure, #disconnect, #disconnect!, #distribute_event, #publish_connection_attempt_failure, #stop_reader, #stop_subtasks, #stop_timer, #wait_for_reader

Methods included from Proxy::Modules::Tasks

#connection_read_failure, #log_processing_statistics, #process_received_line, #processing_speed, #read_protocol_line, #run_reader, #run_timer, #session_active?, #session_tasks, #start_reader, #start_timer, #timer, #wait_for_session

Methods included from Proxy::Modules::Versions

#advertised_core_versions, #check_core_version, #configured_sxls, #core_3_3?, #core_versions, #extraneous_version, #normalized_core_versions, #primary_configured_sxl, #send_legacy_version_response, #send_version, #send_version_message, #send_version_request, #send_version_response, #site_id_items, #sxl_request_items, #valid_sxl_response?, #validate_sxl_response!, #version_acknowledged, #version_items, #version_meets_requirement?, version_meets_requirement?, #version_message_attributes, #version_request_attributes, #version_response_sxls, #wire_core_version_aliases

Methods included from Proxy::Modules::Receive

#expect_version_message, #peer_failure, #process_deferred, #process_incoming_message, #process_packet, #publish_peer_failure, #reject_fatal_message, #reject_invalid_message, #reject_invalid_packet, #reject_malformed_packet, #reject_packet, #reject_processed_message, #should_validate_ingoing_message?, #verify_sequence, #will_not_handle

Methods included from Proxy::Modules::Send

#apply_nts_message_attributes, #buffer_message, #buffer_or_fail, #disconnected_send_result, #invalid_outbound_result, #log_send, #prepare_message, #recover_write_failure, #send_generated_message, #send_message, #send_message!, #send_message_and_collect, #send_message_and_collect!, #write_message, #write_protocol

Methods included from Proxy::Modules::Acknowledgements

#acknowledge, #acknowledged_first_outgoing, #check_ack_timeout, #check_ingoing_acknowledged, #check_outgoing_acknowledged, #dont_acknowledge, #dont_expect_acknowledgement, #expect_acknowledgement, #find_original_for_message, #log_acknowledgement_for_original, #log_acknowledgement_for_unknown, #process_ack, #process_not_ack, #status_subscribe_acknowledged

Methods included from Proxy::Modules::Watchdogs

#check_watchdog_timeout, #process_watchdog, #send_watchdog, #start_watchdog, #stop_watchdog, #watchdog_send_timer, #with_watchdog_disabled

Methods included from Proxy::Modules::State

#wait_for_state, #wait_for_state!

Methods included from Task

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

Methods included from Distributor

#add_receiver, #clear_deferred_distribution, #distribute, #distribute_event, #distribute_immediately, #distribute_queued, #initialize_distributor, #inspect, #remove_receiver, #with_deferred_distribution

Methods included from Logging

#author, #initialize_logging, #log

Constructor Details

#initialize(options) ⇒ SupervisorProxy

Returns a new instance of SupervisorProxy.



14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 14

def initialize(options)
  super(options.merge(node: options[:site]))
  @site = options[:site]
  @site_settings = @site.site_settings.clone
  @ip = options[:ip]
  @port = options[:port]
  @status_subscriptions = {}
  @sxls = configured_sxls
  @accepted_sxls = @sxls.dup
  @rejected_sxls = []
  @synthetic_id = Supervisor.build_id_from_ip_port @ip, @port
  @message_buffer = []
end

Instance Attribute Details

#message_buffer ⇒ Object (readonly)

Returns the value of attribute message_buffer.



12
13
14
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 12

def message_buffer
  @message_buffer
end

#site ⇒ Object (readonly)

Returns the value of attribute site.



12
13
14
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 12

def site
  @site
end

#supervisor_id ⇒ Object (readonly)

Returns the value of attribute supervisor_id.



12
13
14
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 12

def supervisor_id
  @supervisor_id
end

Instance Method Details

#acknowledged_first_ingoing(message) ⇒ Object 



175
176
177
178
179
180
181
182
183
184
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 175

def acknowledged_first_ingoing(message)
  case message.type
  when 'Watchdog'
    if core_3_3?
      send_component_list
    else
      handshake_complete
    end
  end
end

#check_sxl_version(message) ⇒ Object 



213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 213

def check_sxl_version(message)
  if core_3_3?
    @rejected_sxls, @accepted_sxls = message.sxls.partition { |item| item['rejected'] }
    validate_sxl_response! @accepted_sxls
    @receive_alarms = message.attributes.fetch('receiveAlarms', true)
  else
    primary = primary_configured_sxl
    raise HandshakeError, 'Legacy Version response received, but no SXL is configured' unless primary

    @accepted_sxls = [primary]
    @rejected_sxls = []
  end
  build_sxl_interfaces
end

#close ⇒ Object 



69
70
71
72
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 69

def close(...)
  prune_unbuffered_status_subscriptions
  super
end

#component_list_acknowledged ⇒ Object 



232
233
234
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 232

def component_list_acknowledged
  handshake_complete
end

#connect ⇒ Object

connect to the supervisor and initiate handshake supervisor



75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 75

def connect
  log "Connecting to supervisor at #{@ip}:#{@port}", level: :info
  begin_session
  self.state = :connecting
  opened = connect_tcp
  return opened if opened.failure?

  self.state = :connected
  @logger.unmute @ip, @port
  log "Connected to supervisor at #{@ip}:#{@port}", level: :info
  Result.success(self)
end

#connect_tcp ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 93

def connect_tcp
  @endpoint = IO::Endpoint.tcp(@ip, @port)

  # this timeout is a workaround for connect hanging on windows if the other side is not present yet
  timeout = @site_settings.dig('timeouts', 'connect') || 1.1
  task.with_timeout timeout do
    @socket = @endpoint.connect
  end
  delay = @site_settings.dig('intervals', 'after_connect')
  task.sleep delay if delay

  @stream = IO::Stream::Buffered.new(@socket)
  @protocol = RSMP::Protocol.new(@stream) # rsmp messages are json terminated with a form-feed
  Result.success(self)
rescue Errno::ECONNREFUSED => e # rescue to avoid log output
  log 'Connection refused', level: :warning
  failed_connection_result(e)
rescue SystemCallError, SocketError, IOError, Async::TimeoutError => e
  failed_connection_result(e)
end

#failed_connection_result(error) ⇒ Object 



114
115
116
117
118
119
120
121
122
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 114

def failed_connection_result(error)
  Result.failure(
    :connection_failed,
    message: "Could not connect to supervisor at #{@ip}:#{@port}: #{error.message}",
    source: :transport,
    context: { ip: @ip, port: @port, session_id: @session_id },
    cause: error
  )
end

#handle_interface_request(message) ⇒ Object 



154
155
156
157
158
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 154

def handle_interface_request(message)
  interface = sxl_interface_for message
  interface.validate_message! message
  interface.process_message message
end

#handshake_complete ⇒ Object 



124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 124

def handshake_complete
  sxl_summary = accepted_sxls.map { |item| "#{item['name']} #{item['version']}" }.join(', ')
  log "Connection to supervisor established, using core #{@core_version}, SXLs [#{sxl_summary}]",
      level: :info
  self.state = :ready
  start_watchdog
  if @site_settings['send_after_connect']
    send_all_aggregated_status
    send_active_alarms if receive_alarms?
  end
  flush_message_buffer
  super
end

#main ⇒ Object 



236
237
238
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 236

def main
  @site.main
end

#process_message(message) ⇒ Object 



138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 138

def process_message(message)
  case message
  when StatusResponse, StatusUpdate, AggregatedStatus, AlarmIssue
    will_not_handle message
  when AggregatedStatusRequest
    process_aggregated_status_request message
  when CommandResponse
    process_command_response message
  when CommandRequest, StatusRequest, StatusSubscribe, StatusUnsubscribe,
       Alarm, AlarmAcknowledged, AlarmSuspend, AlarmResume, AlarmRequest
    handle_interface_request message
  else
    super
  end
end

#process_sxl_request(message) ⇒ Object 



160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 160

def process_sxl_request(message)
  case message
  when CommandRequest
    process_command_request message
  when StatusRequest
    process_status_request message
  when StatusSubscribe
    process_status_subcribe message
  when StatusUnsubscribe
    process_status_unsubcribe message
  when Alarm, AlarmAcknowledged, AlarmSuspend, AlarmResume, AlarmRequest
    process_alarm message
  end
end

#process_version(message) ⇒ Object 



204
205
206
207
208
209
210
211
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 204

def process_version(message)
  return extraneous_version message if @version_determined

  check_core_version message
  check_sxl_version message
  @site_id = Supervisor.build_id_from_ip_port @ip, @port
  version_accepted message
end

#reconnect_delay? ⇒ Boolean

Returns:

  • (Boolean) 


186
187
188
189
190
191
192
193
194
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 186

def reconnect_delay?
  return false if @site_settings['intervals']['reconnect'] == :no

  interval = @site_settings['intervals']['reconnect']
  log "Will try to reconnect again every #{interval} seconds...", level: :info
  @logger.mute @ip, @port
  @task.sleep interval
  true
end

#run ⇒ Object

handle communication if disconnected, then try to reconnect



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 30

def run
  if @protocol
    run_accepted_connection
    return
  end

  loop do
    connected = connect
    unless connected.success?
      publish_connection_attempt_failure(connected.failure)
      break unless reconnect_delay?

      next
    end
    start_reader
    start_handshake
    close_from_result(wait_for_reader)
    break unless reconnect_delay?
  ensure
    close(reason: :internal_failure) if $ERROR_INFO
    close
  end
end

#run_accepted_connection ⇒ Object 



54
55
56
57
58
59
60
61
62
63
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 54

def run_accepted_connection
  begin_session
  self.state = :connected
  start_reader
  start_handshake
  close_from_result(wait_for_reader)
ensure
  close(reason: :internal_failure) if $ERROR_INFO
  close
end

#send_component_list ⇒ Object 



228
229
230
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 228

def send_component_list
  send_generated_message ComponentList.new('components' => @site.component_list)
end

#start_handshake ⇒ Object 



65
66
67
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 65

def start_handshake
  send_version_request @site_settings['site_id'], advertised_core_versions
end

#stop_task ⇒ Object 



88
89
90
91
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 88

def stop_task
  super
  @last_status_sent = nil
end

#version_accepted(message) ⇒ Object 



196
197
198
199
200
201
202
 
# File 'lib/rsmp/proxy/supervisor/supervisor_proxy.rb', line 196

def version_accepted(message)
  log "Received Version message, using RSMP #{@core_version}", message: message, level: :log
  start_timer
  acknowledge message
  @version_determined = true
  send_watchdog
end