Class: OpenC3::InterfaceCmdHandlerThread

Inherits:

Object

• Object
• OpenC3::InterfaceCmdHandlerThread

Includes:

InterfaceDecomCommon

Defined in:

lib/openc3/microservices/interface_microservice.rb

Instance Method Summary

• #graceful_kill ⇒ Object
• #initialize(interface, tlm, logger: nil, metric: nil, scope:) ⇒ InterfaceCmdHandlerThread constructor

  A new instance of InterfaceCmdHandlerThread.

• #run ⇒ Object
• #start ⇒ Object
• #stop ⇒ Object

Methods included from InterfaceDecomCommon

#handle_build_cmd, #handle_inject_tlm

Constructor Details

#initialize(interface, tlm, logger: nil, metric: nil, scope:) ⇒ InterfaceCmdHandlerThread

Returns a new instance of InterfaceCmdHandlerThread.

# File 'lib/openc3/microservices/interface_microservice.rb', line 47

def initialize(interface, tlm, logger: nil, metric: nil, scope:)
  @interface = interface
  @tlm = tlm
  @scope = scope
  scope_model = ScopeModel.get_model(name: @scope)
  if scope_model
    @critical_commanding = scope_model.critical_commanding
  else
    @critical_commanding = 'OFF'
  end
  @logger = logger
  @logger = Logger unless @logger
  @metric = metric
  @count = 0
  @directive_count = 0
  @metric.set(name: 'interface_directive_total', value: @directive_count, type: 'counter') if @metric
  @metric.set(name: 'interface_cmd_total', value: @count, type: 'counter') if @metric
end

Instance Method Details

#graceful_kill ⇒ Object

# File 'lib/openc3/microservices/interface_microservice.rb', line 79

def graceful_kill
  InterfaceTopic.shutdown(@interface, scope: @scope)
end

#run ⇒ Object

# File 'lib/openc3/microservices/interface_microservice.rb', line 83

def run
  InterfaceTopic.receive_commands(@interface, scope: @scope) do |topic, msg_id, msg_hash, _redis|
    OpenC3.with_context(msg_hash) do
      release_critical = false
      msgid_seconds_from_epoch = msg_id.split('-')[0].to_i / 1000.0
      delta = Time.now.to_f - msgid_seconds_from_epoch
      @metric.set(name: 'interface_topic_delta_seconds', value: delta, type: 'gauge', unit: 'seconds', help: 'Delta time between data written to stream and interface cmd start') if @metric

      if topic == "OPENC3__SYSTEM__EVENTS"
        msg = JSON.parse(msg_hash['event'])
        if msg['type'] == 'scope'
          if msg['name'] == @scope
            @critical_commanding = msg['critical_commanding']
          end
        end
        next 'SUCCESS'
      end

      # Check for a raw write to the interface
      if topic =~ /CMD}INTERFACE/
        @directive_count += 1
        @metric.set(name: 'interface_directive_total', value: @directive_count, type: 'counter') if @metric
        if msg_hash['shutdown']
          @logger.info "#{@interface.name}: Shutdown requested"
          return
        end
        if msg_hash['connect']
          @logger.info "#{@interface.name}: Connect requested"
          params = []
          if msg_hash['params']
            params = JSON.parse(msg_hash['params'], :allow_nan => true, :create_additions => true)
          end
          @interface = @tlm.attempting(*params)
          next 'SUCCESS'
        end
        if msg_hash['disconnect']
          @logger.info "#{@interface.name}: Disconnect requested"
          @tlm.disconnect(false)
          next 'SUCCESS'
        end
        if msg_hash['raw']
          if @interface.connected?
            @logger.info "#{@interface.name}: Write raw"
            # A raw interface write results in an UNKNOWN packet
            command = System.commands.packet('UNKNOWN', 'UNKNOWN')
            command.received_count += 1
            command = command.clone
            command.buffer = msg_hash['raw']
            command.received_time = Time.now
            CommandTopic.write_packet(command, scope: @scope)
            @interface.write_raw(msg_hash['raw'])
            next 'SUCCESS'
          else
            next "Interface not connected: #{@interface.name}"
          end
        end
        if msg_hash.key?('log_stream')
          if msg_hash['log_stream'] == 'true'
            @logger.info "#{@interface.name}: Enable stream logging"
            @interface.start_raw_logging
          else
            @logger.info "#{@interface.name}: Disable stream logging"
            @interface.stop_raw_logging
          end
          next 'SUCCESS'
        end
        if msg_hash.key?('interface_cmd')
          params = JSON.parse(msg_hash['interface_cmd'], allow_nan: true, create_additions: true)
          begin
            @logger.info "#{@interface.name}: interface_cmd: #{params['cmd_name']} #{params['cmd_params'].join(' ')}"
            @interface.interface_cmd(params['cmd_name'], *params['cmd_params'])
            InterfaceStatusModel.set(@interface.as_json(:allow_nan => true), queued: true, scope: @scope)
          rescue => e
            @logger.error "#{@interface.name}: interface_cmd: #{e.formatted}"
            next e.message
          end
          next 'SUCCESS'
        end
        if msg_hash.key?('protocol_cmd')
          params = JSON.parse(msg_hash['protocol_cmd'], allow_nan: true, create_additions: true)
          begin
            @logger.info "#{@interface.name}: protocol_cmd: #{params['cmd_name']} #{params['cmd_params'].join(' ')} read_write: #{params['read_write']} index: #{params['index']}"
            @interface.protocol_cmd(params['cmd_name'], *params['cmd_params'], read_write: params['read_write'], index: params['index'])
            InterfaceStatusModel.set(@interface.as_json(:allow_nan => true), queued: true, scope: @scope)
          rescue => e
            @logger.error "#{@interface.name}: protocol_cmd: #{e.formatted}"
            next e.message
          end
          next 'SUCCESS'
        end
        if msg_hash.key?('inject_tlm')
          handle_inject_tlm(msg_hash['inject_tlm'])
          next 'SUCCESS'
        end
        if msg_hash.key?('release_critical')
          # Note: intentional fall through below this point
          model = CriticalCmdModel.get_model(name: msg_hash['release_critical'], scope: @scope)
          if model
            msg_hash = model.cmd_hash
            release_critical = true
          else
            next "Critical command #{msg_hash['release_critical']} not found"
          end
        end
      end

      target_name = msg_hash['target_name']
      cmd_name = msg_hash['cmd_name']
      manual = ConfigParser.handle_true_false(msg_hash['manual'])
      cmd_params = nil
      cmd_buffer = nil
      hazardous_check = nil
      if msg_hash['cmd_params']
        cmd_params = JSON.parse(msg_hash['cmd_params'], :allow_nan => true, :create_additions => true)
        range_check = ConfigParser.handle_true_false(msg_hash['range_check'])
        raw = ConfigParser.handle_true_false(msg_hash['raw'])
        hazardous_check = ConfigParser.handle_true_false(msg_hash['hazardous_check'])
      elsif msg_hash['cmd_buffer']
        cmd_buffer = msg_hash['cmd_buffer']
      end

      begin
        begin
          if cmd_params
            command = System.commands.build_cmd(target_name, cmd_name, cmd_params, range_check, raw)
          elsif cmd_buffer
            if target_name
              command = System.commands.identify(cmd_buffer, [target_name])
            else
              command = System.commands.identify(cmd_buffer, @interface.cmd_target_names)
            end
            unless command
              command = System.commands.packet('UNKNOWN', 'UNKNOWN')
              command.received_count += 1
              command = command.clone
              command.buffer = cmd_buffer
            end
          else
            raise "Invalid command received:\n #{msg_hash}"
          end
          command.received_time = Time.now
        rescue => e
          @logger.error "#{@interface.name}: #{msg_hash}"
          @logger.error "#{@interface.name}: #{e.formatted}"
          next e.message
        end

        command.extra ||= {}
        command.extra['cmd_string'] = msg_hash['cmd_string']
        command.extra['username'] = msg_hash['username']
        hazardous, hazardous_description = System.commands.cmd_pkt_hazardous?(command)

        # Initial Are you sure? Hazardous check
        # Return back the error, description, and the formatted command
        # This allows the error handler to simply re-send the command
        next "HazardousError\n#{hazardous_description}\n#{msg_hash['cmd_string']}" if hazardous_check and hazardous and not release_critical

        # Check for Critical Command
        if @critical_commanding and @critical_commanding != 'OFF' and not release_critical
          restricted = command.restricted
          if hazardous or restricted or (@critical_commanding == 'ALL' and manual)
            if hazardous
              type = 'HAZARDOUS'
            elsif restricted
              type = 'RESTRICTED'
            else
              type = 'NORMAL'
            end
            model = CriticalCmdModel.new(name: SecureRandom.uuid, type: type, interface_name: @interface.name, username: msg_hash['username'], cmd_hash: msg_hash, scope: @scope)
            model.create
            @logger.info("Critical Cmd Pending: #{msg_hash['cmd_string']}", user: msg_hash['username'], scope: @scope)
            next "CriticalCmdError\n#{model.name}"
          end
        end

        validate = ConfigParser.handle_true_false(msg_hash['validate'])
        begin
          if @interface.connected?
            result = true
            reason = nil
            if command.validator and validate
              begin
                result, reason = command.validator.pre_check(command)
              rescue => e
                result = false
                reason = e.message
              end
              # Explicitly check for false to allow nil to represent unknown
              if result == false
                message = "pre_check returned false for #{msg_hash['cmd_string']} due to #{reason}"
                raise WriteRejectError.new(message)
              end
            end

            @count += 1
            @metric.set(name: 'interface_cmd_total', value: @count, type: 'counter') if @metric

            log_message = ConfigParser.handle_true_false(msg_hash['log_message'])
            if log_message
              @logger.info(msg_hash['cmd_string'], user: msg_hash['username'], scope: @scope)
            end

            @interface.write(command)

            if command.validator and validate
              begin
                result, reason = command.validator.post_check(command)
              rescue => e
                result = false
                reason = e.message
              end
              command.extra['cmd_success'] = result
              command.extra['cmd_reason'] = reason if reason
            end

            CommandDecomTopic.write_packet(command, scope: @scope)
            CommandTopic.write_packet(command, scope: @scope)
            InterfaceStatusModel.set(@interface.as_json(:allow_nan => true), queued: true, scope: @scope)

            # Explicitly check for false to allow nil to represent unknown
            if result == false
              message = "post_check returned false for #{msg_hash['cmd_string']} due to #{reason}"
              raise WriteRejectError.new(message)
            end
            next 'SUCCESS'
          else
            next "Interface not connected: #{@interface.name}"
          end
        rescue WriteRejectError => e
          next e.message
        rescue => e
          @logger.error "#{@interface.name}: #{e.formatted}"
          next e.message
        end
      rescue => e
        @logger.error "#{@interface.name}: #{e.formatted}"
        next e.message
      end
    end
  end
end

#start ⇒ Object

# File 'lib/openc3/microservices/interface_microservice.rb', line 66

def start
  @thread = Thread.new do
    run()
  rescue Exception => e
    @logger.error "#{@interface.name}: Command handler thread died: #{e.formatted}"
    raise e
  end
end

#stop ⇒ Object

# File 'lib/openc3/microservices/interface_microservice.rb', line 75

def stop
  OpenC3.kill_thread(self, @thread)
end