lib/mogilefs/http_reader.rb

   1 # -*- encoding: binary -*-
   2 # internal implementation details here, do not rely on them in your code
   3
   4 # This class is needed because Net::HTTP streaming is still inefficient
   5 # for reading huge response bodies over fast LANs.
   6 class MogileFS::HTTPReader < MogileFS::Socket
   7   attr_accessor :content_length, :uri
   8   include MogileFS::Util
   9
  10   # backwards compat, if anybody cares
  11   alias mogilefs_size content_length # :nodoc:
  12
  13   # this may OOM your system on large files
  14   def to_s
  15     buf = ""
  16     read(@content_length, buf)
  17     return buf if buf.size == @content_length
  18
  19     raise MogileFS::SizeMismatchError,
  20           "read=#{buf.size} bytes, expected=#@content_length from #@uri", []
  21   end
  22
  23   def self.first(paths, http_method, timeout)
  24     errors = nil
  25     paths.each do |path|
  26       begin
  27         sock = new(path, http_method, timeout) and return sock
  28       rescue => e
  29         errors ||= []
  30         errors << "#{path} failed with #{e.message} (#{e.class})"
  31       end
  32     end
  33     raise MogileFS::Error,
  34           "all paths failed with #{http_method}: #{errors.join(', ')}", []
  35   end
  36
  37   # given a path, this returns a readable socket with ready data from the
  38   # body of the response.
  39   def self.new(path, http_method, timeout)
  40     uri = URI.parse(path)
  41     sock = tcp(uri.host, uri.port, timeout)
  42     buf = "#{http_method} #{uri.request_uri} HTTP/1.0\r\n\r\n" # no chunking
  43     sock.timed_write(buf, timeout)
  44
  45     sock.timed_peek(2048, buf, timeout) or
  46       raise MogileFS::InvalidResponseError, "EOF on #{http_method} #{uri}", []
  47
  48     head, _ = buf.split(/\r\n\r\n/, 2)
  49
  50     # we're dealing with a seriously slow/stupid HTTP server if we can't
  51     # get the header in a single recv(2) syscall.
  52     if head =~ %r{\AHTTP/\d+\.\d+\s+200\s*} &&
  53        head =~ %r{^Content-Length:\s*(\d+)}i
  54       sock.content_length = $1.to_i
  55       sock.uri = uri
  56
  57       case http_method
  58       when "HEAD"
  59         sock.close
  60       else # "GET"
  61         # slice off the top of the socket buffer to allow IO.copy_stream
  62         # to work
  63         sock.timed_read(head.bytesize + 4, buf, 0)
  64       end
  65       return sock
  66     end
  67     raise MogileFS::InvalidResponseError,
  68           "#{http_method} on #{uri} returned: #{head.inspect}", []
  69   rescue
  70     sock.close unless sock.closed?
  71     raise
  72   end
  73 end