Class: Pigato::Client

Inherits:
Base
  • Object
show all
Defined in:
lib/pigato/client.rb

Constant Summary collapse

@@mtx = 
Mutex.new
@@ctxs = 
{}
@@sockets = 
{}

Instance Method Summary collapse

Methods inherited from Base

#get_iid, #get_mtx, #get_proc_id, #get_socket, #get_thread_id, #init, #sock_close, #sock_create

Constructor Details

#initialize(broker, conf = {}) ⇒ Client

Returns a new instance of Client.



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
 
# File 'lib/pigato/client.rb', line 8

def initialize broker, conf = {}
  @broker = broker

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

  @conf.merge!(conf)

  init
  
  if @conf[:autostart]
    start
  end
end

Instance Method Details

#_recv(rid) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/pigato/client.rb', line 60

def _recv rid 
  iid = get_iid
  socket = @@sockets[iid]
  socket.rcvtimeo = 2500
  
  data = []
  
  msg = socket.recv_message()
  while 1 do
    break if msg.nil? || msg.size == 0
    data << msg.pop.data
  end

  if data[3] != rid  
    data = []
    if @conf[:logger]
      @conf[:logger].error("PigatoClient: RID mismatch #{data[3]}/#{rid}")
    end
  end

  return nil if data.length == 0

  data.shift
  return data 
end

#request(service, request, opts = {}) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/pigato/client.rb', line 34

def request service, request, opts = {}
  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))}

  res = send msg
  return nil if res.nil?

  rtimer = Time.now + (@conf[:timeout] * 0.001)

  res = [] 
  while Time.now <= rtimer do
    chunk = _recv rid
    next 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

#send(msg) ⇒ Object 



25
26
27
28
29
30
31
32
 
# File 'lib/pigato/client.rb', line 25

def send msg
  iid = get_iid
  start if @@sockets[iid] == nil && @conf[:autostart]
  socket = get_socket
  return nil if socket.nil?
  socket.send_message msg
  true
end

#start ⇒ Object 



86
87
88
89
90
91
92
 
# File 'lib/pigato/client.rb', line 86

def start
  stop
  sock_create
  super 
rescue ZMQ::Error => e
  puts e
end

#stop ⇒ Object 



94
95
96
97
 
# File 'lib/pigato/client.rb', line 94

def stop
  sock_close
  super 
end