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