Class: RSMP::Proxy

Inherits:
Object
  • Object
show all
Includes:
Distributor, Logging, Modules::Acknowledgements, Modules::Receive, Modules::Send, Modules::State, Modules::Tasks, Modules::Versions, Modules::Watchdogs, Task
Defined in:
lib/rsmp/proxy/proxy.rb,
lib/rsmp/proxy/modules/send.rb,
lib/rsmp/proxy/modules/state.rb,
lib/rsmp/proxy/modules/tasks.rb,
lib/rsmp/proxy/modules/receive.rb,
lib/rsmp/proxy/modules/versions.rb,
lib/rsmp/proxy/modules/watchdogs.rb,
lib/rsmp/proxy/modules/acknowledgements.rb

Overview

Represents a connection to a remote site or supervisor. Provides common connection lifecycle and message handling.

Direct Known Subclasses

SiteProxy, SupervisorProxy

Defined Under Namespace

Modules: Modules

Constant Summary collapse

WRAPPING_DELIMITER = 
"\f".freeze

Instance Attribute Summary collapse

Attributes included from Task

#task

Attributes included from Distributor

#receivers

Attributes included from Logging

#logger

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Modules::Tasks

#read_line, #run_reader, #run_timer, #start_reader, #start_timer, #timer

Methods included from Modules::Versions

#check_core_version, #component_list_acknowledged, #configured_sxls, #core_3_3?, #core_versions, #extraneous_version, #normalized_core_versions, #primary_configured_sxl, #process_version, #send_version, #send_version_message, #send_version_request, #send_version_response, #site_id_items, #sxl_request_items, #version_acknowledged, #version_items, #version_meets_requirement?, #version_message_attributes, #version_request_attributes, #version_response_sxls

Methods included from Modules::Receive

#expect_version_message, #handle_fatal_error, #handle_invalid_message, #handle_invalid_packet, #handle_malformed_message, #handle_schema_error, #process_deferred, #process_message, #process_packet, #should_validate_ingoing_message?, #verify_sequence, #will_not_handle

Methods included from Modules::Send

#apply_nts_message_attributes, #buffer_message, #handle_send_schema_error, #log_send, #send_message, #send_message_and_collect

Methods included from Modules::Acknowledgements

#acknowledge, #acknowledged_first_ingoing, #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 Modules::Watchdogs

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

Methods included from Modules::State

#handshake_complete, #wait_for_state

Methods included from Task

#initialize_task, #restart, #run, #start, #stop, #task_status, #wait, #wait_for_condition

Methods included from Distributor

#add_receiver, #clear_deferred_distribution, #distribute, #distribute_error, #distribute_immediately, #distribute_queued, #initialize_distributor, #remove_receiver, #with_deferred_distribution

Methods included from Logging

#initialize_logging

Constructor Details

#initialize(options) ⇒ Proxy

Returns a new instance of Proxy.



24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/rsmp/proxy/proxy.rb', line 24

def initialize(options)
  @node = options[:node]
  options[:logger] = @node&.logger unless options[:logger] # default to node logger
  initialize_logging options
  initialize_distributor
  initialize_task
  setup options
  clear
  @state = :disconnected
  @state_condition = Async::Notification.new
end

Instance Attribute Details

#accepted_sxls ⇒ Object (readonly)

Returns the value of attribute accepted_sxls.



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

def accepted_sxls
  @accepted_sxls
end

#archive ⇒ Object (readonly)

Returns the value of attribute archive.



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

def archive
  @archive
end

#collector ⇒ Object (readonly)

Returns the value of attribute collector.



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

def collector
  @collector
end

#connection_info ⇒ Object (readonly)

Returns the value of attribute connection_info.



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

def connection_info
  @connection_info
end

#core_version ⇒ Object (readonly)

Returns the value of attribute core_version.



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

def core_version
  @core_version
end

#ip ⇒ Object (readonly)

Returns the value of attribute ip.



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

def ip
  @ip
end

#node ⇒ Object (readonly)

Returns the value of attribute node.



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

def node
  @node
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



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

def port
  @port
end

#rejected_sxls ⇒ Object (readonly)

Returns the value of attribute rejected_sxls.



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

def rejected_sxls
  @rejected_sxls
end

#site_settings ⇒ Object (readonly)

Returns the value of attribute site_settings.



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

def site_settings
  @site_settings
end

#state ⇒ Object

Returns the value of attribute state.



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

def state
  @state
end

#sxl_interfaces ⇒ Object (readonly)

Returns the value of attribute sxl_interfaces.



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

def sxl_interfaces
  @sxl_interfaces
end

#sxls ⇒ Object (readonly)

Returns the value of attribute sxls.



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

def sxls
  @sxls
end

Class Method Details

.version_meets_requirement?(version, requirement) ⇒ Boolean

Use Gem class to check version requirement Requirement must be a string like ‘1.1’, ‘>=1.0.3’ or ‘<2.1.4’, or list of strings, like [‘<=1.4’,‘<1.5’]

Returns:

  • (Boolean) 


240
241
242
 
# File 'lib/rsmp/proxy/proxy.rb', line 240

def self.version_meets_requirement?(version, requirement)
  Modules::Versions.version_meets_requirement?(version, requirement)
end

Instance Method Details

#author ⇒ Object 



210
211
212
 
# File 'lib/rsmp/proxy/proxy.rb', line 210

def author
  @node.site_id
end

#build_sxl_interfaces ⇒ Object 



214
215
216
217
218
 
# File 'lib/rsmp/proxy/proxy.rb', line 214

def build_sxl_interfaces
  @sxl_interfaces = accepted_sxls.to_h do |sxl|
    [sxl['name'], RSMP::SXL::Registry.build_for(self, sxl)]
  end
end

#clear ⇒ Object 



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/rsmp/proxy/proxy.rb', line 133

def clear
  @awaiting_acknowledgement = {}
  @latest_watchdog_received = nil
  @watchdog_started = false
  @version_determined = false
  @ingoing_acknowledged = {}
  @outgoing_acknowledged = {}
  @latest_watchdog_send_at = nil
  @component_list_received = false
  @outgoing_watchdog_acknowledged = false
  @sxl_interfaces = {}

  @acknowledgements = {}
  @acknowledgement_condition = Async::Notification.new
end

#clock ⇒ Object 



173
174
175
 
# File 'lib/rsmp/proxy/proxy.rb', line 173

def clock
  @node.clock
end

#close ⇒ Object

close connection, but keep our main task running so we can reconnect



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

def close
  log 'Closing connection', level: :warning
  close_stream
  close_socket
  stop_reader
  self.state = :disconnected
  distribute_error DisconnectError.new('Connection was closed')

  # stop timer
  # as we're running inside the timer, code after stop_timer() will not be called,
  # unless it's in the ensure block
  stop_timer
end

#close_socket ⇒ Object 



94
95
96
97
98
 
# File 'lib/rsmp/proxy/proxy.rb', line 94

def close_socket
  @socket&.close
ensure
  @socket = nil
end

#close_stream ⇒ Object 



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

def close_stream
  @stream&.close
ensure
  @stream = nil
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


111
112
113
 
# File 'lib/rsmp/proxy/proxy.rb', line 111

def connected?
  @state == :connected || @state == :ready
end

#disconnect ⇒ Object

Connection lifecycle methods



46
 
# File 'lib/rsmp/proxy/proxy.rb', line 46

def disconnect; end

#disconnected? ⇒ Boolean

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/rsmp/proxy/proxy.rb', line 115

def disconnected?
  @state == :disconnected
end

#inspect ⇒ Object 



36
37
38
 
# File 'lib/rsmp/proxy/proxy.rb', line 36

def inspect
  "#<#{self.class.name}:#{object_id} state:#{state}}>"
end

#log(str, options = {}) ⇒ Object 



181
182
183
 
# File 'lib/rsmp/proxy/proxy.rb', line 181

def log(str, options = {})
  super(str, options.merge(ip: @ip, port: @port, site_id: @site_id))
end

#now ⇒ Object 



40
41
42
 
# File 'lib/rsmp/proxy/proxy.rb', line 40

def now
  node.now
end

#primary_sxl ⇒ Object 



198
199
200
 
# File 'lib/rsmp/proxy/proxy.rb', line 198

def primary_sxl
  accepted_sxls.first || sxls.first
end

#ready? ⇒ Boolean

State management methods

Returns:

  • (Boolean) 


107
108
109
 
# File 'lib/rsmp/proxy/proxy.rb', line 107

def ready?
  @state == :ready
end

#receive_alarms? ⇒ Boolean

Returns:

  • (Boolean) 


194
195
196
 
# File 'lib/rsmp/proxy/proxy.rb', line 194

def receive_alarms?
  @receive_alarms != false
end

#receive_error(error, options = {}) ⇒ Object 



177
178
179
 
# File 'lib/rsmp/proxy/proxy.rb', line 177

def receive_error(error, options = {})
  @node.receive_error error, options
end

#revive(options) ⇒ Object

revive after a reconnect



150
151
152
 
# File 'lib/rsmp/proxy/proxy.rb', line 150

def revive(options)
  setup options
end

#schemas ⇒ Object 



185
186
187
188
189
190
191
192
 
# File 'lib/rsmp/proxy/proxy.rb', line 185

def schemas
  schemas = { core: RSMP::Schema.latest_core_version } # use latest core
  schemas[:core] = core_version if core_version
  accepted_sxls.each do |sxl|
    schemas[sxl['name'].to_sym] = RSMP::Schema.sanitize_version(sxl['version'].to_s)
  end
  schemas
end

#setup(options) ⇒ Object 



154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/rsmp/proxy/proxy.rb', line 154

def setup(options)
  @settings = options[:settings]
  @socket = options[:socket]
  @stream = options[:stream]
  @protocol = options[:protocol]
  @ip = options[:ip]
  @port = options[:port]
  @connection_info = options[:info]
  @sxls = []
  @accepted_sxls = []
  @rejected_sxls = []
  @receive_alarms = true
  @site_settings = nil # can't pick until we know the site id
  return unless options[:collect]

  @collector = RSMP::Collector.new self, options[:collect]
  @collector.start
end

#state_changed ⇒ Object

the state changed override to to things like notifications



129
130
131
 
# File 'lib/rsmp/proxy/proxy.rb', line 129

def state_changed
  @state_condition.signal @state
end

#stop_reader ⇒ Object 



82
83
84
85
86
 
# File 'lib/rsmp/proxy/proxy.rb', line 82

def stop_reader
  @reader&.stop
ensure
  @reader = nil
end

#stop_subtasks ⇒ Object 



69
70
71
72
73
74
 
# File 'lib/rsmp/proxy/proxy.rb', line 69

def stop_subtasks
  stop_timer
  stop_reader
  clear
  super
end

#stop_task ⇒ Object 



100
101
102
103
 
# File 'lib/rsmp/proxy/proxy.rb', line 100

def stop_task
  close
  super
end

#stop_timer ⇒ Object 



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

def stop_timer
  @timer&.stop
ensure
  @timer = nil
end

#sxl ⇒ Object 



202
203
204
 
# File 'lib/rsmp/proxy/proxy.rb', line 202

def sxl
  primary_sxl && primary_sxl['name']
end

#sxl_interface(name) ⇒ Object 



220
221
222
223
224
 
# File 'lib/rsmp/proxy/proxy.rb', line 220

def sxl_interface(name)
  sxl_interfaces.fetch(name.to_s) do
    raise RSMP::Schema::UnknownSchemaTypeError, "SXL #{name} is not accepted on this connection"
  end
end

#sxl_interface_for(message) ⇒ Object 



230
231
232
233
234
235
 
# File 'lib/rsmp/proxy/proxy.rb', line 230

def sxl_interface_for(message)
  resolved = RSMP::Schema.resolve_sxl(message.attributes, schemas: schemas)
  return unless resolved

  sxl_interface resolved.first
end

#sxl_version ⇒ Object 



206
207
208
 
# File 'lib/rsmp/proxy/proxy.rb', line 206

def sxl_version
  primary_sxl && primary_sxl['version']
end

#tlc ⇒ Object 



226
227
228
 
# File 'lib/rsmp/proxy/proxy.rb', line 226

def tlc
  sxl_interface 'tlc'
end

#wait_for_reader ⇒ Object

wait for the reader task to complete, which is not expected to happen before the connection is closed



50
51
52
 
# File 'lib/rsmp/proxy/proxy.rb', line 50

def wait_for_reader
  @reader&.wait
end