Class: RubyZmqFramework::Flow

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_zmq_framework/flow.rb

Overview

Parses a flow manifest (flow.yml) — the graph as data, Node-RED style. The manifest is the ONLY place that knows the topology; node processes learn their wiring from the environment variables computed here, and a node's code never mentions another node.

nodes:
ecu:
  cmd: ruby nodes/ecu.rb
  publishes: [engine_data]
  subscribes: [throttle_request]
  env: { SOME_KNOB: "42" }   # optional, merged into the process env

Defined Under Namespace

Classes: Node

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(spec) ⇒ Flow

Returns a new instance of Flow.



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/ruby_zmq_framework/flow.rb', line 24

def initialize(spec)
  unless spec.is_a?(Hash) && spec['nodes'].is_a?(Hash)
    raise Error, '[Framework Error] Flow manifest needs a top-level "nodes" map'
  end

  @nodes = spec['nodes'].map do |name, cfg|
    cfg ||= {}
    raise Error, "[Framework Error] Flow node #{name} needs a cmd" unless cfg['cmd']

    Node.new(
      name: name.to_s,
      cmd: cfg['cmd'],
      publishes: Array(cfg['publishes']).map(&:to_s),
      subscribes: Array(cfg['subscribes']).map(&:to_s),
      env: (cfg['env'] || {}).transform_values(&:to_s)
    )
  end
  warn_about_deaf_subscriptions
end

Instance Attribute Details

#nodes ⇒ Object (readonly)

Returns the value of attribute nodes.



18
19
20
 
# File 'lib/ruby_zmq_framework/flow.rb', line 18

def nodes
  @nodes
end

Class Method Details

.load(path) ⇒ Object 



20
21
22
 
# File 'lib/ruby_zmq_framework/flow.rb', line 20

def self.load(path)
  new(YAML.safe_load(File.read(path)))
end

Instance Method Details

#wiring(ports) ⇒ Object

The environment for every node process, given a => port map. This is the whole trick that keeps nodes blackboxes: each node's peer list is computed from who publishes the topics it subscribes to.



47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/ruby_zmq_framework/flow.rb', line 47

def wiring(ports)
  @nodes.to_h do |node|
    peers = peer_names(node).map { |name| "127.0.0.1:#{ports.fetch(name)}" }
    [node.name, {
      'BUS_PORT' => ports.fetch(node.name).to_s,
      'BUS_PEERS' => peers.join(','),
      'BUS_SUBSCRIBES' => node.subscribes.join(','),
      'NODE_NAME' => node.name
    }.merge(node.env)]
  end
end