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