Contents

require 'thread'

class Pigato::Client

  def initialize broker, conf = {}
    @mtx = Mutex.new
    @broker = broker
    @ctx = ZMQ::Context.new
    @ctx.linger = 0

    @sockets = {}

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

    @conf.merge!(conf)

    if @conf[:autostart]
      start
    end
  end

 
  def request service, request, opts = {}
    start if @sockets[Thread.current.object_id] == nil
    socket = @sockets[Thread.current.object_id]
     
    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 = @sockets[Thread.current.object_id]
    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
    socket = @sockets[Thread.current.object_id]
    if socket
      socket.close
      @sockets.delete(Thread.current.object_id)
    end
  end

  def reconnect_to_broker
    stop
    socket = @ctx.socket ZMQ::DEALER
    socket.identity = SecureRandom.uuid
    socket.connect @broker
    @sockets[Thread.current.object_id] = socket
  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
pigato-0.2.12	lib/pigato/client.rb