Class: RJR::Nodes::TCP

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

Overview

TCP node definition, listen for and invoke json-rpc requests via TCP sockets

Clients should specify the hostname / port when listening for requests and when invoking them.

Examples:

Listening for json-rpc requests over tcp

# initialize node
server = RJR::Nodes::TCP.new :node_id => 'server', :host => 'localhost', :port => '7777'

# register rjr dispatchers (see RJR::Dispatcher)
server.dispatcher.handle('hello') { |name|
  "Hello #{name}!"
}

# listen and block
server.listen
server.join

Invoking json-rpc requests over tcp

client = RJR::Nodes::TCP.new :node_id => 'client', :host => 'localhost', :port => '8888'
puts client.invoke('jsonrpc://localhost:7777', 'hello', 'mo')

Constant Summary collapse

RJR_NODE_TYPE = 
:tcp
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 = {}) ⇒ TCP

TCP initializer

Parameters:

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

    the options to create the tcp node with

Options Hash (args):

  • :host (String) —

    the hostname/ip which to listen on

  • :port (Integer) —

    the port which to listen on



111
112
113
114
115
116
117
118
 
# File 'lib/rjr/nodes/tcp.rb', line 111

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

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

Instance Attribute Details

#connections ⇒ Object

Returns the value of attribute connections.



84
85
86
 
# File 'lib/rjr/nodes/tcp.rb', line 84

def connections
  @connections
end

Instance Method Details

#invoke(uri, 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:

  • uri (String) —

    location of node to send request to, should be in format of jsonrpc://hostname:port or tcp://hostname:port

  • rpc_method (String) —

    json-rpc method to invoke on destination

  • args (Array) —

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



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/rjr/nodes/tcp.rb', line 152

def invoke(uri, rpc_method, *args)
  uri = URI.parse(uri)
  host,port = uri.host, uri.port

  message = RequestMessage.new :method => rpc_method,
                               :args   => args,
                               :headers => @message_headers
  connection = nil
  @@em.schedule {
    init_client(:host => host, :port => port,
                :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



134
135
136
137
138
139
 
# File 'lib/rjr/nodes/tcp.rb', line 134

def listen
  @@em.schedule {
    @@em.start_server @host, @port, TCPConnection, { :rjr_node => self }
  }
  self
end

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

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

Implementation of RJR::Node#notify

Parameters:

  • uri (String) —

    location of node to send notification to, should be in format of jsonrpc://hostname:port

  • rpc_method (String) —

    json-rpc method to invoke on destination

  • args (Array) —

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



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/rjr/nodes/tcp.rb', line 185

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

  uri = URI.parse(uri)
  host,port = uri.host, uri.port

  invoked = false
  conn    = nil
  message = NotificationMessage.new :method => rpc_method,
                                    :args   => args,
                                    :headers => @message_headers
  @@em.schedule {
    init_client(:host => host, :port => port,
                :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



127
128
129
 
# File 'lib/rjr/nodes/tcp.rb', line 127

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

#to_s ⇒ Object 



120
121
122
 
# File 'lib/rjr/nodes/tcp.rb', line 120

def to_s
  "RJR::Nodes::TCP<#{@node_id},#{@host},#{@port}>"
end