Class: RSMP::SiteProxy

Inherits:
Proxy
  • Object
show all
Includes:
Components
Defined in:
lib/rsmp/site_proxy.rb

Constant Summary

Constants inherited from Proxy

Proxy::WRAPPING_DELIMITER

Instance Attribute Summary collapse

Attributes included from Components

#components

Attributes inherited from Proxy

#archive, #collector, #connection_info, #ip, #port, #state, #sxl, #task

Attributes included from Notifier

#listeners

Attributes included from Logging

#archive, #logger

Instance Method Summary collapse

Methods included from Components

#add_component, #build_component, #check_main_component, #find_component, #infer_component_type, #initialize_components, #setup_components

Methods inherited from Proxy

#acknowledge, #acknowledged_first_ingoing, #acknowledged_first_outgoing, #author, #buffer_message, #check_ack_timeout, #check_ingoing_acknowledged, #check_outgoing_acknowledged, #check_rsmp_version, #check_watchdog_timeout, #clear, #clock, #close_socket, #collect, #connected?, #dont_acknowledge, #dont_expect_acknowledgement, #expect_acknowledgement, #expect_version_message, #extraneous_version, #find_original_for_message, #get_schemas, #log, #log_acknowledgement_for_original, #log_acknowledgement_for_unknown, #log_send, #process_ack, #process_deferred, #process_not_ack, #process_packet, #ready?, #rsmp_versions, #run, #send_message, #send_version, #send_watchdog, #set_state, #setup, #should_validate_ingoing_message?, #start_reader, #start_timer, #start_watchdog, #stop_tasks, #timer, #verify_sequence, #wait_for_acknowledgement, #wait_for_state, #watchdog_send_timer, #will_not_handle

Methods included from Inspect

#inspector

Methods included from Notifier

#add_listener, #clear_deferred_notify, #deferred_notify, #dequeue_notify, #distribute_error, #initialize_distributor, #notify, #notify_without_defer, #remove_listener

Methods included from Wait

#wait_for

Methods included from Logging

#author, #initialize_logging, #log

Constructor Details

#initialize(options) ⇒ SiteProxy

Returns a new instance of SiteProxy.



9
10
11
12
13
14
15
 
# File 'lib/rsmp/site_proxy.rb', line 9

def initialize options
  super options
  initialize_components
  @supervisor = options[:supervisor]
  @settings = @supervisor.supervisor_settings.clone
  @site_id = options[:site_id]
end

Instance Attribute Details

#site_id ⇒ Object (readonly)

Returns the value of attribute site_id.



7
8
9
 
# File 'lib/rsmp/site_proxy.rb', line 7

def site_id
  @site_id
end

#supervisor ⇒ Object (readonly)

Returns the value of attribute supervisor.



7
8
9
 
# File 'lib/rsmp/site_proxy.rb', line 7

def supervisor
  @supervisor
end

Instance Method Details

#aggregated_status_changed(component, options = {}) ⇒ Object 



133
134
135
 
# File 'lib/rsmp/site_proxy.rb', line 133

def aggregated_status_changed component, options={}
  @supervisor.aggregated_status_changed self, component
end

#check_site_ids(message) ⇒ Object 



303
304
305
306
307
308
309
310
 
# File 'lib/rsmp/site_proxy.rb', line 303

def check_site_ids message
  # RSMP support multiple site ids. we don't support this yet. instead we use the first id only
  site_id = message.attribute("siteId").map { |item| item["sId"] }.first
  @supervisor.check_site_id site_id
  @site_id = site_id
  setup_site_settings
  site_ids_changed
end

#check_sxl_version(message) ⇒ Object 



273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
 
# File 'lib/rsmp/site_proxy.rb', line 273

def check_sxl_version message

  # check that we have a schema for specified sxl type and version
  # note that the type comes from the site config, while the version
  # comes from the Version message send by the site
  type = 'tlc'
  version = message.attribute 'SXL'
  RSMP::Schemer::find_schema! type, version, lenient: true

  # store sxl version requested by site
  # TODO should check agaist site settings
  @site_sxl_version = message.attribute 'SXL'
rescue RSMP::Schemer::UnknownSchemaError => e
  dont_acknowledge message, "Rejected #{message.type} message,", "#{e}"
end

#collect_aggregated_status(task, options) ⇒ Object 



369
370
371
372
373
 
# File 'lib/rsmp/site_proxy.rb', line 369

def collect_aggregated_status task, options
  collector = AggregatedStatusCollector.new(self, options.merge(task:task))
  collector.collect
  collector
end

#collect_alarms(parent_task, options = {}) ⇒ Object 



342
343
344
345
346
347
348
349
 
# File 'lib/rsmp/site_proxy.rb', line 342

def collect_alarms parent_task, options={}
  collect(parent_task,options.merge(type: "Alarm")) do |alarm|
    next if options[:aCId] && options[:aCId] != alarm.attribute("aCId")
    next if options[:aSp] && options[:aSp] != alarm.attribute("aSp")
    next if options[:aS] && options[:aS] != alarm.attribute("aS")
    true
  end
end

#collect_command_responses(task, command_list, options) ⇒ Object 



363
364
365
366
367
 
# File 'lib/rsmp/site_proxy.rb', line 363

def collect_command_responses task, command_list, options
  collector = CommandResponseCollector.new(self, command_list, options.merge(task:task))
  collector.collect
  collector
end

#collect_status_responses(task, status_list, options) ⇒ Object 



357
358
359
360
361
 
# File 'lib/rsmp/site_proxy.rb', line 357

def collect_status_responses task, status_list, options
  collector = StatusResponseCollector.new(self, status_list, options.merge(task:task))
  collector.collect
  collector
end

#collect_status_updates(task, status_list, options) ⇒ Object 



351
352
353
354
355
 
# File 'lib/rsmp/site_proxy.rb', line 351

def collect_status_updates task, status_list, options
  collector = StatusUpdateCollector.new(self, status_list, options.merge(task:task))
  collector.collect
  collector
end

#connection_complete ⇒ Object 



43
44
45
46
47
 
# File 'lib/rsmp/site_proxy.rb', line 43

def connection_complete
  super
  sanitized_sxl_version = RSMP::Schemer.sanitize_version(@site_sxl_version)
  log "Connection to site #{@site_id} established, using core #{@rsmp_version}, #{@sxl} #{sanitized_sxl_version}", level: :info
end

#find_site_settings(site_id) ⇒ Object 



312
313
314
315
316
317
318
319
320
321
322
323
324
325
 
# File 'lib/rsmp/site_proxy.rb', line 312

def find_site_settings site_id
  if @settings['sites'] && @settings['sites'][@site_id]
    log "Using site settings for site id #{@site_id}", level: :debug
    return @settings['sites'][@site_id]
  end

  settings = @settings['guest']
  if @settings['guest']
    log "Using site settings for guest", level: :debug
    return @settings['guest']
  end
  
  nil
end

#inspect ⇒ Object 



24
25
26
27
28
 
# File 'lib/rsmp/site_proxy.rb', line 24

def inspect
  "#<#{self.class.name}:#{self.object_id}, #{inspector(
    :@acknowledgements,:@settings,:@site_settings,:@components
    )}>"
end

#node ⇒ Object 



29
30
31
 
# File 'lib/rsmp/site_proxy.rb', line 29

def node
  supervisor
end

#notify_error(e, options = {}) ⇒ Object 



337
338
339
340
 
# File 'lib/rsmp/site_proxy.rb', line 337

def notify_error e, options={}
  @supervisor.notify_error e, options if @supervisor
  distribute_error e, options
end

#process_aggregated_status(message) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/rsmp/site_proxy.rb', line 117

def process_aggregated_status message
  se = message.attribute("se")
  validate_aggregated_status(message,se) == false
  c_id = message.attributes["cId"]
  component = find_component c_id
  unless component
    reason = "component #{c_id} not found"
    dont_acknowledge message, "Ignoring #{message.type}:", reason
    return
  end

  component.set_aggregated_status_bools se
  log "Received #{message.type} status for component #{c_id} [#{component.aggregated_status.join(', ')}]", message: message
  acknowledge message
end

#process_alarm(message) ⇒ Object 



137
138
139
140
141
142
143
144
145
 
# File 'lib/rsmp/site_proxy.rb', line 137

def process_alarm message
  component = find_component message.attribute("cId")
  status = ["ack","aS","sS"].map { |key| message.attribute(key) }.join(',')
  component.handle_alarm message
  alarm_code = message.attribute("aCId")
  asp = message.attribute("aSp")
  log "Received #{message.type}, #{alarm_code} #{asp} [#{status}]", message: message, level: :log
  acknowledge message
end

#process_command_response(message) ⇒ Object 



77
78
79
80
 
# File 'lib/rsmp/site_proxy.rb', line 77

def process_command_response message
  log "Received #{message.type}", message: message, level: :log
  acknowledge message
end

#process_message(message) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/rsmp/site_proxy.rb', line 49

def process_message message
  case message
    when CommandRequest
    when StatusRequest
    when StatusSubscribe
    when StatusUnsubscribe
      will_not_handle message
    when AggregatedStatus
      process_aggregated_status message
    when AggregatedStatusRequest
      will_not_handle message
    when Alarm
      process_alarm message
    when CommandResponse
      process_command_response message
    when StatusResponse
      process_status_response message
    when StatusUpdate
      process_status_update message
    else
      super message
  end
rescue RSMP::RepeatedAlarmError, RSMP::RepeatedStatusError, TimestampError => e
  str = "Rejected #{message.type} message,"
  dont_acknowledge message, str, "#{e}"
  notify_error e.exception("#{str}#{e.message} #{message.json}")
end

#process_status_response(message) ⇒ Object 



182
183
184
185
186
187
 
# File 'lib/rsmp/site_proxy.rb', line 182

def process_status_response message
  component = find_component message.attribute("cId")
  component.handle_status_response message
  log "Received #{message.type}", message: message, level: :log
  acknowledge message
end

#process_status_update(message) ⇒ Object 



238
239
240
241
242
243
 
# File 'lib/rsmp/site_proxy.rb', line 238

def process_status_update message
  component = find_component message.attribute("cId")
  component.handle_status_update message
  log "Received #{message.type}", message: message, level: :log
  acknowledge message
end

#process_version(message) ⇒ Object 



295
296
297
298
299
300
301
 
# File 'lib/rsmp/site_proxy.rb', line 295

def process_version message
  return extraneous_version message if @version_determined
  check_site_ids message
  check_rsmp_version message
  check_sxl_version message
  version_accepted message
end

#process_watchdog(message) ⇒ Object 



151
152
153
154
155
156
 
# File 'lib/rsmp/site_proxy.rb', line 151

def process_watchdog message
  super
  if @watchdog_started == false
    start_watchdog
  end
end

#request_aggregated_status(component, options = {}) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/rsmp/site_proxy.rb', line 94

def request_aggregated_status component, options={}
  validate_ready 'request aggregated status'
  m_id = options[:m_id] || RSMP::Message.make_m_id

  message = RSMP::AggregatedStatusRequest.new({
      "ntsOId" => '',
      "xNId" => '',
      "cId" => component,
      "mId" => m_id
  })
  send_and_collect_if_needed message, options do |task|
    collect_aggregated_status task, options[:collect].merge(m_id: m_id, num:1)
  end
end

#request_status(component, status_list, options = {}) ⇒ Object 



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/rsmp/site_proxy.rb', line 162

def request_status component, status_list, options={}
  validate_ready 'request status'
  m_id = options[:m_id] || RSMP::Message.make_m_id

  # additional items can be used when verifying the response,
  # but must be removed from the request
  request_list = status_list.map { |item| item.slice('sCI','n') }

  message = RSMP::StatusRequest.new({
      "ntsOId" => '',
      "xNId" => '',
      "cId" => component,
      "sS" => request_list,
      "mId" => m_id
  })
  send_and_collect_if_needed message, options do |task|
    collect_status_responses task, status_list, options[:collect].merge(m_id: m_id)
  end
end

#revive(options) ⇒ Object 



17
18
19
20
21
 
# File 'lib/rsmp/site_proxy.rb', line 17

def revive options
  super options
  @supervisor = options[:supervisor]
  @settings = @supervisor.supervisor_settings.clone
end

#send_alarm_acknowledgement(component, alarm_code, options = {}) ⇒ Object 



245
246
247
248
249
250
251
252
 
# File 'lib/rsmp/site_proxy.rb', line 245

def send_alarm_acknowledgement component, alarm_code, options={}
  message = RSMP::AlarmAcknowledged.new({
      "cId" => component,
      "aCId" => alarm_code,
  })
  send_message message, validate: options[:validate]
  message
end

#send_and_collect_if_needed(message, options, &block) ⇒ Object 



189
190
191
192
193
194
195
196
197
198
 
# File 'lib/rsmp/site_proxy.rb', line 189

def send_and_collect_if_needed message, options, &block
  if options[:collect]
    task = @task.async { |task| yield task }
    send_message message, validate: options[:validate]
    { sent: message, collector: task.wait }
  else
    send_message message, validate: options[:validate]
    return { sent: message }
  end
end

#send_command(component, command_list, options = {}) ⇒ Object 



254
255
256
257
258
259
260
261
262
263
264
265
266
267
 
# File 'lib/rsmp/site_proxy.rb', line 254

def send_command component, command_list, options={}
  validate_ready 'send command'
  m_id = options[:m_id] || RSMP::Message.make_m_id
  message = RSMP::CommandRequest.new({
      "ntsOId" => '',
      "xNId" => '',
      "cId" => component,
      "arg" => command_list,
      "mId" => m_id
  })
  send_and_collect_if_needed message, options do |task|
    collect_command_responses task, command_list, options[:collect].merge(m_id: m_id)
  end
end

#set_watchdog_interval(interval) ⇒ Object 



269
270
271
 
# File 'lib/rsmp/site_proxy.rb', line 269

def set_watchdog_interval interval
  @settings['intervals']['watchdog'] = interval
end

#setup_site_settings ⇒ Object 



327
328
329
330
331
332
333
334
335
 
# File 'lib/rsmp/site_proxy.rb', line 327

def setup_site_settings
  @site_settings = find_site_settings @site_id
  if @site_settings
    @sxl = @site_settings['sxl']
    setup_components @site_settings['components']
  else
    dont_acknowledge message, 'Rejected', "No config found for site #{@site_id}"
  end
end

#site_ids_changed ⇒ Object 



158
159
160
 
# File 'lib/rsmp/site_proxy.rb', line 158

def site_ids_changed
  @supervisor.site_ids_changed
end

#start ⇒ Object 



33
34
35
36
 
# File 'lib/rsmp/site_proxy.rb', line 33

def start
  super
  start_reader
end

#stop ⇒ Object 



38
39
40
41
 
# File 'lib/rsmp/site_proxy.rb', line 38

def stop
  log "Closing connection to site", level: :info
  super
end

#subscribe_to_status(component_id, status_list, options = {}) ⇒ Object 



200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/rsmp/site_proxy.rb', line 200

def subscribe_to_status component_id, status_list, options={}
  validate_ready 'subscribe to status'
  m_id = options[:m_id] || RSMP::Message.make_m_id
  
  # additional items can be used when verifying the response,
  # but must to remove from the subscribe message
  subscribe_list = status_list.map { |item| item.slice('sCI','n','uRt') }

  component = find_component component_id
  component.handle_status_subscribe subscribe_list

  message = RSMP::StatusSubscribe.new({
      "ntsOId" => '',
      "xNId" => '',
      "cId" => component_id,
      "sS" => subscribe_list,
      'mId' => m_id
  })
  send_and_collect_if_needed message, options do |task|
     collect_status_updates task, status_list, options[:collect].merge(m_id: m_id)
  end
end

#sxl_version ⇒ Object 



289
290
291
292
293
 
# File 'lib/rsmp/site_proxy.rb', line 289

def sxl_version
  # a supervisor does not maintain it's own sxl version
  # instead we use what the site requests
  @site_sxl_version
end

#unsubscribe_to_status(component_id, status_list, options = {}) ⇒ Object 



223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/rsmp/site_proxy.rb', line 223

def unsubscribe_to_status component_id, status_list, options={}
  component = find_component component_id
  component.handle_status_subscribe status_list

  validate_ready 'unsubscribe to status'
  message = RSMP::StatusUnsubscribe.new({
      "ntsOId" => '',
      "xNId" => '',
      "cId" => component_id,
      "sS" => status_list
  })
  send_message message, validate: options[:validate]
  message
end

#validate_aggregated_status(message, se) ⇒ Object 



109
110
111
112
113
114
115
 
# File 'lib/rsmp/site_proxy.rb', line 109

def validate_aggregated_status  message, se
  unless se && se.is_a?(Array) && se.size == 8
    reason = "invalid AggregatedStatus, 'se' must be an Array of size 8"
    dont_acknowledge message, "Received", reaons
    raise InvalidMessage
  end
end

#validate_ready(action) ⇒ Object

Raises:



90
91
92
 
# File 'lib/rsmp/site_proxy.rb', line 90

def validate_ready action
  raise NotReady.new("Can't #{action} because connection is not ready. (Currently #{@state})") unless ready?
end

#version_accepted(message) ⇒ Object 



82
83
84
85
86
87
88
 
# File 'lib/rsmp/site_proxy.rb', line 82

def version_accepted message
  log "Received Version message for site #{@site_id}", message: message, level: :log
  start_timer
  acknowledge message
  send_version @site_id, rsmp_versions
  @version_determined = true
end

#version_acknowledged ⇒ Object 



147
148
149
 
# File 'lib/rsmp/site_proxy.rb', line 147

def version_acknowledged
  connection_complete
end