Class: Pigato::Base

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

Direct Known Subclasses

Client, Worker

Constant Summary collapse

@@sockets = 
{}
@@sockets_ids = 
{}
@@mtxs = 
{}
@@mtx = 
Mutex.new
@@global_heartbeat_at = 
Time.now
@@global_thread = 
nil

Instance Method Summary collapse

Instance Method Details

#get_iid ⇒ Object 



24
25
26
27
 
# File 'lib/pigato/base.rb', line 24

def get_iid 
  iid = get_thread_id + '#' + @iid
  iid
end

#get_mtx ⇒ Object 



34
35
36
37
38
39
40
41
42
 
# File 'lib/pigato/base.rb', line 34

def get_mtx
  tid = get_thread_id

  if @@mtxs[tid].nil?
    @@mtxs[tid] = Mutex.new
  end 

  return @@mtxs[tid]
end

#get_proc_id ⇒ Object 



19
20
21
22
 
# File 'lib/pigato/base.rb', line 19

def get_proc_id
  pid = "#" + Process.pid.to_s
  pid
end

#get_socket ⇒ Object 



29
30
31
32
 
# File 'lib/pigato/base.rb', line 29

def get_socket
  socket = @@sockets[get_iid]
  socket
end

#get_thread_id ⇒ Object 



14
15
16
17
 
# File 'lib/pigato/base.rb', line 14

def get_thread_id
  tid = get_proc_id() + "#" + Thread.current.object_id.to_s
  tid
end

#init ⇒ Object 



10
11
12
 
# File 'lib/pigato/base.rb', line 10

def init 
  @iid = SecureRandom.uuid
end

#sock_close ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/pigato/base.rb', line 68

def sock_close
  @@mtx.synchronize {
    pid = get_proc_id()

    iid = get_iid

    socket = @@sockets[iid]
    if socket
      begin
        socket.close
      rescue
      end
      @@sockets.delete(iid)
      @@sockets_ids.delete(iid)
    end
  }
end

#sock_create ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/pigato/base.rb', line 44

def sock_create
  @@mtx.synchronize {
    pid = get_proc_id()
 
    ctx = ZMQ::context
    if ctx == nil then
      ctx = ZMQ::Context.new
      ctx.linger = 0
    end

    socket = ctx.socket ZMQ::DEALER
    sid = SecureRandom.uuid
    socket.identity = sid 
    socket.connect @broker

    if !@conf[:timeout].nil? then
      socket.rcvtimeo = @conf[:timeout]
    end

    @@sockets[get_iid] = socket
    @@sockets_ids[get_iid] = sid 
  }
end

#start ⇒ Object 



86
87
88
 
# File 'lib/pigato/base.rb', line 86

def start
  @active = 1
end

#stop ⇒ Object 



90
91
92
 
# File 'lib/pigato/base.rb', line 90

def stop
  @active = 0
end