Class: Rserve::Connection

Inherits:
Engine
  • Object
show all
Includes:
Protocol
Defined in:
lib/rserve/connection.rb

Overview

class providing TCP/IP connection to an Rserve

Defined Under Namespace

Classes: EvalError

Constant Summary collapse

RserveNotStartedError =

:section: Errors

Class.new(StandardError)
ServerNotAvailableError = 
Class.new(StandardError)
IncorrectServerError = 
Class.new(StandardError)
IncorrectServerVersionError = 
Class.new(StandardError)
IncorrectProtocolError = 
Class.new(StandardError)
IncorrectCredentialsError = 
Class.new(StandardError)
NotConnectedError = 
Class.new(StandardError)
AT_plain =

authorization type: plain text

0
AT_crypt =

authorization type: unix crypt

1
@@connected_object = 
nil

Constants included from Protocol

Protocol::CMD_RESP, Protocol::CMD_SPECIAL_MASK, Protocol::CMD_assignSEXP, Protocol::CMD_attachSession, Protocol::CMD_closeFile, Protocol::CMD_createFile, Protocol::CMD_ctrl, Protocol::CMD_ctrlEval, Protocol::CMD_ctrlShutdown, Protocol::CMD_ctrlSource, Protocol::CMD_detachSession, Protocol::CMD_detachedVoidEval, Protocol::CMD_eval, Protocol::CMD_login, Protocol::CMD_openFile, Protocol::CMD_readFile, Protocol::CMD_removeFile, Protocol::CMD_serAssign, Protocol::CMD_serEEval, Protocol::CMD_serEval, Protocol::CMD_setBufferSize, Protocol::CMD_setEncoding, Protocol::CMD_setSEXP, Protocol::CMD_shutdown, Protocol::CMD_voidEval, Protocol::CMD_writeFile, Protocol::DT_ARRAY, Protocol::DT_BYTESTREAM, Protocol::DT_CHAR, Protocol::DT_DOUBLE, Protocol::DT_INT, Protocol::DT_LARGE, Protocol::DT_SEXP, Protocol::DT_STRING, Protocol::ERROR_DESCRIPTIONS, Protocol::ERR_IOerror, Protocol::ERR_Rerror, Protocol::ERR_accessDenied, Protocol::ERR_auth_failed, Protocol::ERR_conn_broken, Protocol::ERR_ctrl_closed, Protocol::ERR_data_overflow, Protocol::ERR_detach_failed, Protocol::ERR_inv_cmd, Protocol::ERR_inv_par, Protocol::ERR_notOpen, Protocol::ERR_object_too_big, Protocol::ERR_out_of_mem, Protocol::ERR_session_busy, Protocol::ERR_unknownCmd, Protocol::ERR_unsupportedCmd, Protocol::MAX_LONG_SIGNED, Protocol::MAX_LONG_UNSIGNED, Protocol::RESP_ERR, Protocol::RESP_OK

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Protocol

#doubleToRawLongBits, #get_int, #get_int_original, #get_len, #get_long, #get_long_original, #longBitsToDouble, #longBitsToDouble_old, #new_hdr, #set_hdr, #set_int, #set_long

Methods inherited from Engine

#create_reference, #evaluate, #finalize_reference, #get, #get_parent_enviroment, #lock, #new_enviroment, #parse, #parse_and_eval, #resolve_reference, #supports_REPL?, #supports_enviroments?, #supports_references?, #suuports_locking?, #try_lock, #unlock

Constructor Details

#initialize(opts = Hash.new) ⇒ Connection

Make a new local connection You could provide a hash with options. Options are analog to java client: [+:auth_req+] If authentification is required (false by default) [+:transfer_charset+] Transfer charset ("UTF-8" by default) [+:auth_type+] Type of authentication (AT_plain by default) [+:hostname+] Hostname of Rserve ("127.0.0.1" by default) [+:port_number+] Port Number of Rserve (6311 by default) [+:max_tries+] Maximum number of tries before give up (5 by default) [+:cmd_init+] Command to init Rserve if not initialized ("R CMD Rserve" by default) [+:proc_rserve_ok+] Proc testing if Rserve works (uses system by default) [+:username+] Username to use (if authentication is required) [+:password+] Password to use (if authentication is required)



53
54
55
56
57
58
59
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
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/rserve/connection.rb', line 53

def initialize(opts=Hash.new)
  @auth_req         = opts.delete(:auth_req)          || false
  @transfer_charset = opts.delete(:transfer_charset)  || "UTF-8"
  @auth_type        = opts.delete(:auth_type)         || AT_plain
  @hostname         = opts.delete(:hostname)          || "127.0.0.1"
  @port_number      = opts.delete(:port_number)       || 6311
  @max_tries        = opts.delete(:max_tries)         || 5
  @cmd_init         = opts.delete(:cmd_init)          || "R CMD Rserve"
  @proc_rserve_ok   = opts.delete(:proc_rserve_ok)    || lambda { system "killall -s 0 Rserve" } 
  @username         = opts.delete(:username)          || nil
  @password         = opts.delete(:password)          || nil
  @session          = opts.delete(:session)           || nil
  @tries            = 0
  @connected=false
  if (!@session.nil?)
    @hostname=@session.host
    @port_number=@session.port
  end
  begin
    #puts "Tryin to connect..."
    connect
  rescue Errno::ECONNREFUSED
    if @tries<@max_tries
      @tries+=1
      # Rserve is available?
      if @proc_rserve_ok.call
        # Rserve is available. Incorrect host and/or portname
        raise ServerNotAvailableError, "Rserve started, but not available on #{hostname}:#{port_number}"
        # Rserve not available. We should instanciate it first
      else
        if run_server
          # Wait a moment, please
          sleep(0.25)
          retry
        else
          raise RserveNotStartedError, "Can't start Rserve"
        end
      end
      #puts "Init RServe"

    else
      raise
    end
  end
end

Instance Attribute Details

#auth_req ⇒ Object (readonly)

Returns the value of attribute auth_req.



27
28
29
 
# File 'lib/rserve/connection.rb', line 27

def auth_req
  @auth_req
end

#auth_type ⇒ Object (readonly)

Returns the value of attribute auth_type.



28
29
30
 
# File 'lib/rserve/connection.rb', line 28

def auth_type
  @auth_type
end

#connected ⇒ Object (readonly)

Returns the value of attribute connected.



26
27
28
 
# File 'lib/rserve/connection.rb', line 26

def connected
  @connected
end

#hostname ⇒ Object (readonly)

Returns the value of attribute hostname.



22
23
24
 
# File 'lib/rserve/connection.rb', line 22

def hostname
  @hostname
end

#key ⇒ Object (readonly)

Returns the value of attribute key.



29
30
31
 
# File 'lib/rserve/connection.rb', line 29

def key
  @key
end

#last_error ⇒ Object (readonly)

Returns the value of attribute last_error.



25
26
27
 
# File 'lib/rserve/connection.rb', line 25

def last_error
  @last_error
end

#persistent=(value) ⇒ Object (writeonly)

Sets the attribute persistent

Parameters:

  • value —

    the value to set the attribute persistent to.



36
37
38
 
# File 'lib/rserve/connection.rb', line 36

def persistent=(value)
  @persistent = value
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



32
33
34
 
# File 'lib/rserve/connection.rb', line 32

def port
  @port
end

#port_number ⇒ Object (readonly)

Returns the value of attribute port_number.



23
24
25
 
# File 'lib/rserve/connection.rb', line 23

def port_number
  @port_number
end

#protocol ⇒ Object (readonly)

Returns the value of attribute protocol.



24
25
26
 
# File 'lib/rserve/connection.rb', line 24

def protocol
  @protocol
end

#rsrv_version ⇒ Object (readonly)

Returns the value of attribute rsrv_version.



35
36
37
 
# File 'lib/rserve/connection.rb', line 35

def rsrv_version
  @rsrv_version
end

#rt ⇒ Object (readonly)

Returns the value of attribute rt.



30
31
32
 
# File 'lib/rserve/connection.rb', line 30

def rt
  @rt
end

#s ⇒ Object (readonly)

Returns the value of attribute s.



31
32
33
 
# File 'lib/rserve/connection.rb', line 31

def s
  @s
end

#session ⇒ Object (readonly)

Returns the value of attribute session.



33
34
35
 
# File 'lib/rserve/connection.rb', line 33

def session
  @session
end

#transfer_charset=(value) ⇒ Object (writeonly)

Sets the attribute transfer_charset

Parameters:

  • value —

    the value to set the attribute transfer_charset to.



34
35
36
 
# File 'lib/rserve/connection.rb', line 34

def transfer_charset=(value)
  @transfer_charset = value
end

Instance Method Details

#assign(sym, ct) ⇒ Object

assign a string value to a symbol in R. The symbol is created if it doesn't exist already.

Parameters:

  • sym —

    symbol name. Currently assign uses CMD_setSEXP command of Rserve, i.e. the symbol value is NOT parsed. It is the responsibility of the user to make sure that the symbol name is valid in R (recall the difference between a symbol and an expression!). In fact R will always create the symbol, but it may not be accessible (examples: "bar\nfoo" or "bar$foo").

  • ct —

    contents

Raises:



245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/rserve/connection.rb', line 245

def assign(sym, ct)
  raise NotConnectedError if !connected? or rt.nil?
  case ct
  when String
    assign_string(sym,ct)
  when REXP
    assign_rexp(sym,ct)
  else
    assign_rexp(sym, Rserve::REXP::Wrapper.wrap(ct))
  end
end

#assign_rexp(sym, rexp) ⇒ Object 



286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
 
# File 'lib/rserve/connection.rb', line 286

def assign_rexp(sym, rexp)
  r = REXPFactory.new(rexp);
  rl=r.get_binary_length();
  symn=sym.unpack("C*");
  sl=symn.length+1;
  sl=(sl&0xfffffc)+4 if ((sl&3)>0) # make sure the symbol length is divisible by 4
  rq=Array.new(sl+rl+((rl>0xfffff0) ? 12 : 8));
  symn.length.times {|i| rq[i+4]=symn[i]}
  ic=symn.length
  while(ic<sl)
    rq[ic+4]=0;
    ic+=1;
  end # pad with 0

  set_hdr(Rserve::Protocol::DT_STRING,sl,rq,0)
  set_hdr(Rserve::Protocol::DT_SEXP,rl,rq,sl+4);
  r.get_binary_representation(rq, sl+((rl>0xfffff0) ? 12 : 8));
  # puts "ASSIGN RQ: #{rq}" if $DEBUG
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_setSEXP, :cont=>rq)
  if (!rp.nil? and rp.ok?)
    rp
  else
    raise "Assign Failed"
  end
end

#assign_string(sym, ct) ⇒ Object 



257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
 
# File 'lib/rserve/connection.rb', line 257

def assign_string(sym,ct)
  symn = sym.unpack("C*")
  ctn  = ct.unpack("C*")
  sl=symn.length+1
  cl=ctn.length+1
  sl=(sl&0xfffffc)+4 if ((sl&3)>0)  # make sure the symbol length is divisible by 4
  cl=(cl&0xfffffc)+4 if ((cl&3)>0)  # make sure the content length is divisible by 4
  rq=Array.new(sl+4+cl+4)
  symn.length.times {|i| rq[i+4]=symn[i]}
  ic=symn.length
  while (ic<sl)
    rq[ic+4]=0
    ic+=1
  end
  ctn.length.times {|i| rq[i+sl+8]=ctn[i]}
  ic=ctn.length
  while (ic<cl)
    rq[ic+sl+8]=0
    ic+=1
  end
  set_hdr(Rserve::Protocol::DT_STRING,sl,rq,0)
  set_hdr(Rserve::Protocol::DT_STRING,cl,rq,sl+4)
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_setSEXP,:cont=>rq)
  if (!rp.nil? and rp.ok?)
    rp
  else
    raise "Assign Failed"
  end
end

#close ⇒ Object

Closes current connection



155
156
157
158
159
160
161
162
163
164
 
# File 'lib/rserve/connection.rb', line 155

def close
  if !@s.nil? and !@s.closed?
    @s.close_write
    @s.close_read
  end
  raise "Can't close socket" unless @s.closed?
  @connected=false
  @@connected_object=nil
  true
end

#connect ⇒ Object 



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/rserve/connection.rb', line 100

def connect
  # On windows, Rserve doesn't allows concurrent connections. 
  # So, we must close the last open connection first
  if ON_WINDOWS and !@@connected_object.nil?
    @@connected_object.close
  end

  close if @connected
  
  @s = TCPSocket::new(@hostname, @port_number)
  @rt=Rserve::Talk.new(@s)
  if @session.nil?
    #puts "Connected"
    # Accept first input
    input=@s.recv(32).unpack("a4a4a4a4a4a4a4a4")      
    raise IncorrectServerError, "Handshake failed: Rsrv signature expected, but received [#{input[0]}]" unless input[0]=="Rsrv"
    @rsrv_version=input[1].to_i
    raise IncorrectServerVersionError, "Handshake failed: The server uses more recent protocol than this client." if @rsrv_version>103
    @protocol=input[2]
    raise IncorrectProtocolError, "Handshake failed: unsupported transfer protocol #{@protocol}, I talk only QAP1." if @protocol!="QAP1"
    (3..7).each do |i|
      attr=input[i]
      if (attr=="ARpt") 
        @auth_req=true
      elsif (attr=="ARuc") 
        @auth_req=true
        @auth_type=AT_crypt
      elsif (attr[0..0]=='K') 
        @key=attr[1,2]
      end
    end
     if auth_req
  else # we have a session to take care of
    @s.write(@session.key.pack("C*"))
    @rsrv_version=session.rsrv_version
  end
  @connected=true
  @@connected_object=self
  @last_error="OK"
end

#connected? ⇒ Boolean

Check connection state. Note that currently this state is not checked on-the-spot, that is if connection went down by an outside event this is not reflected by the flag. return true if this connection is alive

Returns:

  • (Boolean) 


142
143
144
 
# File 'lib/rserve/connection.rb', line 142

def connected?
  @connected
end

#detach ⇒ Object

Detaches the session and closes the connection (requires Rserve 0.4+). The session can be only resumed by calling RSession.attach

Raises:



327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/rserve/connection.rb', line 327

def detach
  raise NotConnectedError if !connected? or rt.nil?
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_detachSession)
  if !rp.nil? and rp.ok? 
    s=Rserve::Session.new(self,rp)
    close
    s
  else
    raise "Cannot detach"
  end
end

#eval(cmd) ⇒ Object

evaluates the given command and retrieves the result

  • @param cmd command/expression string
  • @return R-xpression or null if an error occured */

Raises:



205
206
207
208
209
210
211
212
213
 
# File 'lib/rserve/connection.rb', line 205

def eval(cmd)
  raise NotConnectedError if !connected? or rt.nil?
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_eval, :cont=>cmd+"\n")
  if !rp.nil? and rp.ok?
    parse_eval_response(rp)
  else
    raise EvalError.new(rp), "eval failed: #{rp.to_s}"
  end
end

#get_server_version ⇒ Object

Get server version as reported during the handshake.



166
167
168
 
# File 'lib/rserve/connection.rb', line 166

def get_server_version
  @rsrv_version
end

#login ⇒ Object

This server requires a login. Send the required credentials to the server.

Raises:



147
148
149
150
151
152
 
# File 'lib/rserve/connection.rb', line 147

def 
  raise IncorrectCredentialsError, "Need username and password to connect" if @username.nil? || @password.nil?
  @password = @password.crypt(key) if key && auth_type == AT_crypt
  rp = @rt.request({:cmd => Rserve::Protocol::CMD_login, :cont => "#{@username}\n#{@password}"})
  raise IncorrectCredentialsError, "Server did not accept credentials" if rp.error?
end

#parse_eval_response(rp) ⇒ Object

NOT TESTED



216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/rserve/connection.rb', line 216

def parse_eval_response(rp)
  rxo=0
  pc=rp.cont
  if (rsrv_version>100) # /* since 0101 eval responds correctly by using DT_SEXP type/len header which is 4 bytes long */
    rxo=4
    # we should check parameter type (should be DT_SEXP) and fail if it's not
    if pc.nil?
      raise "Error while processing eval output: SEXP (type #{Rserve::Protocol::DT_SEXP}) expected but nil returned"
    elsif (pc[0]!=Rserve::Protocol::DT_SEXP and pc[0]!=(Rserve::Protocol::DT_SEXP|Rserve::Protocol::DT_LARGE))
      raise "Error while processing eval output: SEXP (type #{Rserve::Protocol::DT_SEXP}) expected but found result type "+pc[0].to_s+"."
    end
    
    if (pc[0]==(Rserve::Protocol::DT_SEXP|Rserve::Protocol::DT_LARGE))
      rxo=8; # large data need skip of 8 bytes
    end
    # warning: we are not checking or using the length - we assume that only the one SEXP is returned. This is true for the current CMD_eval implementation, but may not be in the future. */
  end
  if pc.length>rxo
    rx=REXPFactory.new;
    rx.parse_REXP(pc, rxo);
    return rx.get_REXP();
  else
    return nil
  end
end

#shutdown ⇒ Object

Shutdown remote Rserve. Note that some Rserves cannot be shut down from the client side

Raises:



315
316
317
318
319
320
321
322
323
 
# File 'lib/rserve/connection.rb', line 315

def shutdown
  raise NotConnectedError if !connected? or rt.nil?
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_shutdown)
  if !rp.nil? and rp.ok? 
    true
  else
    raise "Shutdown failed"
  end
end

#void_eval(cmd) ⇒ Object

evaluates the given command, but does not fetch the result (useful for assignment operations)

  • @param cmd command/expression string */

Raises:



172
173
174
175
176
177
178
179
180
181
 
# File 'lib/rserve/connection.rb', line 172

def void_eval(cmd)
  raise NotConnectedError if !connected? or rt.nil?
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_voidEval, :cont=>cmd+"\n")
  if !rp.nil? and rp.ok?
    true
  else
    raise EvalError.new(rp), "voidEval failed: #{rp.to_s}"
  end

end

#void_eval_detach(cmd) ⇒ Object

  • @param cmd command/expression string.
  • @return session object that can be use to attach back to the session once the command completed

Raises:



188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/rserve/connection.rb', line 188

def void_eval_detach(cmd)
  raise NotConnectedError if !connected? or rt.nil?
  rp=rt.request(:cmd=>Rserve::Protocol::CMD_detachedVoidEval,:cont=>cmd+"\n")
  if rp.nil? or !rp.ok?
    raise EvalError.new(rp), "detached void eval failed : #{rp.to_s}"
  else
    s=Rserve::Session.new(self,rp)
    close
    s
  end
end