Class: RubyZmqFramework::ZeroMQBus

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

Constant Summary collapse

POLL_TIMEOUT_MS =

How long the listener blocks in poll before looping, in milliseconds.

100

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(my_port, peer_ports = [], bind_host: '127.0.0.1') ⇒ ZeroMQBus

my_port may be 0 to bind an OS-assigned ephemeral port (read it back via #port). peer_ports entries may be Integers (loopback ports), "host:port" strings, or full ZeroMQ endpoints ("tcp://10.0.0.5:5555"). Pass bind_host: "0.0.0.0" to accept peers from other machines.



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/ruby_zmq_framework/zeromq_bus.rb', line 17

def initialize(my_port, peer_ports = [], bind_host: '127.0.0.1')
  @context = ZMQ::Context.new

  # Publisher Socket
  @pub = @context.socket(ZMQ::PUB)
  bind_endpoint = "tcp://#{bind_host}:#{my_port.to_i.zero? ? '*' : my_port}"
  check! @pub.bind(bind_endpoint), "bind to #{bind_endpoint}"
  @port = bound_port

  # Subscriber Socket
  @sub = @context.socket(ZMQ::SUB)
  peer_ports.each do |peer|
    endpoint = peer_endpoint(peer)
    check! @sub.connect(endpoint), "connect to #{endpoint}"
  end
  check! @sub.setsockopt(ZMQ::SUBSCRIBE, ''), 'subscribe'

  # Keyed by topic *string*, with no default proc: the listener looks up
  # every topic that arrives off the wire, and a defaulting hash would
  # insert a permanent empty entry per unknown topic — an unbounded,
  # network-driven leak. A Monitor guards it because subscribe runs on
  # the caller's thread while dispatch runs on the listener thread.
  @local_subscribers = {}
  @subscribers_lock = Monitor.new
  @running = true
  start_listener
end

Instance Attribute Details

#port ⇒ Object (readonly)

The port actually bound — differs from my_port when my_port is 0.



11
12
13
 
# File 'lib/ruby_zmq_framework/zeromq_bus.rb', line 11

def port
  @port
end

Instance Method Details

#close ⇒ Object

Stops the listener thread and releases both sockets and the context. Stop anything still publishing on this bus first (heartbeats, reader threads): publishing on a closed bus raises RubyZmqFramework::Error. Idempotent.



49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/ruby_zmq_framework/zeromq_bus.rb', line 49

def close
  return if @closed

  @closed = true
  @running = false
  @listener.join(POLL_TIMEOUT_MS / 1000.0 + 1) unless Thread.current == @listener

  [@sub, @pub].each do |sock|
    sock.setsockopt(ZMQ::LINGER, 0)
    sock.close
  end
  @context.terminate
  nil
end

#publish(topic, payload = {}) ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/ruby_zmq_framework/zeromq_bus.rb', line 70

def publish(topic, payload = {})
  json = payload.to_json
  check! @pub.send_strings([topic.to_s, json]), "publish #{topic}"

  # A SUB socket never connects back to its own PUB, so without this,
  # two modules sharing one bus in the same process could not hear each
  # other. Delivering the serialized form keeps the payload
  # representation identical whether a message arrived locally or over
  # the wire (string keys become symbols either way).
  dispatch(topic.to_s, json)
end

#subscribe(topic, module_instance) ⇒ Object 



64
65
66
67
68
 
# File 'lib/ruby_zmq_framework/zeromq_bus.rb', line 64

def subscribe(topic, module_instance)
  @subscribers_lock.synchronize do
    (@local_subscribers[topic.to_s] ||= []) << module_instance
  end
end