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.



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

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.



454
455
456
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 454

def group
  @group
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



454
455
456
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 454

def name
  @name
end

#scope ⇒ Object (readonly)

Returns the value of attribute scope.



454
455
456
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 454

def scope
  @scope
end

#share ⇒ Object (readonly)

Returns the value of attribute share.



454
455
456
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 454

def share
  @share
end

#thread_pool ⇒ Object (readonly)

Returns the value of attribute thread_pool.



454
455
456
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 454

def thread_pool
  @thread_pool
end

#topics ⇒ Object (readonly)

Returns the value of attribute topics.



454
455
456
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 454

def topics
  @topics
end

Instance Method Details

#block_for_updates ⇒ Object 



513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 513

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 



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

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 



531
532
533
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 531

def refresh
  @read_topic = false
end

#run ⇒ Object 



487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 487

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 



535
536
537
538
539
540
541
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 535

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

#update_topics ⇒ Object 



504
505
506
507
508
509
510
511
 
# File 'lib/openc3/microservices/trigger_group_microservice.rb', line 504

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