lib/mogilefs/put/content_range.rb

   1 # -*- encoding: binary -*-
   2 # here are internal implementation details, do not rely on them in your code
   3 begin
   4   require 'net/http/persistent'
   5 rescue LoadError
   6   raise LoadError,
   7         'net-http-persistent required for :largefile => :content_range', []
   8 end
   9
  10 require 'mogilefs/new_file_common'
  11 require 'mogilefs/new_file_writer'
  12
  13 # an IO-like object
  14 class MogileFS::Put::ContentRange
  15   include MogileFS::NewFileWriter
  16   include MogileFS::NewFileCommon
  17
  18   NHP = Net::HTTP::Persistent.new('mogilefs')
  19   attr_reader :md5
  20
  21   def initialize(dests, opts) # :nodoc:
  22     @dests = dests
  23     @opts = opts
  24     @devid = @uri = @md5 = nil
  25     @bytes_uploaded = 0
  26     @errors = []
  27   end
  28
  29   def get_dest # :nodoc:
  30     return [ @devid, @uri ] if @uri
  31     rv = @dests.shift or no_nodes!
  32     rv[1] = URI.parse(rv[1])
  33     rv
  34   end
  35
  36   def no_nodes! # :nodoc:
  37     raise NoStorageNodesError,
  38           "all paths failed with PUT: #{@errors.join(', ')}", []
  39   end
  40
  41   def request_for(uri, buf) # :nodoc:
  42     put = Net::HTTP::Put.new(uri.path)
  43     put["Content-Type"] = "application/octet-stream"
  44     put["Content-MD5"] = [ Digest::MD5.digest(buf) ].pack("m").chomp!
  45     if @bytes_uploaded > 0
  46       last_byte = @bytes_uploaded + buf.bytesize - 1
  47       put["Content-Range"] = "bytes #@bytes_uploaded-#{last_byte}/*"
  48     end
  49     put.body = buf
  50
  51     put
  52   end
  53
  54   # see IO#write
  55   def write(buf)
  56     buf = String buf
  57     len = buf.bytesize
  58     return 0 if 0 == len
  59
  60     devid, uri = get_dest
  61     put = request_for(uri, buf)
  62     begin
  63       NHP.request(uri, put).value # raises on error
  64     rescue => e
  65       raise if @bytes_uploaded > 0
  66
  67       # nothing uploaded, try another dest
  68       @errors << "#{uri.to_s} - #{e.message} (#{e.class})"
  69       devid, uri = get_dest
  70       put = request_for(uri, buf)
  71       retry
  72     end
  73
  74     @uri, @devid = uri, devid if 0 == @bytes_uploaded
  75     @bytes_uploaded += len
  76     len
  77   end
  78
  79   # called on close, do not use
  80   def commit # :nodoc:
  81     zero_byte_special if @bytes_uploaded == 0
  82
  83     create_close(@devid, @uri, @bytes_uploaded)
  84   end
  85
  86   # special case for zero-byte files :<
  87   def zero_byte_special # :nodoc:
  88     @devid, @uri = get_dest
  89     put = request_for(@uri, "")
  90     begin
  91       NHP.request(@uri, put).value # raises on error
  92     rescue => e
  93       @errors << "#{@uri.to_s} - #{e.message} (#{e.class})"
  94       @devid, @uri = get_dest
  95       put = request_for(@uri, "")
  96       retry
  97     end
  98   end
  99 end if defined?(Net::HTTP::Persistent)