Class: RJR::Nodes::Unix

Inherits:
RJR::Node show all
Defined in:
lib/rjr/nodes/unix.rb

Overview

Unix node definition, listen for and invoke json-rpc requests via Unix Sockets

Clients should specify the socketname when listening for requests and when invoking them.

TODO client / server examples

Constant Summary collapse

RJR_NODE_TYPE = 
:unix
PERSISTENT_NODE = 
true

Instance Attribute Summary collapse

Attributes inherited from RJR::Node

#dispatcher, #message_headers, #node_id

Instance Method Summary collapse

Methods inherited from RJR::Node

#clear_event_handlers, em, #halt, #join, #node_type, #on, #persistent?, persistent?, tp

Constructor Details

#initialize(args = {}) ⇒ Unix

Unix initializer

Options Hash (args):

  • :socketname (String) —

    the name of the socket which to listen on



95
96
97
98
99
100
101
 
# File 'lib/rjr/nodes/unix.rb', line 95

def initialize(args = {})
   super(args)
   @socketname = args[:socketname]

   @connections = []
   @connections_lock = Mutex.new
end

Instance Attribute Details

#connections ⇒ Object

Returns the value of attribute connections.



69
70
71
 
# File 'lib/rjr/nodes/unix.rb', line 69

def connections
  @connections
end

Instance Method Details

#invoke(socketname, rpc_method, *args) ⇒ Object

Instructs node to send rpc request, and wait for / return response.

Implementation of RJR::Node#invoke

Do not invoke directly from em event loop or callback as will block the message subscription used to receive responses



135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/rjr/nodes/unix.rb', line 135

def invoke(socketname, rpc_method, *args)
  message = RequestMessage.new :method => rpc_method,
                               :args   => args,
                               :headers => @message_headers
  connection = nil
  @@em.schedule {
    init_client(:socketname => socketname,
                :rjr_node => self) { |c|
      connection = c
      c.send_msg message.to_s
    }
  }

  # TODO optional timeout for response ?
  result = wait_for_result(message)

  if result.size > 2
    raise Exception, result[2]
  end
  return result[1]
end

#listen ⇒ Object

Instruct Node to start listening for and dispatching rpc requests

Implementation of RJR::Node#listen



117
118
119
120
121
122
 
# File 'lib/rjr/nodes/unix.rb', line 117

def listen
  @@em.schedule {
    @@em.start_unix_domain_server @socketname, nil, UnixConnection, { :rjr_node => self }
  }
  self
end

#notify(socketname, rpc_method, *args) ⇒ Object

Instructs node to send rpc notification (immadiately returns / no response is generated)

Implementation of RJR::Node#notify

destination node is listening on



165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/rjr/nodes/unix.rb', line 165

def notify(socketname, rpc_method, *args)
  # will block until message is published
  published_l = Mutex.new
  published_c = ConditionVariable.new

  invoked = false
  conn    = nil
  message = NotificationMessage.new :method => rpc_method,
                                    :args   => args,
                                    :headers => @message_headers
  @@em.schedule {
    init_client(:socketname => socketname,
                :rjr_node => self) { |c|
      conn = c
      c.send_msg message.to_s
      # XXX, this should be invoked only when we are sure event
      # machine sent message. Shouldn't pose a problem unless event
      # machine is killed immediately after
      published_l.synchronize { invoked = true ; published_c.signal }
    }
  }
  published_l.synchronize { published_c.wait published_l unless invoked }
  #sleep 0.01 until conn.get_outbound_data_size == 0
  nil
end

#send_msg(data, connection) ⇒ Object

Send data using specified connection

Implementation of RJR::Node#send_msg



110
111
112
 
# File 'lib/rjr/nodes/unix.rb', line 110

def send_msg(data, connection)
  connection.send_msg(data)
end

#to_s ⇒ Object 



103
104
105
 
# File 'lib/rjr/nodes/unix.rb', line 103

def to_s
  "RJR::Nodes::Unix<#{@node_id},#{@socketname}>"
end