lib/mogilefs/httpfile.rb

   1 # -*- encoding: binary -*-
   2 # here are internal implementation details, do not use them in your code
   3 require 'stringio'
   4 require 'uri'
   5 require 'mogilefs/backend'
   6 require 'mogilefs/util'
   7
   8 ##
   9 # HTTPFile wraps up the new file operations for storing files onto an HTTP
  10 # storage node.
  11 #
  12 # You really don't want to create an HTTPFile by hand.  Instead you want to
  13 # create a new file using MogileFS::MogileFS.new_file.
  14 #
  15 class MogileFS::HTTPFile < StringIO
  16   include MogileFS::Util
  17
  18   class EmptyResponseError < MogileFS::Error; end
  19   class BadResponseError < MogileFS::Error; end
  20   class UnparseableResponseError < MogileFS::Error; end
  21   class NoStorageNodesError < MogileFS::Error
  22     def message; 'Unable to open socket to storage node'; end
  23   end
  24
  25   ##
  26   # The URI this file will be stored to.
  27
  28   attr_reader :uri
  29
  30   attr_reader :devid
  31
  32   ##
  33   # The big_io name in case we have file > 256M
  34
  35   attr_accessor :big_io
  36
  37   attr_accessor :streaming_io
  38
  39   ##
  40   # Creates a new HTTPFile with MogileFS-specific data.  Use
  41   # MogileFS::MogileFS#new_file instead of this method.
  42
  43   def initialize(dests, content_length)
  44     super ""
  45     @streaming_io = @big_io = @uri = @devid = nil
  46     @dests = dests
  47     @tried = {}
  48   end
  49
  50   ##
  51   # Writes an HTTP PUT request to +sock+ to upload the file and
  52   # returns file size if the socket finished writing
  53   def upload(devid, uri) # :nodoc:
  54     file_size = length
  55     sock = MogileFS::Socket.tcp(uri.host, uri.port)
  56
  57     if @streaming_io
  58       file_size = @streaming_io.length
  59       sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
  60                  "Content-Length: #{file_size}\r\n\r\n")
  61       @streaming_io.call(Proc.new do |data_to_write|
  62         sock.write(data_to_write)
  63       end)
  64     elsif @big_io
  65       # Don't try to run out of memory
  66       File.open(@big_io, "rb") do |fp|
  67         file_size = fp.stat.size
  68         sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
  69                    "Content-Length: #{file_size}\r\n\r\n")
  70         copy_stream(fp, sock)
  71       end
  72     else
  73       sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
  74                  "Content-Length: #{length}\r\n\r\n#{string}")
  75     end
  76
  77     case line = sock.timed_read(23, "")
  78     when %r{^HTTP/\d\.\d\s+(2\d\d)\s} # success!
  79       file_size
  80     when nil
  81       raise EmptyResponseError, 'Unable to read response line from server'
  82     when %r{^HTTP/\d\.\d\s+(\d+)}
  83       raise BadResponseError, "HTTP response status from upload: #$1"
  84     else
  85       raise UnparseableResponseError, "Response line not understood: #{line}"
  86     end
  87     ensure
  88       sock.close if sock && ! sock.closed?
  89   end
  90
  91   def commit
  92     errors = nil
  93     @dests.each do |devid, path|
  94       begin
  95         uri = URI.parse(path)
  96         bytes_uploaded = upload(devid, uri)
  97         @devid, @uri = devid, uri
  98         return bytes_uploaded
  99       rescue => e
 100         errors ||= []
 101         errors << "#{path} failed with #{e.message} (#{e.class})"
 102       end
 103     end
 104
 105     raise NoStorageNodesError,
 106           "all paths failed with PUT: #{errors.join(', ')}", []
 107   end
 108 end