lib/mogilefs/socket_common.rb

   1 # -*- encoding: binary -*-
   2 # internal implementation details here, do not rely on this in your code
   3 require "socket"
   4
   5 module MogileFS::SocketCommon
   6   attr_reader :mogilefs_addr
   7   attr_accessor :mogilefs_size
   8
   9   def post_init(host, port)
  10     @mogilefs_addr = "#{host}:#{port}"
  11     Socket.const_defined?(:TCP_NODELAY) and
  12       setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
  13     self
  14   end
  15
  16   def unreadable_socket!
  17     raise MogileFS::UnreadableSocketError,
  18           "#@mogilefs_addr never became readable"
  19   end
  20
  21   def request_truncated!(written, expect)
  22     raise MogileFS::RequestTruncatedError,
  23           "request truncated (sent #{written} expected #{expect})"
  24   end
  25
  26   SEP_RE = /\A(.*?#{Regexp.escape("\n")})/
  27   def timed_gets(timeout = 5)
  28     unless defined?(@rbuf)
  29       @rbuf = timed_read(1024, "", timeout) or return # EOF
  30     end
  31     begin
  32       @rbuf.sub!(SEP_RE, "") and return $1
  33       tmp ||= ""
  34       if timed_read(1024, tmp, timeout)
  35         @rbuf << tmp
  36       else
  37         # EOF, return the last buffered bit even without SEP_RE matching
  38         # (not ideal for MogileFS, this is an error)
  39         return @rbuf.empty? ? nil : @rbuf.slice!(0, @rbuf.size)
  40       end
  41     end while true
  42   end
  43
  44   def read(size, buf = "", timeout = 5)
  45     timed_read(size, buf, timeout) or return # nil/EOF
  46
  47     while size > buf.bytesize
  48       tmp ||= ""
  49       timed_read(size - buf.bytesize, tmp, timeout) or return buf # truncated
  50       buf << tmp
  51     end
  52
  53     buf # full read
  54   end
  55
  56   def readpartial(size, buf = "", timeout = 5)
  57     timed_read(size, buf, timeout) or raise EOFError, "end of file reached"
  58   end
  59 end