Class: OpenC3::TriggerGroupManager

Inherits:
Object
  • Object
show all
Defined in:
lib/openc3/microservices/trigger_group_microservice.rb

Overview

The trigger manager starts a thread pool and subscribes to the telemtry decom topic add the packet to a queue. TriggerGroupManager adds the “packet” to the thread pool queue and the thread will evaluate the “trigger”.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name:, logger:, scope:, group:, share:) ⇒ TriggerGroupManager

Returns a new instance of TriggerGroupManager.



442
443
444
445
446
447
448
449
450
451
452
453
454
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 442

def initialize(name:, logger:, scope:, group:, share:)
  @name = name
  @logger = logger
  @scope = scope
  @group = group
  @share = share
  @worker_count = 3
  @queue = Queue.new
  @read_topic = true
  @topics = []
  @thread_pool = nil
  @cancel_thread = false
end

Instance Attribute Details

#group ⇒ Object (readonly)

Returns the value of attribute group.



440
441
442
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 440

def group
  @group
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



440
441
442
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 440

def name
  @name
end

#scope ⇒ Object (readonly)

Returns the value of attribute scope.



440
441
442
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 440

def scope
  @scope
end

#share ⇒ Object (readonly)

Returns the value of attribute share.



440
441
442
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 440

def share
  @share
end

#thread_pool ⇒ Object (readonly)

Returns the value of attribute thread_pool.



440
441
442
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 440

def thread_pool
  @thread_pool
end

#topics ⇒ Object (readonly)

Returns the value of attribute topics.



440
441
442
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 440

def topics
  @topics
end

Instance Method Details

#block_for_updates ⇒ Object 



499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 499

def block_for_updates
  @read_topic = true
  while @read_topic
    begin
      Topic.read_topics(@topics) do |topic, _msg_id, msg_hash, _redis|
        @logger.debug "TriggerGroupManager block_for_updates: #{topic} #{msg_hash.to_s}"
        if topic != @share.trigger_base.autonomic_topic
          packet = JSON.parse(msg_hash['json_data'], :allow_nan => true, :create_additions => true)
          @share.packet_base.add(topic: topic, packet: packet)
        end
        @queue << "#{topic}"
      end
    rescue StandardError => e
      @logger.error "TriggerGroupManager failed to read topics #{@topics}\n#{e.formatted}"
    end
  end
end

#generate_thread_pool ⇒ Object 



456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 456

def generate_thread_pool()
  thread_pool = []
  @worker_count.times do | i |
    worker = TriggerGroupWorker.new(
      name: @name,
      logger: @logger,
      scope: @scope,
      group: @group,
      queue: @queue,
      share: @share,
      ident: i,
    )
    thread_pool << Thread.new { worker.run }
  end
  return thread_pool
end

#refresh ⇒ Object 



517
518
519
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 517

def refresh
  @read_topic = false
end

#run ⇒ Object 



473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 473

def run
  @logger.info "TriggerGroupManager running"
  @thread_pool = generate_thread_pool()
  loop do
    begin
      update_topics()
    rescue StandardError => e
      @logger.error "TriggerGroupManager failed to update topics.\n#{e.formatted}"
    end
    break if @cancel_thread

    block_for_updates()
    break if @cancel_thread
  end
  @logger.info "TriggerGroupManager exiting"
end

#shutdown ⇒ Object 



521
522
523
524
525
526
527
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 521

def shutdown
  @read_topic = false
  @cancel_thread = true
  @worker_count.times do | i |
    @queue << nil
  end
end

#update_topics ⇒ Object 



490
491
492
493
494
495
496
497
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 490

def update_topics
  past_topics = @topics
  @topics = @share.trigger_base.topics()
  @logger.debug "TriggerGroupManager past_topics: #{past_topics} topics: #{@topics}"
  (past_topics - @topics).each do | removed_topic |
    @share.packet_base.remove(topic: removed_topic)
  end
end