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

Instance Attribute Summary collapse

Attributes inherited from RJR::Node

#dispatcher, #message_headers, #node_id

Instance Method Summary collapse

Methods inherited from RJR::Node

em, #halt, #join, #node_type, #on, tp

Constructor Details

#initialize(args = {}) ⇒ Unix

Unix initializer

Parameters:

  • args (Hash) (defaults to: {}) —

    the options to create the unix node with

Options Hash (args):

  • :socketname (String) —

    the name of the socket which to listen on



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

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.



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

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

Parameters:

  • socketname (String) —

    name of socket which destination node is listening on

  • rpc_method (String) —

    json-rpc method to invoke on destination

  • args (Array) —

    array of arguments to convert to json and invoke remote method wtih



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

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



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

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

Parameters:

  • socketname (String) —

    name of socket which

  • rpc_method (String) —

    json-rpc method to invoke on destination

  • args (Array) —

    array of arguments to convert to json and invoke remote method wtih



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

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



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

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

#to_s ⇒ Object 



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

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