Class: Legion::Gaia::Router::AgentBridge

Inherits:
Object
  • Object
show all
Includes:
Logging::Helper
Defined in:
lib/legion/gaia/router/agent_bridge.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(worker_id:) ⇒ AgentBridge

Returns a new instance of AgentBridge.



13
14
15
16
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 13

def initialize(worker_id:)
  @worker_id = worker_id
  @started = false
end

Instance Attribute Details

#started ⇒ Object (readonly)

Returns the value of attribute started.



11
12
13
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 11

def started
  @started
end

#worker_id ⇒ Object (readonly)

Returns the value of attribute worker_id.



11
12
13
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 11

def worker_id
  @worker_id
end

Instance Method Details

#ingest_from_payload(payload) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 50

def ingest_from_payload(payload)
  frame = reconstruct_input_frame(payload)
  return { ingested: false, reason: :invalid_frame } unless frame

  if Legion::Gaia.respond_to?(:ingest) && Legion::Gaia.started?
    log.debug("AgentBridge ingesting frame_id=#{frame.id} worker_id=#{@worker_id}")
    Legion::Gaia.ingest(frame)
  else
    log.error("AgentBridge ingest failed frame_id=#{frame.id} worker_id=#{@worker_id} error=gaia_not_started")
    { ingested: false, reason: :gaia_not_started }
  end
end

#publish_output(output_frame) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 34

def publish_output(output_frame)
  return { published: false, reason: :not_started } unless started?

  if transport_available?
    Transport::Messages::OutputFrameMessage.new(frame: output_frame).publish
    log.info("AgentBridge published output frame_id=#{output_frame.id} worker_id=#{@worker_id}")
    { published: true, frame_id: output_frame.id }
  else
    log.error(
      'AgentBridge publish failed ' \
      "frame_id=#{output_frame.id} worker_id=#{@worker_id} error=no_transport"
    )
    { published: false, reason: :no_transport, frame_id: output_frame.id }
  end
end

#start ⇒ Object 



18
19
20
21
22
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 18

def start
  @started = true
  subscribe_inbound if transport_available?
  log.info("AgentBridge started worker_id=#{@worker_id}")
end

#started? ⇒ Boolean

Returns:

  • (Boolean) 


30
31
32
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 30

def started?
  @started == true
end

#stop ⇒ Object 



24
25
26
27
28
 
# File 'lib/legion/gaia/router/agent_bridge.rb', line 24

def stop
  @consumer&.cancel if @consumer.respond_to?(:cancel)
  @started = false
  log.info("AgentBridge stopped worker_id=#{@worker_id}")
end