Class: Cosmos::InterfaceMicroservice

# File 'lib/cosmos/microservices/interface_microservice.rb', line 240

def initialize(name)
  super(name)
  @interface_or_router = self.class.name.to_s.split("Microservice")[0].upcase.split("::")[-1]
  @scope = name.split("__")[0]
  interface_name = name.split("__")[2]
  if @interface_or_router == 'INTERFACE'
    @interface = InterfaceModel.get_model(name: interface_name, scope: @scope).build
  else
    @interface = RouterModel.get_model(name: interface_name, scope: @scope).build
  end
  @interface.name = interface_name
  # Map the interface to the interface's targets
  @interface.target_names do |target_name|
    target = System.targets[target_name]
    target.interface = @interface
  end
  if @interface.connect_on_startup
    @interface.state = 'ATTEMPTING'
  else
    @interface.state = 'DISCONNECTED'
  end
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end

  @interface_thread_sleeper = Sleeper.new
  @cancel_thread = false
  @connection_failed_messages = []
  @connection_lost_messages = []
  @mutex = Mutex.new
  if @interface_or_router == 'INTERFACE'
    @handler_thread = InterfaceCmdHandlerThread.new(@interface, self, scope: @scope)
  else
    @handler_thread = RouterTlmHandlerThread.new(@interface, self, scope: @scope)
  end
  @handler_thread.start
end

Instance Method Details

#attempting ⇒ `Object`

External method to be called by the InterfaceCmdHandlerThread to connect Thus we just set the state and allow the run method to handle the action

# File 'lib/cosmos/microservices/interface_microservice.rb', line 282

def attempting
  @interface.state = 'ATTEMPTING'
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end
end

#connect ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 456

def connect
  Logger.info "#{@interface.name}: Connecting ..."
  @interface.connect
  @interface.state = 'CONNECTED'
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end
  Logger.info "#{@interface.name}: Connection Success"
end

#disconnect(allow_reconnect = true) ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 468

def disconnect(allow_reconnect = true)
  return if @interface.state == 'DISCONNECTED' && !@interface.connected?

  # Synchronize the calls to @interface.disconnect since it takes an unknown
  # amount of time. If two calls to disconnect stack up, the if statement
  # should avoid multiple calls to disconnect.
  @mutex.synchronize do
    begin
      @interface.disconnect if @interface.connected?
    rescue => e
      Logger.error "Disconnect: #{@interface.name}: #{e.formatted}"
    end
  end

  # If the interface is set to auto_reconnect then delay so the thread
  # can come back around and allow the interface a chance to reconnect.
  if allow_reconnect and @interface.auto_reconnect and @interface.state != 'DISCONNECTED'
    attempting()
    if !@cancel_thread
      # Logger.debug "reconnect delay: #{@interface.reconnect_delay}"
      @interface_thread_sleeper.sleep(@interface.reconnect_delay)
    end
  else
    @interface.state = 'DISCONNECTED'
    if @interface_or_router == 'INTERFACE'
      InterfaceStatusModel.set(@interface.as_json, scope: @scope)
    else
      RouterStatusModel.set(@interface.as_json, scope: @scope)
    end
  end
end

#graceful_kill ⇒ `Object`



525
526
527

# File 'lib/cosmos/microservices/interface_microservice.rb', line 525

def graceful_kill
  # Just to avoid warning
end

#handle_connection_failed(connect_error) ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 410

def handle_connection_failed(connect_error)
  @error = connect_error
  Logger.error "#{@interface.name}: Connection Failed: #{connect_error.formatted(false, false)}"
  case connect_error
  when Interrupt
    Logger.info "#{@interface.name}: Closing from signal"
    @cancel_thread = true
  when Errno::ECONNREFUSED, Errno::ECONNRESET, Errno::ETIMEDOUT, Errno::ENOTSOCK, Errno::EHOSTUNREACH, IOError
    # Do not write an exception file for these extremely common cases
  else
    if RuntimeError === connect_error and (connect_error.message =~ /canceled/ or connect_error.message =~ /timeout/)
      # Do not write an exception file for these extremely common cases
    else
      Logger.error "#{@interface.name}: #{connect_error.formatted}"
      unless @connection_failed_messages.include?(connect_error.message)
        Cosmos.write_exception_file(connect_error)
        @connection_failed_messages << connect_error.message
      end
    end
  end
  disconnect() # Ensure we do a clean disconnect
end

#handle_connection_lost(err = nil, reconnect: true) ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 433

def handle_connection_lost(err = nil, reconnect: true)
  if err
    @error = err
    Logger.info "#{@interface.name}: Connection Lost: #{err.formatted(false, false)}"
    case err
    when Interrupt
      Logger.info "#{@interface.name}: Closing from signal"
      @cancel_thread = true
    when Errno::ECONNABORTED, Errno::ECONNRESET, Errno::ETIMEDOUT, Errno::EBADF, Errno::ENOTSOCK, IOError
      # Do not write an exception file for these extremely common cases
    else
      Logger.error "#{@interface.name}: #{err.formatted}"
      unless @connection_lost_messages.include?(err.message)
        Cosmos.write_exception_file(err)
        @connection_lost_messages << err.message
      end
    end
  else
    Logger.info "#{@interface.name}: Connection Lost"
  end
  disconnect(reconnect) # Ensure we do a clean disconnect
end

#handle_packet(packet) ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 355

def handle_packet(packet)
  InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  packet.received_time = Time.now.sys unless packet.received_time

  if packet.stored
    # Stored telemetry does not update the current value table
    identified_packet = System.telemetry.identify_and_define_packet(packet, @target_names)
  else
    # Identify and update packet
    if packet.identified?
      begin
        # Preidentifed packet - place it into the current value table
        identified_packet = System.telemetry.update!(packet.target_name,
                                                     packet.packet_name,
                                                     packet.buffer)
      rescue RuntimeError
        # Packet identified but we don't know about it
        # Clear packet_name and target_name and try to identify
        Logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}"
        packet.target_name = nil
        packet.packet_name = nil
        identified_packet = System.telemetry.identify!(packet.buffer,
                                                       @target_names)
      end
    else
      # Packet needs to be identified
      identified_packet = System.telemetry.identify!(packet.buffer,
                                                     @target_names)
    end
  end

  if identified_packet
    identified_packet.received_time = packet.received_time
    identified_packet.stored = packet.stored
    identified_packet.extra = packet.extra
    packet = identified_packet
  else
    unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer)
    unknown_packet.received_time = packet.received_time
    unknown_packet.stored = packet.stored
    unknown_packet.extra = packet.extra
    packet = unknown_packet
    json_hash = CvtModel.build_json_from_packet(packet)
    CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, scope: scope)
    num_bytes_to_print = [UNKNOWN_BYTES_TO_PRINT, packet.length].min
    data = packet.buffer(false)[0..(num_bytes_to_print - 1)]
    prefix = data.each_byte.map { | byte | sprintf("%02X", byte) }.join()
    Logger.warn "#{@interface.name} #{packet.target_name} packet length: #{packet.length} starting with: #{prefix}"
  end

  # Write to stream
  packet.received_count += 1
  TelemetryTopic.write_packet(packet, scope: @scope)
end

#run ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 291

def run
  begin
    if @interface.read_allowed?
      Logger.info "#{@interface.name}: Starting packet reading"
    else
      Logger.info "#{@interface.name}: Starting connection maintenance"
    end
    while true
      break if @cancel_thread

      case @interface.state
      when 'DISCONNECTED'
        begin
          # Just wait to see if we should connect later
          @interface_thread_sleeper.sleep(1)
        rescue Exception => err
          break if @cancel_thread
        end
      when 'ATTEMPTING'
        begin
          @mutex.synchronize do
            # We need to make sure connect is not called after stop() has been called
            connect() unless @cancel_thread
          end
        rescue Exception => connect_error
          handle_connection_failed(connect_error)
          break if @cancel_thread
        end
      when 'CONNECTED'
        if @interface.read_allowed?
          begin
            packet = @interface.read
            if packet
              handle_packet(packet)
              @count += 1
            else
              Logger.info "#{@interface.name}: Internal disconnect requested (returned nil)"
              handle_connection_lost()
              break if @cancel_thread
            end
          rescue Exception => err
            handle_connection_lost(err)
            break if @cancel_thread
          end
        else
          @interface_thread_sleeper.sleep(1)
          handle_connection_lost() if !@interface.connected?
        end
      end
    end
  rescue Exception => error
    Logger.error "#{@interface.name}: Packet reading thread died: #{error.formatted}"
    Cosmos.handle_fatal_exception(error)
    # Try to do clean disconnect because we're going down
    disconnect(false)
  end
  if @interface_or_router == 'INTERFACE'
    InterfaceStatusModel.set(@interface.as_json, scope: @scope)
  else
    RouterStatusModel.set(@interface.as_json, scope: @scope)
  end
  Logger.info "#{@interface.name}: Stopped packet reading"
end

#shutdown(sig = nil) ⇒ `Object`

# File 'lib/cosmos/microservices/interface_microservice.rb', line 519

def shutdown(sig = nil)
  Logger.info "#{@interface.name}: shutdown requested"
  stop()
  super()
end

#stop ⇒ `Object`

Disconnect from the interface and stop the thread

# File 'lib/cosmos/microservices/interface_microservice.rb', line 501

def stop
  Logger.info "#{@interface.name}: stop requested"
  @mutex.synchronize do
    # Need to make sure that @cancel_thread is set and the interface disconnected within
    # mutex to ensure that connect() is not called when we want to stop()
    @cancel_thread = true
    @handler_thread.stop
    @interface_thread_sleeper.cancel
    @interface.disconnect
    if @interface_or_router == 'INTERFACE'
      valid_interface = InterfaceStatusModel.get_model(name: @interface.name, scope: @scope)
    else
      valid_interface = RouterStatusModel.get_model(name: @interface.name, scope: @scope)
    end
    valid_interface.destroy if valid_interface
  end
end

Class: Cosmos::InterfaceMicroservice

Direct Known Subclasses

Constant Summary collapse

Instance Attribute Summary

Attributes inherited from Microservice

Instance Method Summary collapse

Methods inherited from Microservice

Constructor Details

#initialize(name) ⇒ InterfaceMicroservice

Instance Method Details

#attempting ⇒ Object

#connect ⇒ Object

#disconnect(allow_reconnect = true) ⇒ Object

#graceful_kill ⇒ Object

#handle_connection_failed(connect_error) ⇒ Object

#handle_connection_lost(err = nil, reconnect: true) ⇒ Object

#handle_packet(packet) ⇒ Object

#run ⇒ Object

#shutdown(sig = nil) ⇒ Object

#stop ⇒ Object

#initialize(name) ⇒ `InterfaceMicroservice`

#attempting ⇒ `Object`

#connect ⇒ `Object`

#disconnect(allow_reconnect = true) ⇒ `Object`

#graceful_kill ⇒ `Object`

#handle_connection_failed(connect_error) ⇒ `Object`

#handle_connection_lost(err = nil, reconnect: true) ⇒ `Object`

#handle_packet(packet) ⇒ `Object`

#run ⇒ `Object`

#shutdown(sig = nil) ⇒ `Object`

#stop ⇒ `Object`