Merge pull request #8 from senid231/use-own-socket-wrapper

use own socket wrapper

Author: senid231

jrpc.gemspec

@@ -18,7 +18,6 @@ Gem::Specification.new do |spec|
   spec.require_paths = ['lib']
 
   spec.add_dependency 'netstring', '~> 0'
-  spec.add_dependency 'net_tcp_client', '~> 2.0'
   spec.add_dependency 'oj', '~> 2.0'
 
   spec.add_development_dependency 'bundler', '~> 1.10'

lib/jrpc.rb

@@ -1,6 +1,8 @@
 require 'jrpc/version'
 require 'jrpc/utils'
 require 'jrpc/base_client'
+require 'jrpc/transport/socket_base'
+require 'jrpc/transport/socket_tcp'
 require 'jrpc/tcp_client'
 require 'jrpc/error/error'
 require 'jrpc/error/connection_error'

lib/jrpc/base_client.rb

@@ -9,6 +9,13 @@ class BaseClient
     ID_CHARACTERS = (('a'..'z').to_a + ('0'..'9').to_a + ('A'..'Z').to_a).freeze
     REQUEST_TYPES = [:request, :notification].freeze
 
+    def self.connect(uri, options)
+      client = new(uri, options)
+      yield(client)
+    ensure
+      client.close if client
+    end
+
     def initialize(uri, options)
       @uri = uri
       @options = options

lib/jrpc/tcp_client.rb

@@ -1,4 +1,3 @@
-require 'net/tcp_client'
 require 'netstring'
 require 'logger'
 require 'benchmark'
@@ -18,22 +17,34 @@ def initialize(uri, options = {})
       timeout = @options.fetch(:timeout, 5)
       connect_timeout = @options.fetch(:connect_timeout, timeout)
       read_timeout = @options.fetch(:read_timeout, timeout)
-      write_timeout = @options.fetch(:write_timeout, nil) # default 60
-      connect_retry_count = @options.fetch(:connect_retry_count, nil) # default 10
+      write_timeout = @options.fetch(:write_timeout, 60) # default 60
+      connect_retry_count = @options.fetch(:connect_retry_count, 10) # default 10
+      @close_after_sent = @options.fetch(:close_after_sent, false)
 
-      @transport = Net::TCPClient.new server: @uri,
-                                   connect_retry_count: connect_retry_count,
-                                   connect_timeout: connect_timeout,
-                                   read_timeout: read_timeout,
-                                   write_timeout: write_timeout,
-                                   buffered: false # recommended for RPC
-    rescue ::SocketError
-      raise ConnectionError, "Can't connect to #{@uri}"
+      @transport = JRPC::Transport::SocketTcp.new server: @uri,
+                                                  connect_retry_count: connect_retry_count,
+                                                  connect_timeout: connect_timeout,
+                                                  read_timeout: read_timeout,
+                                                  write_timeout: write_timeout
+      begin
+        @transport.connect
+      rescue JRPC::SocketTcp::Error
+        raise ConnectionError, "Can't connect to #{@uri}"
+      end
     end
 
     private
 
-    def send_command(request, options={})
+    def ensure_connected
+      if @transport.closed?
+        logger.debug { 'Connecting transport...' }
+        @transport.connect
+        logger.debug { 'Connected.' }
+      end
+    end
+
+    def send_command(request, options = {})
+      ensure_connected
       read_timeout = options.fetch(:read_timeout)
       write_timeout = options.fetch(:write_timeout)
       response = nil
@@ -49,15 +60,20 @@ def send_command(request, options={})
         "(#{'%.2f' % (t * 1000)}ms) Response message: #{Utils.truncate(response, MAX_LOGGED_MESSAGE_LENGTH)}"
       end
       response
+    ensure
+      @transport.close if @close_after_sent
     end
 
-    def send_notification(request, options={})
+    def send_notification(request, options = {})
+      ensure_connected
       write_timeout = options.fetch(:write_timeout)
       logger.debug { "Request address: #{uri}" }
       logger.debug { "Request message: #{Utils.truncate(request, MAX_LOGGED_MESSAGE_LENGTH)}" }
       logger.debug { "Request write_timeout: #{write_timeout}" }
       send_request(request, write_timeout)
       logger.debug { 'No response required' }
+    ensure
+      @transport.close if @close_after_sent
     end
 
     def create_message(method, params)
@@ -74,8 +90,7 @@ def send_request(request, timeout)
     def receive_response(timeout)
       timeout ||= @transport.read_timeout
       length = get_msg_length(timeout)
-      response = ''
-      @transport.read(length+1, response, timeout)
+      response = @transport.read(length + 1, timeout)
       raise ClientError.new('invalid response. missed comma as terminator') if response[-1] != ','
       response.chomp(',')
     rescue ::SocketError
@@ -85,7 +100,7 @@ def receive_response(timeout)
     def get_msg_length(timeout)
       length = ''
       while true do
-        character = @transport.read(1, nil, timeout)
+        character = @transport.read(1, timeout)
         break if character == ':'
         length += character
       end

lib/jrpc/transport/socket_base.rb

@@ -0,0 +1,82 @@
+module JRPC
+  module Transport
+    class SocketBase
+
+      class Error < StandardError
+      end
+
+      class TimeoutError < Error
+        def initialize
+          super(self.class.to_s.split('::').last)
+        end
+      end
+
+      class ReadTimeoutError < TimeoutError
+      end
+
+      class WriteTimeoutError < TimeoutError
+      end
+
+      class ConnectionTimeoutError < TimeoutError
+      end
+
+      class ConnectionFailedError < Error
+      end
+
+      attr_reader :options, :read_timeout, :write_timeout
+
+      def self.connect(options)
+        connection = new(options)
+        yield(connection)
+      ensure
+        connection.close if connection
+      end
+
+      def initialize(options)
+        @server = options.fetch(:server)
+        @read_timeout = options.fetch(:read_timeout, nil)
+        @write_timeout = options.fetch(:write_timeout, nil)
+        @connect_timeout = options.fetch(:connect_timeout, nil)
+        @connect_retry_count = options.fetch(:connect_retry_count, 0)
+        @options = options
+      end
+
+      def connect
+        retries = @connect_retry_count
+
+        while retries >= 0
+          begin
+            connect_socket
+            break
+          rescue Error => e
+            retries -= 1
+            raise e if retries < 0
+          end
+        end
+      end
+
+      def read(_length, _timeout = @read_timeout)
+        raise NotImplementedError
+      end
+
+      def write(_data, _timeout = @write_timeout)
+        raise NotImplementedError
+      end
+
+      def close
+        raise NotImplementedError
+      end
+
+      def closed?
+        raise NotImplementedError
+      end
+
+      private
+
+      def connect_socket
+        raise NotImplementedError
+      end
+
+    end
+  end
+end

lib/jrpc/transport/socket_tcp.rb

@@ -0,0 +1,78 @@
+module JRPC
+  module Transport
+    class SocketTcp < SocketBase
+
+      attr_reader :socket
+
+      def initialize(options)
+        super
+        @socket = build_socket
+      end
+
+      def read(length, timeout = @read_timeout)
+        received = ''
+        length_to_read = length
+        while length_to_read > 0
+          io_read, = IO.select([@socket], [], [], timeout)
+          raise ReadTimeoutError unless io_read
+          chunk = io_read[0].read_nonblock(length_to_read)
+          received += chunk
+          length_to_read -= chunk.bytesize
+        end
+        received
+      end
+
+      def write(data, timeout = @write_timeout)
+        length_written = 0
+        data_to_write = data
+        while data_to_write.bytesize > 0
+          _, io_write, = IO.select([], [@socket], [], timeout)
+          raise WriteTimeoutError unless io_write
+          chunk_length = io_write[0].write_nonblock(data_to_write)
+          length_written += chunk_length
+          data_to_write = data.byteslice(length_written, data.length)
+        end
+        length_written
+      end
+
+      def close
+        @socket.close
+      end
+
+      def closed?
+        @socket.closed?
+      end
+
+      private
+
+      def set_timeout_to(socket, type, value)
+        secs = Integer(value)
+        u_secs = Integer((value - secs) * 1_000_000)
+        opt_val = [secs, u_secs].pack('l_2')
+        socket.setsockopt Socket::SOL_SOCKET, type, opt_val
+      end
+
+      def build_socket
+        host = @server.split(':').first
+        addr = Socket.getaddrinfo(host, nil)
+        sock = Socket.new(Socket.const_get(addr[0][0]), Socket::SOCK_STREAM, 0)
+        set_timeout_to(sock, Socket::SO_RCVTIMEO, @connect_timeout) if @connect_timeout
+        sock
+      end
+
+      def connect_socket
+        host, port = @server.split(':')
+        addr = Socket.getaddrinfo(host, nil)
+        full_addr = Socket.pack_sockaddr_in(port, addr[0][3])
+        @socket.connect(full_addr)
+      rescue Errno::EISCONN => _
+        # already connected
+      rescue Errno::ETIMEDOUT => _
+        raise ConnectionTimeoutError
+      rescue Errno::ECONNREFUSED, Errno::EHOSTUNREACH, Errno::ETIMEDOUT => e
+        raise ConnectionFailedError, "#{e.class} #{e.message}"
+      end
+
+    end
+  end
+end

lib/jrpc/version.rb

@@ -1,3 +1,3 @@
 module JRPC
-  VERSION = '1.0.1'
+  VERSION = '1.1.0'
 end

spec/fake_transport.rb

@@ -12,9 +12,9 @@ def write(_, write_timeout = nil)
     @write_timeout = write_timeout if write_timeout
   end
 
-  def read(_, buffer, read_timeout = nil)
+  def read(_, read_timeout = nil)
     @read_timeout = read_timeout if read_timeout
-    buffer << @response
+    @response
   end
 
   def response=(str)

spec/tcp_client_spec.rb

@@ -50,11 +50,11 @@
 
       json_result = expected_result.to_json
       socket_stub.response = json_result
-      expect(socket_stub).to receive(:read).with(1, nil, 30).exactly(json_result.size.to_s.size).times.
+      expect(socket_stub).to receive(:read).with(1, 30).exactly(json_result.size.to_s.size).times.
           and_return(
               *(json_result.size.to_s.split('') + [':'])
           )
-      expect(socket_stub).to receive(:read).with(json_result.size + 1, '', 30).and_return(json_result + ',').and_call_original
+      expect(socket_stub).to receive(:read).with(json_result.size + 1, 30).and_return(json_result + ',').and_call_original
 
       expect(subject).to eq JSON.parse(json_result)['result']
     end
@@ -113,7 +113,7 @@
     let(:invoke_request_params) { [1, 2] }
 
     before do
-      allow(Net::TCPClient).to receive(:new).with(any_args).once.and_return(socket_stub)
+      allow(JRPC::Transport::SocketTcp).to receive(:new).with(any_args).once.and_return(socket_stub)
     end
 
     it 'calls perform_request("sum", params: [1, 2])' do
@@ -149,7 +149,7 @@
     let(:invoke_notification_params) { [1, 2] }
 
     before do
-      allow(Net::TCPClient).to receive(:new).with(any_args).once.and_return(socket_stub)
+      allow(JRPC::Transport::SocketTcp).to receive(:new).with(any_args).once.and_return(socket_stub)
     end
 
     it 'calls perform_request("sum", params: [1, 2], type: :notification)' do
@@ -187,7 +187,8 @@
 
     before do
       allow_any_instance_of(JRPC::TcpClient).to receive(:generate_id).with(no_args).and_return(stubbed_generated_id)
-      allow(Net::TCPClient).to receive(:new).with(any_args).once.and_return(socket_stub)
+      allow(JRPC::Transport::SocketTcp).to receive(:new).with(any_args).once.and_return(socket_stub)
+      allow(socket_stub).to receive(:closed?).and_return(false)
     end
 
     context 'with array params' do