ZeroMQ experiment. See http://steppedpyramids.com/2013/11/27/ch-zeromq-experiments/

duplex_msg.rb

require 'ffi-rzmq'

inport, outport, label = ARGV[0], ARGV[1], ARGV[2]

context = ZMQ::Context.new(1)

simulation = Thread.new do
  shoutsock = context.socket(ZMQ::PAIR)
  shoutsock.bind("inproc://shouter")

  listsock = context.socket(ZMQ::PAIR)
  listsock.bind("inproc://listener")

  uisock = context.socket(ZMQ::PAIR)
  uisock.connect("inproc://ui")

  poller = ZMQ::Poller.new
  poller.register(uisock, ZMQ::POLLIN)
  poller.register(listsock, ZMQ::POLLIN)

  ticks = 0
  lastmsg = "nothing"
  loop do
    poller.poll(1000 / 60)
    poller.readables.each do |sock|
      case sock
      when uisock
        sock.recv_string(msg = '') 
        shoutsock.send_string("#{label}: #{ticks}: #{lastmsg}? #{msg}!")
      when listsock
        sock.recv_string(msg = '') 
        lastmsg = msg
      end
    end
    ticks += 1
    sleep(1.0 / 60.0)
  end
end

listener = Thread.new do
  subscriber = context.socket(ZMQ::SUB)
  subscriber.connect("tcp://localhost:#{inport}")
  subscriber.setsockopt(ZMQ::SUBSCRIBE, "")

  simsock = context.socket(ZMQ::PAIR)
  simsock.connect("inproc://listener")

  loop do
    subscriber.recv_string(msg = '')
    puts "HEARING >> #{msg}"
    if msg =~ /\? (.+)\!/
      simsock.send_string($1)
    end
  end
end

shouter = Thread.new do
  publisher = context.socket(ZMQ::PUB)
  publisher.bind("tcp://*:#{outport}")

  simsock = context.socket(ZMQ::PAIR)
  simsock.connect("inproc://shouter")

  loop do
    msg = ''
    simsock.recv_string(msg)
    publisher.send_string(msg)
  end
end

[simulation, listener, shouter].each do |which|
  which.abort_on_exception = true  
end

inputsock = context.socket(ZMQ::PAIR)
inputsock.bind("inproc://ui")

while msg = STDIN.gets
  inputsock.send_string(msg.chomp) 
end