Class: OpenC3::RouterTlmHandlerThread

Inherits:

Object

• Object
• OpenC3::RouterTlmHandlerThread

Defined in:

lib/openc3/microservices/interface_microservice.rb

Instance Method Summary

• #graceful_kill ⇒ Object
• #initialize(router, tlm, logger: nil, metric: nil, db_shard: 0, scope:) ⇒ RouterTlmHandlerThread constructor

  A new instance of RouterTlmHandlerThread.

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

Constructor Details

#initialize(router, tlm, logger: nil, metric: nil, db_shard: 0, scope:) ⇒ RouterTlmHandlerThread

Returns a new instance of RouterTlmHandlerThread.

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

def initialize(router, tlm, logger: nil, metric: nil, db_shard: 0, scope:)
  @router = router
  @tlm = tlm
  @scope = scope
  @db_shard = db_shard.to_i
  @logger = logger
  @logger = Logger unless @logger
  @metric = metric
  @count = 0
  @directive_count = 0
  @metric.set(name: 'router_directive_total', value: @directive_count, type: 'counter') if @metric
  @metric.set(name: 'router_tlm_total', value: @count, type: 'counter') if @metric
end

Instance Method Details

#graceful_kill ⇒ Object

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

def graceful_kill
  RouterTopic.shutdown(@router, scope: @scope)
  sleep(0.001) # Allow other threads to run
end

#run ⇒ Object

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

def run
  RouterTopic.receive_telemetry(@router, scope: @scope, db_shard: @db_shard) do |topic, msg_id, msg_hash, _redis|
    msgid_seconds_from_epoch = msg_id.split('-')[0].to_i / 1000.0
    delta = Time.now.to_f - msgid_seconds_from_epoch
    @metric.set(name: 'router_topic_delta_seconds', value: delta, type: 'gauge', unit: 'seconds', help: 'Delta time between data written to stream and router tlm start') if @metric

    # Check for commands to the router itself
    if /CMD}ROUTER/.match?(topic)
      @directive_count += 1
      @metric.set(name: 'router_directive_total', value: @directive_count, type: 'counter') if @metric

      if msg_hash['shutdown']
        @logger.info "#{@router.name}: Shutdown requested"
        RouterTopic.clear_topics(RouterTopic.topics(@router, scope: @scope), db_shard: @db_shard)
        return
      end
      if msg_hash['connect']
        @logger.info "#{@router.name}: Connect requested"
        params = []
        if msg_hash['params']
          params = JSON.parse(msg_hash['params'], allow_nan: true, create_additions: true)
        end
        @router = @tlm.attempting(*params)
        next 'SUCCESS'
      end
      if msg_hash['disconnect']
        @logger.info "#{@router.name}: Disconnect requested"
        @tlm.disconnect(false)
        next 'SUCCESS'
      end
      if msg_hash.key?('log_stream')
        if msg_hash['log_stream'] == 'true'
          @logger.info "#{@router.name}: Enable stream logging"
          @router.start_raw_logging
        else
          @logger.info "#{@router.name}: Disable stream logging"
          @router.stop_raw_logging
        end
        next 'SUCCESS'
      end
      if msg_hash.key?('router_cmd')
        params = JSON.parse(msg_hash['router_cmd'], allow_nan: true, create_additions: true)
        begin
          @logger.info "#{@router.name}: router_cmd: #{params['cmd_name']} #{params['cmd_params'].join(' ')}"
          @router.interface_cmd(params['cmd_name'], *params['cmd_params'])
          RouterStatusModel.set(@router.as_json(), queued: true, scope: @scope)
        rescue => e
          @logger.error "#{@router.name}: router_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 "#{@router.name}: protocol_cmd: #{params['cmd_name']} #{params['cmd_params'].join(' ')} read_write: #{params['read_write']} index: #{params['index']}"
          @router.protocol_cmd(params['cmd_name'], *params['cmd_params'], read_write: params['read_write'], index: params['index'])
          RouterStatusModel.set(@router.as_json(), queued: true, scope: @scope)
        rescue => e
          @logger.error "#{@router.name}: protocol_cmd: #{e.formatted}"
          next e.message
        end
        next 'SUCCESS'
      end
      if msg_hash.key?('target_control')
        begin
          params = JSON.parse(msg_hash['target_control'], allow_nan: true, create_additions: true)
          target_name = params['target_name']
          cmd_only = params['cmd_only']
          tlm_only = params['tlm_only']
          action = params['action']
          if action == 'disable'
            @router.cmd_target_enabled[target_name] = false unless tlm_only
            @router.tlm_target_enabled[target_name] = false unless cmd_only
            @logger.info "#{@router.name}: target_disable: #{target_name} cmd_only:#{cmd_only} tlm_only:#{tlm_only}"
          else # enable
            @router.cmd_target_enabled[target_name] = true unless tlm_only
            @router.tlm_target_enabled[target_name] = true unless cmd_only
            @logger.info "#{@router.name}: target_enable: #{target_name} cmd_only:#{cmd_only} tlm_only:#{tlm_only}"
          end
        rescue => e
          @logger.error "#{@router.name}: target_control: #{e.formatted}"
          next e.message
        end
        next 'SUCCESS'
      end
      if msg_hash.key?('router_details')
        begin
          next @router.details.as_json.to_json(allow_nan: true)
        rescue => e
          @logger.error "#{@router.name}: router_details: #{e.formatted}"
          next e.message
        end
      end
      next 'SUCCESS'
    end

    if @router.connected?
      @count += 1
      @metric.set(name: 'router_tlm_total', value: @count, type: 'counter') if @metric

      target_name = msg_hash["target_name"]
      packet_name = msg_hash["packet_name"]

      if @router.tlm_target_enabled[target_name]
        packet = System.telemetry.packet(target_name, packet_name)
        packet.stored = ConfigParser.handle_true_false(msg_hash["stored"])
        packet.received_time = Time.from_nsec_from_epoch(msg_hash["time"].to_i)
        packet.received_count = msg_hash["received_count"].to_i
        packet.buffer = msg_hash["buffer"]

        begin
          @router.write(packet)
          RouterStatusModel.set(@router.as_json(), queued: true, scope: @scope)
          next 'SUCCESS'
        rescue => e
          @logger.error "#{@router.name}: #{e.formatted}"
          next e.message
        end
      end
    else
      next nil # Don't ack disabled targets
    end
  end
end

#start ⇒ Object

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

def start
  @thread = Thread.new do
    run()
  rescue Exception => e
    @logger.error "#{@router.name}: Telemetry handler thread died: #{e.formatted}"
    raise e
  end
  ThreadManager.instance.register(@thread, stop_object: self)
end

#stop ⇒ Object

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

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