Contents

class Pigato::Client

  def initialize broker, conf = {}
    @broker = broker
    @context = ZMQ::Context.new(1)
    @socket = nil

    @conf = {
      :autostart => false,
      :timeout => 2500
    }

    @conf.merge!(conf)

    if @conf[:autostart]
      start
    end
  end

  def request service, request, opts = {} 
    return nil if @socket == nil;

    request = [Oj.dump(request), Oj.dump(opts)]

    rid = SecureRandom.uuid
    request = [Pigato::C_CLIENT, Pigato::W_REQUEST, service, rid].concat(request)
    msg = ZMQ::Message.new
    request.reverse.each{|p| msg.push(ZMQ::Frame(p))}
    @socket.send_message msg

    res = [] 
    while 1 do
      chunk = _recv rid
      break if chunk == nil
      res << Oj.load(chunk[4])
      break if chunk[0] == Pigato::W_REPLY
    end

    return nil if res.length == 0
    return res[0] if res.length == 1
    res
  end

  def _recv rid 
    @socket.rcvtimeo = @conf[:timeout]
    data = []
    d1 = Time.now
    msg = @socket.recv_message()
    while 1 do
      break if !msg || msg.size == 0
      data << msg.pop.data
    end
    data = [] if data[3] != rid

    return nil if data.length == 0

    data.shift
    return data 
  end
  
  def start
    reconnect_to_broker
  end

  def stop
    if @socket
      @socket.close
    end
  end

  def reconnect_to_broker
    stop
    @socket = @context.socket ZMQ::DEALER
    @context.linger = 0
    @socket.identity = SecureRandom.uuid
    @socket.connect @broker
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
pigato-0.2.9	lib/pigato/client.rb
pigato-0.2.8	lib/pigato/client.rb