class AgentZMQ::RepAgent

Attributes

reply[RW]

Public Class Methods

new(name) click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 10
def initialize name
  @name=name
  @socket_opts=[]

  @read_thread=nil
  @mutex = Mutex.new
  @do_run_read_thread = true
end

Public Instance Methods

do_read() click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 19
def do_read 
  while true 
    @mutex.synchronize do
      return unless @do_run_read_thread
    end
    @zmq_poller.poll(1000)

    @zmq_poller.readables.each do |sock|
      request=AgentZMQ::Helpers.read_msg sock
      add_msg request

      unless @reply.nil?
        AgentZMQ::Helpers.publish(sock, @reply.call(request))
      end
    end
  end

  zmq_socket.close 
end
reset() click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 62
def reset
  clear
end
sock_type() click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 39
def sock_type
  zmq_context.socket(ZMQ::REP)
end
start() click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 50
def start
  start_read_thread
end
start_read_thread() click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 43
def start_read_thread
  @zmq_poller = ZMQ::Poller.new
  @zmq_poller.register(zmq_socket, ZMQ::POLLIN)

  @read_thread = Thread.new {do_read}
end
stop() click to toggle source
# File lib/agent_zmq/agents/rep_agent.rb, line 54
def stop
  @mutex.synchronize do
    @do_run_read_thread=false
  end
  @read_thread.join
  zmq_socket.close
end