Class: OpenC3::TcpipServerInterface

Inherits:
StreamInterface show all
Defined in:
lib/openc3/interfaces/tcpip_server_interface.rb

Overview

TCP/IP Server which can both read and write on a single port or two independent ports. A listen thread is setup which waits for client connections. For each connection to the read port, a thread is spawned that calls the read method from the interface. This data is then available by calling the TcpipServer read method. For each connection to the write port, a thread is spawned that calls the write method from the interface when data is send to the TcpipServer via the write method.

Defined Under Namespace

Classes: InterfaceInfo

Constant Summary

Constants included from Api

Api::DELAY_METRICS, Api::DURATION_METRICS, Api::SUBSCRIPTION_DELIMITER, Api::SUM_METRICS

Constants included from ApiShared

ApiShared::DEFAULT_TLM_POLLING_RATE

Constants included from Extract

Extract::SCANNING_REGULAR_EXPRESSION

Instance Attribute Summary collapse

Attributes inherited from StreamInterface

#stream

Attributes inherited from Interface

#auto_reconnect, #bytes_read, #bytes_written, #cmd_target_enabled, #cmd_target_names, #config_params, #connect_on_startup, #disable_disconnect, #interfaces, #name, #options, #protocol_info, #read_count, #read_protocols, #read_raw_data, #read_raw_data_time, #reconnect_delay, #routers, #save_raw_data, #scheduler, #secrets, #state, #target_names, #tlm_target_enabled, #tlm_target_names, #write_count, #write_protocols, #written_raw_data, #written_raw_data_time

Instance Method Summary collapse

Methods inherited from StreamInterface

#read_interface, #write_interface

Methods inherited from Interface

#_write, #add_protocol, #as_json, #convert_data_to_packet, #convert_packet_to_data, #copy_to, #interface_cmd, #post_connect, #protocol_cmd, #read_allowed?, #read_interface, #read_interface_base, #write_allowed?, #write_interface, #write_interface_base, #write_raw_allowed?

Methods included from Api

#_cmd_implementation, #_extract_target_command_names, #_extract_target_command_parameter_names, #_extract_target_packet_item_names, #_extract_target_packet_names, #_get_and_set_cmd, #_get_item, #_limits_group, #_set_tlm_process_args, #_tlm_process_args, #_validate_tlm_type, #build_cmd, #cmd, #cmd_no_checks, #cmd_no_hazardous_check, #cmd_no_range_check, #cmd_raw, #cmd_raw_no_checks, #cmd_raw_no_hazardous_check, #cmd_raw_no_range_check, #config_tool_names, #connect_interface, #connect_router, #delete_config, #disable_cmd, #disable_limits, #disable_limits_group, #disconnect_interface, #disconnect_router, #enable_cmd, #enable_limits, #enable_limits_group, #get_all_cmd_names, #get_all_cmds, #get_all_interface_info, #get_all_router_info, #get_all_settings, #get_all_tlm, #get_all_tlm_item_names, #get_all_tlm_names, #get_cmd, #get_cmd_buffer, #get_cmd_cnt, #get_cmd_cnts, #get_cmd_hazardous, #get_cmd_time, #get_cmd_value, #get_interface, #get_interface_names, #get_item, #get_limits, #get_limits_events, #get_limits_groups, #get_limits_set, #get_limits_sets, #get_metrics, #get_out_of_limits, #get_overall_limits_state, #get_overrides, #get_packet_derived_items, #get_packets, #get_param, #get_router, #get_router_names, #get_setting, #get_settings, #get_target, #get_target_interfaces, #get_target_names, #get_tlm, #get_tlm_available, #get_tlm_buffer, #get_tlm_cnt, #get_tlm_cnts, #get_tlm_packet, #get_tlm_values, #inject_tlm, #interface_cmd, #interface_details, #interface_protocol_cmd, #interface_target_disable, #interface_target_enable, #limits_enabled?, #list_configs, #list_settings, #load_config, #map_target_to_interface, #map_target_to_router, #normalize_tlm, #offline_access_needed, #override_tlm, #router_cmd, #router_details, #router_protocol_cmd, #router_target_disable, #router_target_enable, #save_config, #send_raw, #set_limits, #set_limits_set, #set_offline_access, #set_setting, #set_tlm, #start_raw_logging_interface, #start_raw_logging_router, #stash_all, #stash_delete, #stash_get, #stash_keys, #stash_set, #stop_raw_logging_interface, #stop_raw_logging_router, #subscribe_packets, #tlm, #tlm_formatted, #tlm_raw, #tlm_with_units, #unmap_target_from_interface, #unmap_target_from_router, #update_news, #update_plugin_store

Methods included from CmdLog

#_build_cmd_output_string

Constructor Details

#initialize(write_port, read_port, write_timeout, read_timeout, protocol_type = nil, *protocol_args) ⇒ TcpipServerInterface

Returns a new instance of TcpipServerInterface.

Parameters:

  • write_port (Integer) —

    The server write port. Clients should connect and expect to receive data from this port.

  • read_port (Integer) —

    The server read port. Clients should connect and expect to send data to this port.

  • write_timeout (Float) —

    Seconds to wait before aborting writes

  • read_timeout (Float|nil) —

    Seconds to wait before aborting reads. Pass nil to block until the read is complete.

  • protocol_type (String) (defaults to: nil) —

    The name of the stream to use for both the read and write ports. This name is combined with 'Protocol' to result in a OpenC3 Protocol class.

  • protocol_args (Array) —

    Arguments to pass to the Protocol



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 67

def initialize(write_port,
               read_port,
               write_timeout,
               read_timeout,
               protocol_type = nil,
               *protocol_args)
  super(protocol_type, protocol_args)
  @write_port = ConfigParser.handle_nil(write_port)
  @write_port = Integer(write_port) if @write_port
  @read_port = ConfigParser.handle_nil(read_port)
  @read_port = Integer(read_port) if @read_port
  @write_timeout = ConfigParser.handle_nil(write_timeout)
  @write_timeout = @write_timeout.to_f if @write_timeout
  @read_timeout = ConfigParser.handle_nil(read_timeout)
  @read_timeout = @read_timeout.to_f if @read_timeout
  @listen_sockets = []
  @listen_pipes = []
  @listen_threads = []
  @read_threads = []
  @write_thread = nil
  @write_raw_thread = nil
  @write_interface_infos = []
  @read_interface_infos = []
  @write_queue = nil
  @write_queue = Queue.new if @write_port
  @write_raw_queue = nil
  @write_raw_queue = Queue.new if @write_port
  @read_queue = nil
  @read_queue = Queue.new if @read_port
  @write_condition_variable = nil
  @write_condition_variable = ConditionVariable.new if @write_port
  @write_raw_mutex = nil
  @write_raw_mutex = Mutex.new if @write_port
  @write_raw_condition_variable = nil
  @write_raw_condition_variable = ConditionVariable.new if @write_port
  @write_connection_callback = nil
  @read_connection_callback = nil
  @stream_log_pair = nil
  @raw_logging_enabled = false
  @connection_mutex = Mutex.new
  @listen_address = "0.0.0.0"

  @read_allowed = false unless ConfigParser.handle_nil(read_port)
  @write_allowed = false unless ConfigParser.handle_nil(write_port)
  @write_raw_allowed = false unless ConfigParser.handle_nil(write_port)

  @connected = false
end

Instance Attribute Details

#listen_address ⇒ String

Returns The ip address to bind to. Default to ANY (0.0.0.0).

Returns:

  • (String) —

    The ip address to bind to. Default to ANY (0.0.0.0)



54
55
56
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 54

def listen_address
  @listen_address
end

#read_connection_callback ⇒ Object

Callback method to call when a new client connects to the read port. This method will be called with the Interface as the only argument.



50
51
52
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 50

def read_connection_callback
  @read_connection_callback
end

#stream_log_pair ⇒ StreamLogPair

Returns StreamLogPair instance or nil.

Returns:



52
53
54
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 52

def stream_log_pair
  @stream_log_pair
end

#write_connection_callback ⇒ Object

Callback method to call when a new client connects to the write port. This method will be called with the Interface as the only argument.



47
48
49
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 47

def write_connection_callback
  @write_connection_callback
end

Instance Method Details

#change_raw_logging(method) ⇒ Object 



307
308
309
310
311
312
313
314
315
316
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 307

def change_raw_logging(method)
  if @stream_log_pair
    @write_interface_infos.each do |interface_info|
      interface_info.interface.stream_log_pair.public_send(method) if interface_info.interface.stream_log_pair
    end
    @read_interface_infos.each do |interface_info|
      interface_info.interface.stream_log_pair.public_send(method) if interface_info.interface.stream_log_pair
    end
  end
end

#check_for_dead_clients ⇒ Object 



543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 543

def check_for_dead_clients
  indexes_to_delete = []
  index = 0

  @connection_mutex.synchronize do
    @write_interface_infos.each do |interface_info|
      if @write_port != @read_port
        # Socket should return EWOULDBLOCK if it is still cleanly connected
        interface_info.interface.stream.write_socket.recvfrom_nonblock(10)
      elsif !interface_info.interface.stream.write_socket.closed?
        # Let read thread detect disconnect
        next
      end
      # Client has disconnected (or is invalidly sending data on the socket)
      Logger.info "#{@name}: Tcpip server lost write connection to #{interface_info.hostname}(#{interface_info.host_ip}):#{interface_info.port}"
      interface_info.interface.disconnect
      interface_info.interface.stream_log_pair.stop if interface_info.interface.stream_log_pair
      indexes_to_delete.unshift(index) # Put later indexes at front of array
    rescue Errno::ECONNRESET, Errno::ECONNABORTED, IOError
      # Client has disconnected
      Logger.info "#{@name}: Tcpip server lost write connection to #{interface_info.hostname}(#{interface_info.host_ip}):#{interface_info.port}"
      interface_info.interface.disconnect
      interface_info.interface.stream_log_pair.stop if interface_info.interface.stream_log_pair
      indexes_to_delete.unshift(index) # Put later indexes at front of array
    rescue Errno::EWOULDBLOCK
      # Client is still cleanly connected as far as we can tell without writing to the socket
    ensure
      index += 1
    end

    # Delete any dead sockets
    indexes_to_delete.each do |index_to_delete|
      @write_interface_infos.delete_at(index_to_delete)
    end
  end # connection_mutex.synchronize

  # Sleep until we receive a packet or for 100ms
  @write_mutex.synchronize do
    @write_condition_variable.wait(@write_mutex, 0.1)
  end
end

#connect ⇒ Object

Create the read and write port listen threads. Incoming connections will spawn separate threads to process the reads and writes.



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 128

def connect
  @cancel_threads = false
  @read_queue.clear if @read_queue
  if @write_port == @read_port # One socket
    start_listen_thread(@read_port, true, true)
  else
    start_listen_thread(@write_port, true, false) if @write_port
    start_listen_thread(@read_port, false, true) if @read_port
  end

  if @write_port
    @write_thread = Thread.new do
      loop do
        write_thread_body()
        break if @cancel_threads
      end
    rescue Exception => e
      shutdown_interfaces(@write_interface_infos)
      Logger.error("#{@name}: Tcpip server write thread unexpectedly died")
      Logger.error(e.formatted)
    end
    @write_raw_thread = Thread.new do
      loop do
        write_raw_thread_body()
        break if @cancel_threads
      end
    rescue Exception => e
      shutdown_interfaces(@write_interface_infos)
      Logger.error("#{@name}: Tcpip server write raw thread unexpectedly died")
      Logger.error(e.formatted)
    end
  else
    @write_thread = nil
    @write_raw_thread = nil
  end
  super()
  @connected = true
end

#connected? ⇒ Boolean

Returns Whether the server is listening for connections.

Returns:

  • (Boolean) —

    Whether the server is listening for connections



168
169
170
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 168

def connected?
  @connected
end

#connection_string ⇒ Object 



116
117
118
119
120
121
122
123
124
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 116

def connection_string
  if @write_port == @read_port
    return "listening on #{@listen_address}:#{@write_port} (R/W)"
  end
  result = "listening on"
  result += " #{@listen_address}:#{@write_port} (write)" if @write_port
  result += " #{@listen_address}:#{@read_port} (read)" if @read_port
  return result
end

#details ⇒ Object 



625
626
627
628
629
630
631
632
633
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 625

def details
  result = super()
  result['write_port'] = @write_port
  result['read_port'] = @read_port
  result['write_timeout'] = @write_timeout
  result['read_timeout'] = @read_timeout
  result['listen_address'] = @listen_address
  return result
end

#disconnect ⇒ Object

Shutdowns the listener threads for both the read and write ports as well as any client connections.



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 174

def disconnect
  @cancel_threads = true
  @read_queue << nil if @read_queue
  @listen_pipes.each do |pipe|
    pipe.write('.')
  rescue Exception
    # Oh well
  end
  @listen_pipes.clear

  # Shutdown listen thread(s)
  @listen_threads.each { |listen_thread| OpenC3.kill_thread(self, listen_thread) }
  @listen_threads.clear

  # Shutdown listen socket(s)
  @listen_sockets.each do |listen_socket|
    OpenC3.close_socket(listen_socket)
  rescue IOError
    # Ok may have been closed by the thread
  end
  @listen_sockets.clear

  # This will unblock read threads
  shutdown_interfaces(@read_interface_infos)

  @read_threads.each { |thread| OpenC3.kill_thread(self, thread) }
  @read_threads.clear
  if @write_thread
    OpenC3.kill_thread(self, @write_thread)
    @write_thread = nil
  end
  if @write_raw_thread
    OpenC3.kill_thread(self, @write_raw_thread)
    @write_raw_thread = nil
  end

  shutdown_interfaces(@write_interface_infos)
  @connected = false
  super()
end

#graceful_kill ⇒ Object

Gracefully kill all the threads



216
217
218
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 216

def graceful_kill
  # This method is just here to prevent warnings
end

#listen_thread_body(listen_socket, listen_write, listen_read, thread_reader) ⇒ Object 



362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 362

def listen_thread_body(listen_socket, listen_write, listen_read, thread_reader)
  begin
    socket, address = listen_socket.accept_nonblock
  rescue Errno::EAGAIN, Errno::ECONNABORTED, Errno::EINTR, Errno::EWOULDBLOCK
    read_ready, _ = IO.select([listen_socket, thread_reader])
    if read_ready && read_ready.include?(thread_reader)
      return
    else
      retry
    end
  end

  port, host_ip = Socket.unpack_sockaddr_in(address)
  hostname = ''
  hostname = Socket.lookup_hostname_from_ip(host_ip)
  # if System.instance.acl
  #   addr = ["AF_INET", 10, "lc630", host_ip.to_s]
  #   if not System.instance.acl.allow_addr?(addr)
  #     # Reject connection
  #     OpenC3.close_socket(socket)
  #     Logger.info "#{@name}: Tcpip server rejected connection from #{hostname}(#{host_ip}):#{port}"
  #     return
  #   end
  # end

  # Configure TCP_NODELAY option
  socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)

  # Accept Connection
  write_socket = nil
  read_socket = nil
  write_socket = socket if listen_write
  read_socket = socket if listen_read
  stream = TcpipSocketStream.new(write_socket, read_socket, @write_timeout, @read_timeout)

  # Pass down options to the stream
  @options.each do |option_name, option_values|
    stream.set_option(option_name, option_values)
  end

  interface = StreamInterface.new
  interface.target_names = @target_names
  interface.cmd_target_names = @cmd_target_names
  interface.tlm_target_names = @tlm_target_names
  if @stream_log_pair
    interface.stream_log_pair = @stream_log_pair.clone
    interface.stream_log_pair.start if @raw_logging_enabled
  end
  @protocol_info.each do |protocol_class, protocol_args, read_write|
    interface.add_protocol(protocol_class, protocol_args, read_write)
  end
  interface.stream = stream
  interface.connect

  if listen_write
    @write_connection_callback.call(interface) if @write_connection_callback
    @connection_mutex.synchronize do
      @write_interface_infos << InterfaceInfo.new(interface, hostname, host_ip, port)
    end
  end
  if listen_read
    @read_connection_callback.call(interface) if @read_connection_callback
    @connection_mutex.synchronize do
      @read_interface_infos << InterfaceInfo.new(interface, hostname, host_ip, port)
    end
    start_read_thread(@read_interface_infos[-1])
  end
  Logger.info "#{@name}: Tcpip server accepted connection from #{hostname}(#{host_ip}):#{port}"
end

#num_clients ⇒ Integer

Returns The number of connected clients.

Returns:

  • (Integer) —

    The number of connected clients



266
267
268
269
270
271
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 266

def num_clients
  interfaces = []
  @write_interface_infos.each { |wii| interfaces << wii.interface }
  @read_interface_infos.each { |rii| interfaces << rii.interface }
  interfaces.uniq.length
end

#read ⇒ Packet

Returns Latest packet read from any of the connected clients. Note this method blocks until data is available.

Returns:

  • (Packet) —

    Latest packet read from any of the connected clients. Note this method blocks until data is available.



222
223
224
225
226
227
228
229
230
231
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 222

def read
  raise "Interface not connected for read: #{@name}" unless connected?
  raise "Interface not readable: #{@name}" unless read_allowed?

  packet = @read_queue.pop
  return nil unless packet

  @read_count += 1
  packet
end

#read_queue_size ⇒ Integer

Returns The number of packets waiting on the read queue.

Returns:

  • (Integer) —

    The number of packets waiting on the read queue



256
257
258
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 256

def read_queue_size
  @read_queue ? @read_queue.size : 0
end

#read_thread_body(interface) ⇒ Object 



519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 519

def read_thread_body(interface)
  thread_bytes_read = 0
  loop do
    packet = interface.read
    interface_bytes_read = interface.bytes_read
    if interface_bytes_read != thread_bytes_read
      diff = interface_bytes_read - thread_bytes_read
      @bytes_read += diff # This would be better if mutex protected, but not that important for telemetry
      thread_bytes_read = interface_bytes_read
    end
    return if !packet || @cancel_threads

    packet = read_thread_hook(packet) # Do work on received packet
    @read_raw_data_time = interface.read_raw_data_time
    @read_raw_data = interface.read_raw_data
    @read_queue << packet.clone
  end
end

#read_thread_hook(packet) ⇒ Packet

Return the packet

Returns:

  • (Packet) —

    Return the packet



539
540
541
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 539

def read_thread_hook(packet)
  packet
end

#set_option(option_name, option_values) ⇒ Object

Supported Options LISTEN_ADDRESS - Ip address of the interface to accept connections on - Default: 0.0.0.0 (see Interface#set_option)



288
289
290
291
292
293
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 288

def set_option(option_name, option_values)
  super(option_name, option_values)
  if option_name.upcase == 'LISTEN_ADDRESS'
    @listen_address = option_values[0]
  end
end

#shutdown_interfaces(interface_infos) ⇒ Object

protected



297
298
299
300
301
302
303
304
305
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 297

def shutdown_interfaces(interface_infos)
  @connection_mutex.synchronize do
    interface_infos.each do |interface_info|
      interface_info.interface.disconnect
      interface_info.interface.stream_log_pair.stop if interface_info.interface.stream_log_pair
    end
    interface_infos.clear
  end
end

#start_listen_thread(port, listen_write = false, listen_read = false) ⇒ Object 



318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 318

def start_listen_thread(port, listen_write = false, listen_read = false)
  # Create a socket to accept connections from clients
  addr = Socket.pack_sockaddr_in(port, @listen_address)
  if RUBY_ENGINE == 'ruby'
    listen_socket = Socket.new(Socket::AF_INET, Socket::SOCK_STREAM, 0)
    listen_socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_REUSEADDR, 1) unless Kernel.is_windows?
    begin
      listen_socket.bind(addr)
    rescue Errno::EADDRINUSE
      raise "Error binding to port #{port}.\n" +
            "Either another application is using this port\n" +
            "or the operating system is being slow cleaning up.\n" +
            "Make sure all sockets/streams are closed in all applications,\n" +
            "wait 1 minute and try again."
    end

    listen_socket.listen(5)
  else
    listen_socket = ServerSocket.new(Socket::AF_INET, Socket::SOCK_STREAM, 0)
    listen_socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_REUSEADDR, 1) unless Kernel.is_windows?
    begin
      listen_socket.bind(addr, 5)
    rescue Errno::EADDRINUSE
      raise "Error binding to port #{port}.\n" +
            "Either another application is using this port\n" +
            "or the operating system is being slow cleaning up.\n" +
            "Make sure all sockets/streams are closed in all applications,\n" +
            "wait 1 minute and try again."
    end
  end
  @listen_sockets << listen_socket
  @listen_threads << Thread.new do
    thread_reader, thread_writer = IO.pipe
    @listen_pipes << thread_writer
    loop do
      listen_thread_body(listen_socket, listen_write, listen_read, thread_reader)
      break if @cancel_threads
    end
  rescue => e
    Logger.error("#{@name}: Tcpip server listen thread unexpectedly died")
    Logger.error(e.formatted)
  end
end

#start_raw_logging ⇒ Object

Start raw logging for this interface



274
275
276
277
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 274

def start_raw_logging
  @raw_logging_enabled = true
  change_raw_logging(:start)
end

#start_read_thread(interface_info) ⇒ Object 



432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 432

def start_read_thread(interface_info)
  @read_threads << Thread.new do
    index_to_delete = nil
    begin
      begin
        read_thread_body(interface_info.interface)
      rescue Exception => e
        Logger.error "#{@name}: Tcpip server read thread unexpectedly died"
        Logger.error e.formatted
      end
      Logger.info "#{@name}: Tcpip server lost read connection to #{interface_info.hostname}(#{interface_info.host_ip}):#{interface_info.port}"
      @read_threads.delete(Thread.current)

      index_to_delete = nil
      @connection_mutex.synchronize do
        index = 0
        @read_interface_infos.each do |read_interface_info|
          if interface_info.interface == read_interface_info.interface
            index_to_delete = index
            read_interface_info.interface.disconnect
            read_interface_info.interface.stream_log_pair.stop if read_interface_info.interface.stream_log_pair
            break
          end
          index += 1
        end
      ensure
        if index_to_delete
          @read_interface_infos.delete_at(index_to_delete)
        end
      end
    rescue Exception => e
      Logger.error "#{@name}: Tcpip server read thread unexpectedly died"
      Logger.error e.formatted
    end
  end
end

#stop_raw_logging ⇒ Object

Stop raw logging for this interface



280
281
282
283
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 280

def stop_raw_logging
  @raw_logging_enabled = false
  change_raw_logging(:stop)
end

#write(packet) ⇒ Object

Parameters:

  • packet (Packet) —

    Packet to write to all clients connected to the write port.



235
236
237
238
239
240
241
242
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 235

def write(packet)
  raise "Interface not connected for write: #{@name}" unless connected?
  raise "Interface not writable: #{@name}" unless write_allowed?

  @write_count += 1
  @write_queue << packet.clone
  @write_condition_variable.broadcast
end

#write_queue_size ⇒ Integer

Returns The number of packets waiting on the write queue.

Returns:

  • (Integer) —

    The number of packets waiting on the write queue



261
262
263
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 261

def write_queue_size
  @write_queue ? @write_queue.size : 0
end

#write_raw(data) ⇒ Object

Parameters:

  • data (String) —

    Data to write to all clients connected to the write port.



246
247
248
249
250
251
252
253
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 246

def write_raw(data)
  raise "Interface not connected for write_raw: #{@name}" unless connected?
  raise "Interface not write-rawable: #{@name}" unless write_raw_allowed?

  @write_raw_queue << data
  @write_raw_condition_variable.broadcast
  return data
end

#write_raw_thread_body ⇒ Object 



489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 489

def write_raw_thread_body
  # Retrieve the next data to be sent out to clients
  data = nil

  loop do
    break if @cancel_threads

    begin
      data = @write_raw_queue.pop(true) # non_block to raise ThreadError
      break
    rescue ThreadError
      # Sleep until we receive data or for 100ms
      @write_raw_mutex.synchronize do
        @write_raw_condition_variable.wait(@write_raw_mutex, 0.1)
      end
    end
  end

  data = write_raw_thread_hook(data)
  write_to_clients(:write_raw, data) if data
end

#write_raw_thread_hook(data) ⇒ Object 



515
516
517
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 515

def write_raw_thread_hook(data)
  data # By default just return the data
end

#write_thread_body ⇒ Object 



469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 469

def write_thread_body
  # Retrieve the next packet to be sent out to clients
  # Handles disconnected clients even when packets aren't flowing
  packet = nil

  loop do
    break if @cancel_threads

    begin
      packet = @write_queue.pop(true) # non_block to raise ThreadError
      break
    rescue ThreadError
      check_for_dead_clients()
    end
  end

  packet = write_thread_hook(packet)
  write_to_clients(:write, packet) if packet
end

#write_thread_hook(packet) ⇒ Object 



511
512
513
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 511

def write_thread_hook(packet)
  packet # By default just return the packet
end

#write_to_clients(method, packet_or_data) ⇒ Object 



585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
 
# File 'lib/openc3/interfaces/tcpip_server_interface.rb', line 585

def write_to_clients(method, packet_or_data)
  @connection_mutex.synchronize do
    # Send data to each client - On error drop the client
    indexes_to_delete = []
    index = 0
    @write_interface_infos.each do |interface_info|
      need_disconnect = false
      begin
        interface_bytes_written = interface_info.interface.bytes_written
        interface_info.interface.public_send(method, packet_or_data)
        diff = interface_info.interface.bytes_written - interface_bytes_written
        @written_raw_data_time = interface_info.interface.written_raw_data_time
        @written_raw_data = interface_info.interface.written_raw_data
        @bytes_written += diff
      rescue Errno::EPIPE, Errno::ECONNABORTED, IOError, Errno::ECONNRESET
        # Client has normally disconnected
        need_disconnect = true
      rescue Exception => e
        if e.message != "Stream not connected for write_raw"
          Logger.error "#{@name}: Error sending to client: #{e.class} #{e.message}"
        end
        need_disconnect = true
      end

      if need_disconnect
        Logger.info "#{@name}: Tcpip server lost write connection to #{interface_info.hostname}(#{interface_info.host_ip}):#{interface_info.port}"
        interface_info.interface.disconnect
        interface_info.interface.stream_log_pair.stop if interface_info.interface.stream_log_pair
        indexes_to_delete.unshift(index) # Put later indexes at front of array
      end
      index += 1
    end

    # Delete any dead sockets
    indexes_to_delete.each do |index_to_delete|
      @write_interface_infos.delete_at(index_to_delete)
    end
  end # connection_mutex.synchronize
end