ceph/src/jaegertracing/thrift/lib/rb/lib/thrift/transport/socket.rb

   1 # encoding: ascii-8bit
   2 #
   3 # Licensed to the Apache Software Foundation (ASF) under one
   4 # or more contributor license agreements. See the NOTICE file
   5 # distributed with this work for additional information
   6 # regarding copyright ownership. The ASF licenses this file
   7 # to you under the Apache License, Version 2.0 (the
   8 # "License"); you may not use this file except in compliance
   9 # with the License. You may obtain a copy of the License at
  10 #
  11 #   http://www.apache.org/licenses/LICENSE-2.0
  12 #
  13 # Unless required by applicable law or agreed to in writing,
  14 # software distributed under the License is distributed on an
  15 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  16 # KIND, either express or implied. See the License for the
  17 # specific language governing permissions and limitations
  18 # under the License.
  19 #
  20
  21 require 'socket'
  22
  23 module Thrift
  24   class Socket < BaseTransport
  25     def initialize(host='localhost', port=9090, timeout=nil)
  26       @host = host
  27       @port = port
  28       @timeout = timeout
  29       @desc = "#{host}:#{port}"
  30       @handle = nil
  31     end
  32
  33     attr_accessor :handle, :timeout
  34
  35     def open
  36       for addrinfo in ::Socket::getaddrinfo(@host, @port, nil, ::Socket::SOCK_STREAM) do
  37         begin
  38           socket = ::Socket.new(addrinfo[4], ::Socket::SOCK_STREAM, 0)
  39           socket.setsockopt(::Socket::IPPROTO_TCP, ::Socket::TCP_NODELAY, 1)
  40           sockaddr = ::Socket.sockaddr_in(addrinfo[1], addrinfo[3])
  41           begin
  42             socket.connect_nonblock(sockaddr)
  43           rescue Errno::EINPROGRESS
  44             unless IO.select(nil, [ socket ], nil, @timeout)
  45               next
  46             end
  47             begin
  48               socket.connect_nonblock(sockaddr)
  49             rescue Errno::EISCONN
  50             end
  51           end
  52           return @handle = socket
  53         rescue StandardError => e
  54           next
  55         end
  56       end
  57       raise TransportException.new(TransportException::NOT_OPEN, "Could not connect to #{@desc}: #{e}")
  58     end
  59
  60     def open?
  61       !@handle.nil? and !@handle.closed?
  62     end
  63
  64     def write(str)
  65       raise IOError, "closed stream" unless open?
  66       str = Bytes.force_binary_encoding(str)
  67       begin
  68         if @timeout.nil? or @timeout == 0
  69           @handle.write(str)
  70         else
  71           len = 0
  72           start = Time.now
  73           while Time.now - start < @timeout
  74             rd, wr, = IO.select(nil, [@handle], nil, @timeout)
  75             if wr and not wr.empty?
  76               len += @handle.write_nonblock(str[len..-1])
  77               break if len >= str.length
  78             end
  79           end
  80           if len < str.length
  81             raise TransportException.new(TransportException::TIMED_OUT, "Socket: Timed out writing #{str.length} bytes to #{@desc}")
  82           else
  83             len
  84           end
  85         end
  86       rescue TransportException => e
  87         # pass this on
  88         raise e
  89       rescue StandardError => e
  90         @handle.close
  91         @handle = nil
  92         raise TransportException.new(TransportException::NOT_OPEN, e.message)
  93       end
  94     end
  95
  96     def read(sz)
  97       raise IOError, "closed stream" unless open?
  98
  99       begin
 100         if @timeout.nil? or @timeout == 0
 101           data = @handle.readpartial(sz)
 102         else
 103           # it's possible to interrupt select for something other than the timeout
 104           # so we need to ensure we've waited long enough, but not too long
 105           start = Time.now
 106           timespent = 0
 107           rd = loop do
 108             rd, = IO.select([@handle], nil, nil, @timeout - timespent)
 109             timespent = Time.now - start
 110             break rd if (rd and not rd.empty?) or timespent >= @timeout
 111           end
 112           if rd.nil? or rd.empty?
 113             raise TransportException.new(TransportException::TIMED_OUT, "Socket: Timed out reading #{sz} bytes from #{@desc}")
 114           else
 115             data = @handle.readpartial(sz)
 116           end
 117         end
 118       rescue TransportException => e
 119         # don't let this get caught by the StandardError handler
 120         raise e
 121       rescue StandardError => e
 122         @handle.close unless @handle.closed?
 123         @handle = nil
 124         raise TransportException.new(TransportException::NOT_OPEN, e.message)
 125       end
 126       if (data.nil? or data.length == 0)
 127         raise TransportException.new(TransportException::UNKNOWN, "Socket: Could not read #{sz} bytes from #{@desc}")
 128       end
 129       data
 130     end
 131
 132     def close
 133       @handle.close unless @handle.nil? or @handle.closed?
 134       @handle = nil
 135     end
 136
 137     alias to_io handle
 138
 139     def to_s
 140       "socket(#{@host}:#{@port})"
 141     end
 142   end
 143 end